ruststream 0.7.5

Async messaging framework for Rust: broker-agnostic traits, router, codecs, and a conformance harness for broker authors.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

//! A middleware chain assembled at run time: one `Block` per tenant the configuration lists,
//! frozen into a `DynStack` around one registration. The scenario is a test.
//!
//! ```text
//! cargo test --example dyn_stack --features testing,macros,memory,json
//! ```

// A plain build compiles the services; the tests that use them exist only in the test build.
#![cfg_attr(not(test), allow(dead_code, unused_imports))]

use std::pin::Pin;

use ruststream::HeaderMap;
use ruststream::memory::prelude::*;
use ruststream::runtime::{DynMiddleware, Next};
use serde::{Deserialize, Serialize};

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct Order {
    id: u64,
}

/// The headers of a delivery that names its tenant.
fn tenant(name: &'static str) -> HeaderMap {
    let mut headers = HeaderMap::new();
    headers.insert("tenant", name);
    headers
}

// --8<-- [start:middleware]
/// Drops the deliveries of one tenant.
struct Block {
    tenant: String,
}

impl<I: Send + Sync> DynMiddleware<I> for Block {
    fn handle<'a>(
        &'a self,
        input: &'a I,
        ctx: &'a mut Context<'_>,
        next: Next<'a, I>,
    ) -> Pin<Box<dyn Future<Output = HandlerOutcome> + Send + 'a>> {
        Box::pin(async move {
            if ctx.headers().get("tenant") == Some(self.tenant.as_bytes()) {
                tracing::warn!(subscription = ctx.name(), tenant = %self.tenant, "blocked");
                return HandlerOutcome::drop();
            }
            next.run(input, ctx).await
        })
    }
}
// --8<-- [end:middleware]

mod service {
    use std::error::Error;
    use std::sync::Arc;

    use ruststream::memory::MemoryMessage;
    use ruststream::memory::prelude::*;
    use ruststream::runtime::{DynMiddleware, DynStack};
    use ruststream::testing::{Outcome, TestApp};

    use super::{Block, Order, tenant};

    // --8<-- [start:app]
    #[subscriber("orders")]
    async fn accept(order: &Order) -> HandlerOutcome {
        tracing::info!(order.id, "accepted");
        HandlerOutcome::ack()
    }

    /// The chain comes from configuration: one `Block` per tenant the operator listed.
    fn app(blocked: &[&str]) -> impl App {
        let chain = blocked.iter().map(|tenant| {
            let block = Block {
                tenant: (*tenant).to_owned(),
            };
            Arc::new(block) as Arc<dyn DynMiddleware<MemoryMessage>>
        });
        let stack = DynStack::new(chain);
        RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
            // The stack wraps one registration, which sees the broker's own message.
            b.include_router(Router::new().include(accept).layer(stack));
        })
    }
    // --8<-- [end:app]

    #[tokio::test]
    async fn a_blocked_tenant_is_dropped() -> Result<(), Box<dyn Error>> {
        let tb = TestApp::start(app(&["globex"])).await?;
        let broker = tb.broker::<MemoryBroker>();
        let order = Order { id: 7 };
        for name in ["acme", "globex"] {
            broker
                .message(&order)
                .with_headers(tenant(name))
                .to("orders")
                .publish()
                .await?;
        }

        let outcomes = broker.subscriber("orders").outcomes();
        assert_eq!(outcomes, [Outcome::Ack, Outcome::Drop]);
        Ok(())
    }
}

// The scenarios are the tests above; the command in the header runs them.
fn main() {}