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

Code before:
	}
}

/// Collection of senders.
pub struct OutChannels {
	event_streams: Vec<Sender>,
	/// The metrics we collect. A clone of this is sent to each [`Receiver`] associated with this
	/// object.
	metrics: Option<Metrics>,
}

impl OutChannels {
	/// 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 })
	}

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

		if let Some(metrics) = self.metrics.as_ref() {
			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| {
			let current_pending = sender.inner.len();
			if current_pending >= 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(),
				);
			} else if sender.warning_fired &&
				current_pending < sender.queue_size_warning.wrapping_div(2)
			{
				sender.warning_fired = false;

Review comment (suggestion):
```suggestion
		if let Some(metrics) = &self.metrics {
```

Code after:
impl fmt::Debug for Receiver {
	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
		f.debug_tuple("Receiver").finish()
	}
}

impl Drop for Receiver {
	fn drop(&mut self) {
		if !self.inner.is_terminated() {
			// Empty the list to properly decrease the metrics.
			while let Some(Some(_)) = self.next().now_or_never() {}
		}
	}
}

/// Collection of senders.
pub struct OutChannels {
	event_streams: Vec<Sender>,
	/// The metrics we collect. A clone of this is sent to each [`Receiver`] associated with this
	/// object.
	metrics: Option<Metrics>,
}

impl OutChannels {
	/// 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 })
	}

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

		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| {
			let current_pending = sender.inner.len();
			if current_pending >= sender.queue_size_warning {
				if sender.warning_fired == SenderWarningState::NotFired {
					error!(