mithril_aggregator/event_store/
transmitter_service.rs

1use anyhow::Context;
2use slog::{Logger, warn};
3use std::fmt::Debug;
4use tokio::sync::mpsc::UnboundedSender;
5
6use mithril_common::{StdResult, logging::LoggerExtensions};
7
8use super::EventMessage;
9
10/// The transmitter service is used to allow inter process channel
11/// communication. This service is used to create multiple transmitters.
12pub struct TransmitterService<MSG>
13where
14    MSG: Debug + Sync + Send,
15{
16    transmitter: UnboundedSender<MSG>,
17    logger: Logger,
18}
19
20impl<MSG> TransmitterService<MSG>
21where
22    MSG: Debug + Sync + Send,
23{
24    /// Instantiate a new Service by passing a MPSC transmitter.
25    pub fn new(transmitter: UnboundedSender<MSG>, logger: Logger) -> Self {
26        Self {
27            transmitter,
28            logger: logger.new_with_component_name::<Self>(),
29        }
30    }
31
32    /// Clone the internal transmitter and return it.
33    pub fn get_transmitter(&self) -> UnboundedSender<MSG> {
34        self.transmitter.clone()
35    }
36}
37
38impl TransmitterService<EventMessage> {
39    /// Send an [EventMessage].
40    pub fn try_send(&self, message: EventMessage) -> StdResult<()> {
41        self.get_transmitter().send(message.clone()).with_context(|| {
42            format!("An error occurred when sending message {message:?} to monitoring.")
43        })
44    }
45
46    /// Send an [EventMessage].
47    ////
48    /// An error when sending a message has no impact on the business.
49    /// If there is one, a warning is issued so the resulting error may be safely ignored by the caller.
50    pub fn send(&self, message: EventMessage) {
51        if let Err(e) = self.try_send(message.clone()) {
52            warn!(self.logger, "Event message error"; "error" => ?e);
53        };
54    }
55}
mithril_aggregator/event_store/transmitter_service.rs

mithril_aggregator/event_store/
transmitter_service.rs
