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

use std::time::Duration;

use async_pub_sub::{DebuggingPublisherLayer, Publisher, PublisherBuilder, PublisherImpl};
use async_pub_sub_macros::DerivePublisher;
use rand::Rng;

const NAME: &str = "Timer";

// TODO: work on the broadcast

#[derive(Debug)]
pub struct DataConsumerTimerNotification;
#[derive(Debug)]
pub struct DataProducerTimerNotification;
#[derive(Debug)]
pub struct CacheTimerNotification;

#[derive(DerivePublisher)]
pub struct TimerService {
    #[publisher(DataConsumerTimerNotification)]
    data_consumer_publisher: Box<dyn Publisher<Message = DataConsumerTimerNotification>>,
    #[publisher(DataProducerTimerNotification)]
    data_producer_publisher: Box<dyn Publisher<Message = DataProducerTimerNotification>>,
    #[publisher(CacheTimerNotification)]
    cache_publisher: Box<dyn Publisher<Message = CacheTimerNotification>>,
}

impl TimerService {
    pub fn new() -> Self {
        Self {
            data_consumer_publisher: Box::new(
                PublisherBuilder::new()
                    .layer(DebuggingPublisherLayer)
                    .publisher(PublisherImpl::new(NAME, 10)),
            ),
            data_producer_publisher: Box::new(
                PublisherBuilder::new()
                    .layer(DebuggingPublisherLayer)
                    .publisher(PublisherImpl::new(NAME, 10)),
            ),
            cache_publisher: Box::new(
                PublisherBuilder::new()
                    .layer(DebuggingPublisherLayer)
                    .publisher(PublisherImpl::new(NAME, 10)),
            ),
        }
    }

    pub async fn run(self) {
        let data_consumer_task = async move {
            let mut rng = rand::rng();
            loop {
                let second_count = rng.random_range(3..6);
                log::info!("[{}] notifying data customer in {}s", NAME, second_count);
                tokio::time::sleep(Duration::from_secs(second_count)).await;
                self.data_consumer_publisher
                    .publish(DataConsumerTimerNotification)
                    .await
                    .unwrap();
            }
        };

        let data_producer_task = async move {
            let mut rng = rand::rng();
            loop {
                let second_count = rng.random_range(1..4);
                log::info!("[{}] notifying data producer in {}s", NAME, second_count);
                tokio::time::sleep(Duration::from_secs(second_count)).await;
                self.data_producer_publisher
                    .publish(DataProducerTimerNotification)
                    .await
                    .unwrap();
            }
        };

        let cache_publisher = async move {
            let mut rng = rand::rng();
            loop {
                let second_count = rng.random_range(5..8);
                log::info!("[{}] notifying data cache in {}s", NAME, second_count);
                tokio::time::sleep(Duration::from_secs(second_count)).await;
                self.cache_publisher
                    .publish(CacheTimerNotification)
                    .await
                    .unwrap();
            }
        };

        tokio::join!(data_consumer_task, data_producer_task, cache_publisher);
    }
}