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

use std::{fmt::Display, pin::Pin};

use async_pub_sub::{Publisher, PublisherImpl, Request, Result, Subscriber, SubscriberImpl};
use futures::{
    FutureExt, Stream, StreamExt,
    future::BoxFuture,
    stream::{self, SelectAll},
};

// TODO: fix the request response logging in the forwarder
struct LoggingForwarder<Message>
where
    Message: Display + Send + 'static,
{
    name: &'static str,
    messages: Option<SelectAll<Pin<Box<dyn Stream<Item = Message> + Send + Sync + 'static>>>>,
}

impl<Message> LoggingForwarder<Message>
where
    Message: Display + Send + 'static,
{
    pub fn new(name: &'static str) -> Self {
        Self {
            name,
            messages: Some(SelectAll::new()),
        }
    }
}

impl<Message> Subscriber for LoggingForwarder<Message>
where
    Message: Display + Send + 'static,
{
    type Message = Message;

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

    fn subscribe_to(&mut self, publisher: &mut dyn Publisher<Message = Message>) -> Result<()> {
        let stream = publisher.get_message_stream(self.name)?;

        self.messages.as_mut().unwrap().push(stream);

        Ok(())
    }

    fn receive(&mut self) -> BoxFuture<Message> {
        panic!("LoggingForwarder does not implement receive method")
    }
}

impl<Message> Publisher for LoggingForwarder<Message>
where
    Message: Display + Send + 'static,
{
    type Message = Message;

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

    fn publish(&self, _message: Message) -> futures::future::BoxFuture<Result<()>> {
        async move { panic!("LoggingForwarder does not implement publish method") }.boxed()
    }

    fn get_message_stream(
        &mut self,
        subscriber_name: &'static str,
    ) -> Result<Pin<Box<dyn Stream<Item = Message> + Send + Sync + 'static>>> {
        let messages = self.messages.take().unwrap();
        let name = self.name;

        let stream = Box::pin(stream::unfold(messages, move |mut messages| async move {
            let message = messages.select_next_some().await;
            log::info!("[{}] -> [{}]: {}", name, subscriber_name, message);
            Some((message, messages))
        }));

        log::info!(
            "({}) <-> ({}): {}",
            self.name,
            subscriber_name,
            std::any::type_name::<Message>()
        );

        Ok(stream)
    }
}

#[test_log::test(tokio::test)]
async fn test_message_forwarder() -> Result<()> {
    // -- Setup & Fixtures
    let mut subscriber = SubscriberImpl::<Request<i32, i32>>::new("subscriber");
    let mut forwarder = LoggingForwarder::new("forwarder");
    let mut publisher = PublisherImpl::new("publisher", 10);

    forwarder.subscribe_to(&mut publisher)?;
    subscriber.subscribe_to(&mut forwarder)?;

    // -- Exec
    let publisher_task = tokio::spawn(async move {
        let (request, response) = Request::new(42);
        publisher
            .publish(request)
            .await
            .expect("request published successfully");
        assert_eq!(response.await.expect("request successful"), 43);
    });

    let subscriber_task = tokio::spawn(async move {
        let request = subscriber.receive().await;
        let response = request.content + 1;

        request.respond(response);
    });

    // -- Check
    tokio::try_join!(publisher_task, subscriber_task)?;

    Ok(())
}