#![cfg(all(
feature = "testing",
feature = "macros",
feature = "memory",
feature = "json"
))]
use ruststream::prelude::*;
use serde::{Deserialize, Serialize};
#[derive(Debug, Deserialize, Outgoing, Serialize, PartialEq)]
struct Order {
id: u64,
quantity: u32,
}
#[derive(Debug, Deserialize, Outgoing, Serialize, PartialEq)]
#[outgoing(name = "confirmations")]
struct Confirmation {
id: u64,
accepted: bool,
}
#[subscriber("orders", publish)]
async fn confirm(order: &Order) -> Confirmation {
Confirmation {
id: order.id,
accepted: order.quantity > 0,
}
}
use ruststream::memory::prelude::*;
use ruststream::testing::TestApp;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn confirms_valid_orders() {
let app = RustStream::new(AppInfo::new("orders-test", "0.0.0")).with_broker(
MemoryBroker::new(),
|b| {
b.include(confirm).out(Reply, Publish);
},
);
let tb = TestApp::start(app).await.expect("start harness");
tb.broker::<MemoryBroker>()
.message(&Order { id: 1, quantity: 2 })
.to("orders")
.publish()
.await
.expect("publish");
tb.broker::<MemoryBroker>()
.subscriber("orders")
.assert_called_once()
.with(&Order { id: 1, quantity: 2 })
.settled(HandlerOutcome::ack());
tb.broker::<MemoryBroker>()
.published::<Confirmation>("confirmations")
.assert_called_once()
.with(&Confirmation {
id: 1,
accepted: true,
});
}