pipecircuit 0.0.21

Correct. Robust. Fast. Portable. Pick four.
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

use core::convert::Infallible;
use pipecircuit::{
    CircuitFlowOutcome, CircuitSendOutcome, PendingRoutes, Route, Switchboard, flow,
};

#[global_allocator]
static ALLOCATOR: dhat::Alloc = dhat::Alloc;

struct Board;
impl Switchboard for Board {
    type From<'message> = &'message [u8; 1024];
    type To<'message> = &'message [u8; 1024];
    fn route<'message>(&mut self, bytes: Self::From<'message>) -> Route<Self::To<'message>> {
        Route::fanout((bytes, bytes, bytes, bytes))
    }
}
#[test]
fn fanout_moves_borrows_without_allocating_or_copying_payloads() -> Result<(), &'static str> {
    let payload = [47; 1024];
    let mut pending = PendingRoutes::<_, 3>::new();
    let _profiler = dhat::Profiler::builder().testing().build();
    for _ in 0..1024 {
        let mut destinations = [(); 4].into_iter();
        let result = flow(&mut Board, &payload, &mut pending, |bytes| {
            assert!(destinations.next().is_some());
            assert!(core::ptr::eq(bytes, &payload));
            Ok::<CircuitSendOutcome<&[u8; 1024], Infallible>, Infallible>(
                CircuitSendOutcome::Submitted,
            )
        })
        .map_err(|_| "routing refused")?;
        assert_eq!(result, CircuitFlowOutcome::Submitted);
        assert!(destinations.next().is_none());
        assert!(pending.is_empty());
    }
    let stats = dhat::HeapStats::get();
    assert_eq!(stats.total_blocks, 0);
    assert_eq!(stats.total_bytes, 0);
    Ok(())
}