ruststream 0.7.2

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

// The harness macros generate the group module, its items and the paths between them, and a
// benchmark function takes its setup value by value because the harness owns the drop; the
// crate's lints are written for the library surface, not for generated benchmark scaffolding.
#![allow(
    missing_docs,
    unused_qualifications,
    unreachable_pub,
    clippy::must_use_candidate,
    clippy::needless_pass_by_value
)]
//! Replying to a transport that reads the payload: the handler returns a value, the runtime
//! encodes it into the dispatch loop's own buffer and lends the bytes to a sink that keeps
//! nothing. This is the form every transport that packs the body into a frame of its own
//! declares, and the framework allocates nothing per message on it.
//!
//! A binary of its own rather than a second scenario in `reply`: mounted beside the in-memory
//! reply in one binary, it gives every leaf of that dispatch stack a second caller, and the
//! in-memory scenario loses its inlining to it.

mod common;

use std::hint::black_box;

use common::{Latch, MESSAGES, Order, Pending, SinkPublish};
use gungraun::{library_benchmark, library_benchmark_group, main};
use ruststream::memory::prelude::*;
use serde::Serialize;

/// A reply with a destination of its own: the mount site adds nothing to it.
#[derive(Debug, Serialize, Outgoing)]
#[outgoing(name = "confirmations")]
struct Confirmation {
    id: u64,
}

#[subscriber("orders", reply)]
async fn confirm(order: &Order, ctx: &mut Context<'_, (), Latch>) -> Confirmation {
    ctx.state().arrived();
    Confirmation {
        id: black_box(order.id),
    }
}

fn app(messages: usize) -> Pending {
    common::pending(messages, 0, |b| {
        b.include(confirm).out_reply(SinkPublish);
    })
}

// The framework allocates nothing per message here, so what is left in the steady state is the
// delivery the bus made.
#[library_benchmark(config = common::config(0, 30))]
#[bench::first(app(1))]
#[bench::base(app(MESSAGES))]
#[bench::twice(app(2 * MESSAGES))]
fn service(app: Pending) {
    common::start_and_drain(app);
}

library_benchmark_group!(name = reply_lending; benchmarks = service);
main!(library_benchmark_groups = reply_lending);