72 lines
1.6 KiB
Rust
72 lines
1.6 KiB
Rust
|
use async_trait::async_trait;
|
||
|
|
use messagebus::{
|
||
|
|
derive::{Error as MbError, Message},
|
||
|
|
error, receivers, AsyncHandler, Bus, Message,
|
||
|
|
};
|
||
|
|
use thiserror::Error;
|
||
|
|
|
||
|
|
#[derive(Debug, Error, MbError)]
|
||
|
|
enum Error {
|
||
|
|
#[error("Error({0})")]
|
||
|
|
Error(anyhow::Error),
|
||
|
|
}
|
||
|
|
|
||
|
|
impl<M: Message> From<error::Error<M>> for Error {
|
||
|
|
fn from(err: error::Error<M>) -> Self {
|
||
|
|
Self::Error(err.into())
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, Message)]
|
||
|
|
struct MsgF32(pub f32);
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, Message)]
|
||
|
|
struct MsgF64(pub f64);
|
||
|
|
|
||
|
|
struct TmpReceiver;
|
||
|
|
|
||
|
|
#[async_trait]
|
||
|
|
impl AsyncHandler<MsgF64> for TmpReceiver {
|
||
|
|
type Error = Error;
|
||
|
|
type Response = ();
|
||
|
|
|
||
|
|
async fn handle(&self, _msg: MsgF64, _bus: &Bus) -> Result<Self::Response, Self::Error> {
|
||
|
|
std::thread::sleep(std::time::Duration::from_millis(100));
|
||
|
|
|
||
|
|
Ok(())
|
||
|
|
}
|
||
|
|
}
|
||
|
|
#[async_trait]
|
||
|
|
impl AsyncHandler<MsgF32> for TmpReceiver {
|
||
|
|
type Error = Error;
|
||
|
|
type Response = ();
|
||
|
|
|
||
|
|
async fn handle(&self, _msg: MsgF32, bus: &Bus) -> Result<Self::Response, Self::Error> {
|
||
|
|
bus.send(MsgF64(12.0)).await.unwrap();
|
||
|
|
bus.flush::<MsgF64>().await;
|
||
|
|
|
||
|
|
Ok(())
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
#[tokio::test]
|
||
|
|
async fn test_backpressure() {
|
||
|
|
let (b, poller) = Bus::build()
|
||
|
|
.register(TmpReceiver)
|
||
|
|
.subscribe_async::<MsgF32>(
|
||
|
|
1,
|
||
|
|
receivers::BufferUnorderedConfig {
|
||
|
|
buffer_size: 1,
|
||
|
|
max_parallel: 1,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
.done()
|
||
|
|
.build();
|
||
|
|
|
||
|
|
b.send(MsgF32(10.0)).await.unwrap();
|
||
|
|
// b.idle_all().await;
|
||
|
|
b.flush_all().await;
|
||
|
|
b.close().await;
|
||
|
|
poller.await;
|
||
|
|
}
|