bromine 0.22.1

A flexible ipc protocol (previously rmp-ipc)
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

use crate::utils::call_counter::{get_counter_from_context, increment_counter_for_event};
use crate::utils::protocol::TestProtocolListener;
use crate::utils::{get_free_port, start_server_and_client};
use bromine::prelude::*;
use byteorder::ReadBytesExt;
use bytes::Bytes;
use futures::StreamExt;
use std::io::Read;
use std::time::Duration;

mod utils;

/// When awaiting the reply to an event the handler for the event doesn't get called.
/// Therefore we expect it to have a call count of 0.
#[tokio::test]
async fn it_receives_responses() {
    let port = get_free_port();
    let ctx = get_client_with_server(port).await;
    let mut reply_stream = ctx
        .emit("stream", EmptyPayload)
        .stream_replies()
        .await
        .unwrap();

    let mut reply_stream_2 = ctx
        .emit("stream", EmptyPayload)
        .stream_replies()
        .await
        .unwrap();

    for i in 0u8..=100 {
        if let Some(result) = reply_stream.next().await {
            let event = result.unwrap();
            assert_eq!(event.payload::<NumberPayload>().unwrap().0, i)
        } else {
            panic!("stream 1 has no value {}", i);
        }
        if let Some(result) = reply_stream_2.next().await {
            let event = result.unwrap();
            assert_eq!(event.payload::<NumberPayload>().unwrap().0, i)
        } else {
            panic!("stream 2 has no value {}", i);
        }
    }
    let counter = get_counter_from_context(&ctx).await;
    assert_eq!(counter.get("stream").await, 2);
}

async fn get_client_with_server(port: u8) -> Context {
    start_server_and_client(move || get_builder(port)).await
}

fn get_builder(port: u8) -> IPCBuilder<TestProtocolListener> {
    IPCBuilder::new()
        .address(port)
        .timeout(Duration::from_millis(100))
        .on("stream", callback!(handle_stream_event))
}

async fn handle_stream_event(ctx: &Context, event: Event) -> IPCResult<Response> {
    increment_counter_for_event(ctx, &event).await;
    for i in 0u8..=99 {
        ctx.emit("number", NumberPayload(i)).await?;
    }

    ctx.response(NumberPayload(100))
}

pub struct EmptyPayload;

impl IntoPayload for EmptyPayload {
    fn into_payload(self, _: &Context) -> IPCResult<Bytes> {
        Ok(Bytes::new())
    }
}

pub struct NumberPayload(u8);

impl IntoPayload for NumberPayload {
    fn into_payload(self, _: &Context) -> IPCResult<Bytes> {
        Ok(Bytes::from(vec![self.0]))
    }
}

impl FromPayload for NumberPayload {
    fn from_payload<R: Read>(mut reader: R) -> IPCResult<Self> {
        let num = reader.read_u8()?;

        Ok(NumberPayload(num))
    }
}