-
Notifications
You must be signed in to change notification settings - Fork 264
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Adds the ability to monitor data flow through a connection, i.e. reports the quant metrics of incoming and outgoing data through a connection to the broker.
- Loading branch information
Showing
16 changed files
with
316 additions
and
203 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| use rumqttd::{Broker, Config, GetMeter, Notification}; | ||
| use std::{thread, time::Duration}; | ||
|
|
||
| fn main() { | ||
| pretty_env_logger::init(); | ||
|
|
||
| // As examples are compiled as seperate binary so this config is current path dependent. Run it | ||
| // from root of this crate | ||
| let config = config::Config::builder() | ||
| .add_source(config::File::with_name("demo.toml")) | ||
| .build() | ||
| .unwrap(); | ||
|
|
||
| let config: Config = config.try_deserialize().unwrap(); | ||
|
|
||
| dbg!(&config); | ||
|
|
||
| let broker = Broker::new(config); | ||
| let meters = broker.meters().unwrap(); | ||
|
|
||
| let (mut link_tx, mut link_rx) = broker.link("consumer").unwrap(); | ||
| link_tx.subscribe("hello/+/world").unwrap(); | ||
| thread::spawn(move || { | ||
| let mut count = 0; | ||
| loop { | ||
| let notification = match link_rx.recv().unwrap() { | ||
| Some(v) => v, | ||
| None => continue, | ||
| }; | ||
|
|
||
| match notification { | ||
| Notification::Forward(forward) => { | ||
| count += 1; | ||
| println!( | ||
| "Topic = {:?}, Count = {}, Payload = {} bytes", | ||
| forward.publish.topic, | ||
| count, | ||
| forward.publish.payload.len() | ||
| ); | ||
| } | ||
| v => { | ||
| println!("{:?}", v); | ||
| } | ||
| } | ||
| } | ||
| }); | ||
|
|
||
| for i in 0..5 { | ||
| let client_id = format!("client_{i}"); | ||
| let topic = format!("hello/{}/world", client_id); | ||
| let payload = vec![0u8; 1_000]; // 0u8 is one byte, so total ~1KB | ||
| let (mut link_tx, _link_rx) = broker.link(&client_id).expect("New link should be made"); | ||
|
|
||
| thread::spawn(move || { | ||
| for _ in 0..100 { | ||
| thread::sleep(Duration::from_secs(1)); | ||
| link_tx.publish(topic.clone(), payload.clone()).unwrap(); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| loop { | ||
| // Router meters | ||
| let request = GetMeter::Router; | ||
| let v = meters.get(request).unwrap(); | ||
| println!("{:#?}", v); | ||
|
|
||
| // Publisher meters | ||
| for i in 0..5 { | ||
| let client_id = format!("client_{i}"); | ||
| let request = GetMeter::Connection(client_id); | ||
| let v = meters.get(request).unwrap(); | ||
| println!("{:#?}", v); | ||
| } | ||
|
|
||
| // Commitlog meters | ||
| let request = GetMeter::Subscription("hello/+/world".to_owned()); | ||
| let v = meters.get(request).unwrap(); | ||
| println!("{:#?}", v); | ||
|
|
||
| // Consumer meters | ||
| let request = GetMeter::Connection("consumer".to_owned()); | ||
| let v = meters.get(request).unwrap(); | ||
| println!("{:#?}", v); | ||
|
|
||
| thread::sleep(Duration::from_secs(5)); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| use crate::router::{Event, GetMeter, Meter}; | ||
| use crate::ConnectionId; | ||
| use flume::{Receiver, RecvError, RecvTimeoutError, SendError, Sender, TrySendError}; | ||
|
|
||
| #[derive(Debug, thiserror::Error)] | ||
| pub enum LinkError { | ||
| #[error("Channel try send error")] | ||
| TrySend(#[from] TrySendError<(ConnectionId, Event)>), | ||
| #[error("Channel send error")] | ||
| Send(#[from] SendError<(ConnectionId, Event)>), | ||
| #[error("Channel recv error")] | ||
| Recv(#[from] RecvError), | ||
| #[error("Channel timeout recv error")] | ||
| RecvTimeout(#[from] RecvTimeoutError), | ||
| #[error("Timeout = {0}")] | ||
| Elapsed(#[from] tokio::time::error::Elapsed), | ||
| } | ||
|
|
||
| pub struct MetersLink { | ||
| pub(crate) meter_id: ConnectionId, | ||
| router_tx: Sender<(ConnectionId, Event)>, | ||
| router_rx: Receiver<(ConnectionId, Meter)>, | ||
| } | ||
|
|
||
| impl MetersLink { | ||
| pub fn new(router_tx: Sender<(ConnectionId, Event)>) -> Result<MetersLink, LinkError> { | ||
| let (tx, rx) = flume::bounded(5); | ||
| router_tx.send((0, Event::NewMeter(tx)))?; | ||
| let (meter_id, _meter) = rx.recv()?; | ||
|
|
||
| let link = MetersLink { | ||
| meter_id, | ||
| router_tx, | ||
| router_rx: rx, | ||
| }; | ||
|
|
||
| Ok(link) | ||
| } | ||
|
|
||
| pub async fn init(router_tx: Sender<(ConnectionId, Event)>) -> Result<MetersLink, LinkError> { | ||
| let (tx, rx) = flume::bounded(5); | ||
| router_tx.send((0, Event::NewMeter(tx)))?; | ||
| let (meter_id, _meter) = rx.recv_async().await?; | ||
|
|
||
| let link = MetersLink { | ||
| meter_id, | ||
| router_tx, | ||
| router_rx: rx, | ||
| }; | ||
|
|
||
| Ok(link) | ||
| } | ||
|
|
||
| pub fn get(&self, meter: GetMeter) -> Result<Meter, LinkError> { | ||
| self.router_tx | ||
| .send((self.meter_id, Event::GetMeter(meter)))?; | ||
| let (_meter_id, meter) = self.router_rx.recv()?; | ||
| Ok(meter) | ||
| } | ||
|
|
||
| pub async fn fetch(&self, meter: GetMeter) -> Result<Meter, LinkError> { | ||
| self.router_tx | ||
| .send_async((self.meter_id, Event::GetMeter(meter))) | ||
| .await?; | ||
| let (_meter_id, meter) = self.router_rx.recv_async().await?; | ||
| Ok(meter) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.