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

Code before:
		debug_assert!(stats.is_none());
		*stats = Some(self.stats.clone());
		drop(stats);
		self.event_streams.push(sender);
	}

	/// Returns the number of senders in the collection.
	pub fn len(&self) -> usize {
		self.event_streams.len()
	}

	/// Sends an event.
	pub fn send(&mut self, event: Event) {
		self.event_streams.retain(|sender| {
			sender.inner.unbounded_send(event.clone()).is_ok()
		});

		// The number of elements remaining in `event_streams` corresponds to the number of times
		// we have sent an element on the channel.
		let mut stats = self.stats.lock();
		for _ in 0..self.event_streams.len() {
			increase_stats(&mut stats, &event);
		}
	}

	/// Returns statistics about the content of the channel.
	///
	/// > **Important**: This function returns a mutex lock. As long as it is not disposed of,
	/// >				 receivers will block receiving any message.
	pub fn lock_stats<'a>(&'a self) -> impl Deref<Target = StatsCollect> + 'a {
		self.stats.lock()
	}
}

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()
	}
}

fn increase_stats(stats: &mut StatsCollect, event: &Event) {
	match event {
		Event::Dht(_) => stats.dht_events = stats.dht_events.wrapping_add(1),
		Event::NotificationStreamOpened { engine_id, .. } => {
			let val = stats.notifications_open_messages_count.entry(*engine_id).or_insert(0);
			*val = val.wrapping_add(1);
		},
		Event::NotificationStreamClosed { engine_id, .. } => {
			let val = stats.notifications_closed_messages_count.entry(*engine_id).or_insert(0);

Review comment (suggestion):
```suggestion
	/// Returns statistics about the content of the channels.
```

Code after:

	/// Sends an event.
	pub fn send(&mut self, event: Event) {
		self.event_streams.retain(|sender| {
			sender.inner.unbounded_send(event.clone()).is_ok()
		});

		if let Some(metrics) = &*self.metrics {
			metrics.event_in(&event, self.event_streams.len() as u64);
		}
	}
}

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()
	}
}

struct Metrics {
	// This list is ordered alphabetically
	events_total: CounterVec<U64>,
	notifications_sizes: CounterVec<U64>,
	num_channels: Gauge<U64>,
}

impl Metrics {
	fn register(registry: &Registry) -> Result<Self, PrometheusError> {
		Ok(Self {
			events_total: register(CounterVec::new(
				Opts::new(
					"sub_libp2p_out_events_events_total",
					"Number of broadcast network events that have been sent or received across all \
					 channels"
				),
				&["event_name", "action"]
			)?, registry)?,
			notifications_sizes: register(CounterVec::new(
				Opts::new(
					"sub_libp2p_out_events_notifications_sizes",
					"Size of notification events that have been sent or received across all \
					 channels"
				),
				&["protocol", "action"]
			)?, registry)?,
			num_channels: register(Gauge::new(
				"sub_libp2p_out_events_num_channels",
				"Number of internal active channels that broadcast network events",
			)?, registry)?,