strev 0.5.0

Event-driven pub/sub messaging library with compile-time ack safety
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

use std::sync::Arc;

use async_trait::async_trait;
use tokio_stream::StreamExt;

use crate::error::{CloseError, PublishError, SubscribeError};
use crate::message::{Message, Pending};
use crate::outcome::Outcome;
use crate::publisher::Publisher;
use crate::stream::MessageStream;
use crate::subscriber::Subscriber;
use crate::topic::Topic;

type MessageTransformFn = Arc<dyn Fn(&mut Message<Pending>) + Send + Sync>;

pub trait PublisherDecorator: Send + Sync {
    fn decorate(&self, publisher: Box<dyn Publisher>) -> Box<dyn Publisher>;
}

pub trait SubscriberDecorator: Send + Sync {
    fn decorate(&self, subscriber: Box<dyn Subscriber>) -> Box<dyn Subscriber>;
}

pub struct MessageTransformPublisherDecorator {
    pub transform: MessageTransformFn,
}

impl PublisherDecorator for MessageTransformPublisherDecorator {
    fn decorate(&self, publisher: Box<dyn Publisher>) -> Box<dyn Publisher> {
        Box::new(TransformedPublisher {
            inner: publisher,
            transform: self.transform.clone(),
        })
    }
}

struct TransformedPublisher {
    inner: Box<dyn Publisher>,
    transform: MessageTransformFn,
}

#[async_trait]
impl Publisher for TransformedPublisher {
    async fn publish(
        &self,
        topic: &Topic,
        messages: Vec<Message<Pending>>,
    ) -> Result<Vec<Outcome>, PublishError> {
        let transformed = messages
            .into_iter()
            .map(|mut m| {
                (self.transform)(&mut m);
                m
            })
            .collect();
        self.inner.publish(topic, transformed).await
    }

    async fn close(&mut self) -> Result<(), CloseError> {
        self.inner.close().await
    }
}

pub struct MessageTransformSubscriberDecorator {
    pub transform: MessageTransformFn,
}

impl SubscriberDecorator for MessageTransformSubscriberDecorator {
    fn decorate(&self, subscriber: Box<dyn Subscriber>) -> Box<dyn Subscriber> {
        Box::new(TransformedSubscriber {
            inner: subscriber,
            transform: self.transform.clone(),
        })
    }
}

struct TransformedSubscriber {
    inner: Box<dyn Subscriber>,
    transform: MessageTransformFn,
}

#[async_trait]
impl Subscriber for TransformedSubscriber {
    async fn subscribe(&self, topic: &Topic) -> Result<MessageStream, SubscribeError> {
        let mut inner_stream = self.inner.subscribe(topic).await?;
        let (tx, stream) = MessageStream::channel(256);
        let transform = self.transform.clone();

        tokio::spawn(async move {
            while let Some(mut msg) = inner_stream.next().await {
                transform(&mut msg);
                if tx.send(msg).await.is_err() {
                    break;
                }
            }
        });

        Ok(stream)
    }

    async fn close(&mut self) -> Result<(), CloseError> {
        self.inner.close().await
    }
}