async_pub_sub 0.2.1

A library aiming at making async pub-sub easier in Rust
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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

use std::fmt::Display;

use async_pub_sub::{Publisher, PublisherImpl, Request, Result, Subscriber, SubscriberImpl};
use futures::{FutureExt, future::BoxFuture};

#[derive(Debug, PartialEq)]
struct Foo(i32);

impl Display for Foo {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:?}", self.0)
    }
}

#[derive(Debug, PartialEq)]
struct Bar(String);

impl Display for Bar {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:?}", self.0)
    }
}

#[derive(Debug)]
enum ServiceRequest {
    Foo(Request<Foo, i32>),
    Bar(Request<Bar, String>),
}

impl From<Request<Foo, i32>> for ServiceRequest {
    fn from(request: Request<Foo, i32>) -> Self {
        Self::Foo(request)
    }
}

impl From<Request<Bar, String>> for ServiceRequest {
    fn from(request: Request<Bar, String>) -> Self {
        Self::Bar(request)
    }
}

impl Display for ServiceRequest {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:?}", self)
    }
}

struct Service {
    subscriber: SubscriberImpl<ServiceRequest>,
}

impl Service {
    pub fn new() -> Self {
        let subscriber = SubscriberImpl::new("Service");

        Self { subscriber }
    }

    pub async fn run(&mut self) -> Result<()> {
        loop {
            let request = self.subscriber.receive().await;
            self.handle_request(request);
        }
    }

    fn handle_request(&mut self, request: ServiceRequest) {
        match request {
            ServiceRequest::Foo(request) => {
                let response = self.foo(&request.content);
                request.respond(response);
            }
            ServiceRequest::Bar(request) => {
                let response = self.bar(&request.content);
                request.respond(response);
            }
        }
        {}
    }

    fn foo(&self, value: &Foo) -> i32 {
        let Foo(value) = value;
        value + 1
    }

    fn bar(&mut self, value: &Bar) -> String {
        let Bar(value) = value;
        format!("bar: {}", value)
    }
}

impl Subscriber for Service {
    type Message = ServiceRequest;

    fn get_name(&self) -> &'static str {
        self.subscriber.get_name()
    }

    fn subscribe_to(
        &mut self,
        publisher: &mut dyn Publisher<Message = Self::Message>,
    ) -> Result<()> {
        self.subscriber.subscribe_to(publisher)
    }

    fn receive(&mut self) -> BoxFuture<Self::Message> {
        self.subscriber.receive().boxed()
    }
}

#[test_log::test(tokio::test)]
async fn test_direct_rpc() -> Result<()> {
    // -- Setup & Fixtures
    let mut publisher = PublisherImpl::new("publisher", 1);
    let mut service = Service::new();

    service.subscribe_to(&mut publisher)?;

    tokio::spawn(async move {
        service.run().await.unwrap();
    });

    // -- Exec & Check
    let (request, response) = Request::new(Foo(42));
    publisher
        .publish(request.into())
        .await
        .expect("request published successfully");
    assert_eq!(response.await.expect("request successul"), 43);

    let (request, response) = Request::new(Bar("hello".to_string()));
    publisher
        .publish(request.into())
        .await
        .expect("request published successfully");
    assert_eq!(response.await.expect("request successul"), "bar: hello");

    Ok(())
}