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

use alloc::boxed::Box;
use alloc::sync::Arc;
use core::future::Future;
use core::pin::Pin;

use cgp::prelude::*;
use futures_core::stream::Stream;
use hermes_runtime_components::traits::mutex::{HasMutex, MutexOf};

use crate::subscription::traits::subscription::Subscription;

/**
   An auto trait that is implemented by all runtime contexts that implement
   [`HasMutex`]. This allows simple creation of [`Subscription`] values by
   wrapping an async closure that returns the same thing as
   [`subscribe`](Subscription::subscribe).

   The returned [`Subscription`] also implements guard to skip calling the
   underlying closure once the closure returns `None`.
*/
pub trait CanCreateClosureSubscription {
    fn new_closure_subscription<T: Async>(
        subscribe: impl Fn() -> Pin<
                Box<
                    dyn Future<
                            Output = Option<Pin<Box<dyn Stream<Item = T> + Send + Sync + 'static>>>,
                        > + Send
                        + 'static,
                >,
            > + Send
            + Sync
            + 'static,
    ) -> Arc<dyn Subscription<Item = T>>;
}

impl<Runtime> CanCreateClosureSubscription for Runtime
where
    Runtime: HasMutex,
{
    fn new_closure_subscription<T: Async>(
        subscribe: impl Fn() -> Pin<
                Box<
                    dyn Future<
                            Output = Option<Pin<Box<dyn Stream<Item = T> + Send + Sync + 'static>>>,
                        > + Send
                        + 'static,
                >,
            > + Send
            + Sync
            + 'static,
    ) -> Arc<dyn Subscription<Item = T>> {
        let subscription: SubscriptionClosure<Runtime, T> = SubscriptionClosure {
            terminated: Runtime::new_mutex(false),
            subscribe: Box::new(subscribe),
        };

        Arc::new(subscription)
    }
}

struct SubscriptionClosure<Runtime, T>
where
    Runtime: HasMutex,
{
    terminated: MutexOf<Runtime, bool>,
    subscribe: Box<
        dyn Fn() -> Pin<
                Box<
                    dyn Future<
                            Output = Option<Pin<Box<dyn Stream<Item = T> + Send + Sync + 'static>>>,
                        > + Send
                        + 'static,
                >,
            > + Send
            + Sync
            + 'static,
    >,
}

#[async_trait::async_trait]
impl<Runtime, T: Async> Subscription for SubscriptionClosure<Runtime, T>
where
    Runtime: HasMutex,
{
    type Item = T;

    async fn subscribe(
        &self,
    ) -> Option<Pin<Box<dyn Stream<Item = Self::Item> + Send + Sync + 'static>>> {
        let mut terminated = Runtime::acquire_mutex(&self.terminated).await;

        if *terminated {
            // If a subscription is terminated, it always return `None` from
            // that point onward.
            None
        } else {
            let m_stream = (self.subscribe)().await;

            if m_stream.is_none() {
                *terminated = true;
            }

            m_stream
        }
    }
}