hermes-async-runtime-components 0.1.0

Modular runtime components implemented using Tokio
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

use alloc::sync::Arc;
use alloc::vec::Vec;
use core::ops::DerefMut;

use cgp::prelude::*;
use futures_core::stream::Stream;
use futures_util::stream::StreamExt;
use hermes_runtime_components::traits::channel::{
    CanCreateChannels, CanStreamReceiver, CanUseChannels, HasChannelTypes,
};
use hermes_runtime_components::traits::mutex::HasMutex;
use hermes_runtime_components::traits::spawn::CanSpawnTask;
use hermes_runtime_components::traits::task::Task;

use crate::stream::traits::boxed::HasBoxedStreamType;
use crate::subscription::impls::multiplex::MultiplexingSubscription;
use crate::subscription::traits::subscription::Subscription;

/**
   Allows multiplexing of a single [`Stream`] into a subscription.
   This is an auto trait implemented by all runtime contexts that implement
   `HasSpawner`, [`HasMutex`], [`CanCreateChannels`], [`CanUseChannels`],
   and [`CanStreamReceiver`].

   When the stream terminates, the subscription also terminates.
*/
pub trait CanStreamSubscription {
    fn stream_subscription<S, T>(&self, stream: S) -> Arc<dyn Subscription<Item = T>>
    where
        S: Stream<Item = T> + Async,
        T: Async + Clone;
}

pub struct StreamSubscriptionTask<Runtime, S, T>
where
    Runtime: HasMutex + HasChannelTypes,
    T: Async,
    S: Stream<Item = T> + Async,
{
    pub stream: S,
    pub task_senders: Arc<Runtime::Mutex<Option<Vec<Runtime::Sender<T>>>>>,
}

impl<Runtime, S, T> Task for StreamSubscriptionTask<Runtime, S, T>
where
    Runtime: HasMutex + CanCreateChannels + CanUseChannels + CanStreamReceiver + HasBoxedStreamType,
    T: Clone + Async,
    S: Stream<Item = T> + Async,
{
    async fn run(self) {
        let task_senders = &self.task_senders;

        self.stream
            .for_each(|item| async move {
                let mut m_senders = Runtime::acquire_mutex(task_senders).await;

                if let Some(senders) = m_senders.deref_mut() {
                    let mut new_senders = Vec::new();

                    for sender in senders.drain(..) {
                        let send_result = Runtime::send(&sender, item.clone()).await;
                        // Remove senders where the receiver side has been dropped,
                        // i.e. keep the ones where sending is successful
                        if send_result.is_ok() {
                            new_senders.push(sender);
                        }
                    }

                    *senders = new_senders;
                }
            })
            .await;

        let mut senders = Runtime::acquire_mutex(&self.task_senders).await;
        *senders = None;
    }
}

impl<Runtime> CanStreamSubscription for Runtime
where
    Runtime: CanSpawnTask
        + HasMutex
        + CanCreateChannels
        + CanUseChannels
        + CanStreamReceiver
        + HasBoxedStreamType,
{
    fn stream_subscription<S, T>(&self, stream: S) -> Arc<dyn Subscription<Item = T>>
    where
        S: Stream<Item = T> + Async,
        T: Async + Clone,
    {
        let stream_senders = Arc::new(Runtime::new_mutex(Some(Vec::new())));

        let task: StreamSubscriptionTask<Runtime, S, T> = StreamSubscriptionTask {
            stream,
            task_senders: stream_senders.clone(),
        };

        self.spawn_task(task);

        let subscription: MultiplexingSubscription<Runtime, T> =
            MultiplexingSubscription { stream_senders };

        Arc::new(subscription)
    }
}