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
124
|
//! Run commands and keep track of subscriptions.
use crate::{subscription, Command, Executor, Subscription};
use futures::{channel::mpsc, Sink};
use std::marker::PhantomData;
/// A batteries-included runtime of commands and subscriptions.
///
/// If you have an [`Executor`], a [`Runtime`] can be leveraged to run any
/// [`Command`] or [`Subscription`] and get notified of the results!
///
/// [`Runtime`]: struct.Runtime.html
/// [`Executor`]: executor/trait.Executor.html
/// [`Command`]: struct.Command.html
/// [`Subscription`]: subscription/struct.Subscription.html
#[derive(Debug)]
pub struct Runtime<Hasher, Event, Executor, Sender, Message> {
executor: Executor,
sender: Sender,
subscriptions: subscription::Tracker<Hasher, Event>,
_message: PhantomData<Message>,
}
impl<Hasher, Event, Executor, Sender, Message>
Runtime<Hasher, Event, Executor, Sender, Message>
where
Hasher: std::hash::Hasher + Default,
Event: Send + Clone + 'static,
Executor: self::Executor,
Sender:
Sink<Message, Error = mpsc::SendError> + Unpin + Send + Clone + 'static,
Message: Send + 'static,
{
/// Creates a new empty [`Runtime`].
///
/// You need to provide:
/// - an [`Executor`] to spawn futures
/// - a `Sender` implementing `Sink` to receive the results
///
/// [`Runtime`]: struct.Runtime.html
pub fn new(executor: Executor, sender: Sender) -> Self {
Self {
executor,
sender,
subscriptions: subscription::Tracker::new(),
_message: PhantomData,
}
}
/// Runs the given closure inside the [`Executor`] of the [`Runtime`].
///
/// See [`Executor::enter`] to learn more.
///
/// [`Executor`]: executor/trait.Executor.html
/// [`Runtime`]: struct.Runtime.html
/// [`Executor::enter`]: executor/trait.Executor.html#method.enter
pub fn enter<R>(&self, f: impl FnOnce() -> R) -> R {
self.executor.enter(f)
}
/// Spawns a [`Command`] in the [`Runtime`].
///
/// The resulting `Message` will be forwarded to the `Sender` of the
/// [`Runtime`].
///
/// [`Command`]: struct.Command.html
/// [`Runtime`]: struct.Runtime.html
pub fn spawn(&mut self, command: Command<Message>) {
use futures::{FutureExt, SinkExt};
let futures = command.futures();
for future in futures {
let mut sender = self.sender.clone();
let future = future.then(|message| async move {
let _ = sender.send(message).await;
()
});
self.executor.spawn(future);
}
}
/// Tracks a [`Subscription`] in the [`Runtime`].
///
/// It will spawn new streams or close old ones as necessary! See
/// [`Tracker::update`] to learn more about this!
///
/// [`Subscription`]: subscription/struct.Subscription.html
/// [`Runtime`]: struct.Runtime.html
/// [`Tracker::update`]: subscription/struct.Tracker.html#method.update
pub fn track(
&mut self,
subscription: Subscription<Hasher, Event, Message>,
) {
let Runtime {
executor,
subscriptions,
sender,
..
} = self;
let futures = executor
.enter(|| subscriptions.update(subscription, sender.clone()));
for future in futures {
executor.spawn(future);
}
}
/// Broadcasts an event to all the subscriptions currently alive in the
/// [`Runtime`].
///
/// See [`Tracker::broadcast`] to learn more.
///
/// [`Runtime`]: struct.Runtime.html
/// [`Tracker::broadcast`]:
/// subscription/struct.Tracker.html#method.broadcast
pub fn broadcast(&mut self, event: Event) {
self.subscriptions.broadcast(event);
}
}
|
