Repository: paritytech/substrate
File: client/network/src/service/out_events.rs (Rust)

Code before:
	/// Creates a new empty collection of senders.
	pub fn new(registry: Option<&Registry>) -> Result<Self, PrometheusError> {
		let metrics =
			if let Some(registry) = registry { Some(Metrics::register(registry)?) } else { None };

		Ok(Self { event_streams: Vec::new(), metrics: Arc::new(metrics) })
	}

	/// Adds a new [`Sender`] to the collection.
	pub fn push(&mut self, sender: Sender) {
		let mut metrics = sender.metrics.lock();
		debug_assert!(metrics.is_none());
		*metrics = Some(self.metrics.clone());
		drop(metrics);

		if let Some(metrics) = &*self.metrics {
			metrics.num_channels.with_label_values(&[sender.name]).inc();
		}

		self.event_streams.push(sender);
	}

	/// Sends an event.
	pub fn send(&mut self, event: Event) {
		self.event_streams.retain_mut(|sender| {
			if sender.inner.len() == sender.queue_size_warning && !sender.warning_fired {
				sender.warning_fired = true;
				sender.creation_backtrace.resolve();
				error!(
					"The number of unprocessed events in channel `{}` exceeded {}.\n\
					 The channel was created at:\n{:?}",
					sender.name, sender.queue_size_warning, sender.creation_backtrace,
				);
			}
			sender.inner.try_send(event.clone()).is_ok()
		});

		if let Some(metrics) = &*self.metrics {
			for ev in &self.event_streams {
				metrics.event_in(&event, ev.name);
			}
		}
	}
}

impl fmt::Debug for OutChannels {
	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
		f.debug_struct("OutChannels")
			.field("num_channels", &self.event_streams.len())
			.finish()
	}

Review comment (suggestion):
```suggestion
			if sender.inner.len() >= sender.queue_size_warning && !sender.warning_fired {
```

If this is the only sender for this channel then it shouldn't make a difference, but just in case.

Code after:
	/// Creates a new empty collection of senders.
	pub fn new(registry: Option<&Registry>) -> Result<Self, PrometheusError> {
		let metrics =
			if let Some(registry) = registry { Some(Metrics::register(registry)?) } else { None };

		Ok(Self { event_streams: Vec::new(), metrics: Arc::new(metrics) })
	}

	/// Adds a new [`Sender`] to the collection.
	pub fn push(&mut self, sender: Sender) {
		let mut metrics = sender.metrics.lock();
		debug_assert!(metrics.is_none());
		*metrics = Some(self.metrics.clone());
		drop(metrics);

		if let Some(metrics) = &*self.metrics {
			metrics.num_channels.with_label_values(&[sender.name]).inc();
		}

		self.event_streams.push(sender);
	}

	/// Sends an event.
	pub fn send(&mut self, event: Event) {
		self.event_streams.retain_mut(|sender| {
			if sender.inner.len() >= sender.queue_size_warning && !sender.warning_fired {
				sender.warning_fired = true;
				error!(
					"The number of unprocessed events in channel `{}` exceeded {}.\n\
					 The channel was created at:\n{:}\n
					 The last event was sent from:\n{:}",
					sender.name,
					sender.queue_size_warning,
					sender.creation_backtrace,
					Backtrace::force_capture(),
				);
			}
			sender.inner.try_send(event.clone()).is_ok()
		});

		if let Some(metrics) = &*self.metrics {
			for ev in &self.event_streams {
				metrics.event_in(&event, ev.name);
			}
		}
	}
}

impl fmt::Debug for OutChannels {
	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
		f.debug_struct("OutChannels")