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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
|
//! Listen to external events in your application.
use crate::event::{self, Event};
use crate::Hasher;
use iced_futures::futures::channel::mpsc;
use iced_futures::futures::{self, Future, Stream};
use iced_futures::BoxStream;
use std::hash::Hash;
/// A request to listen to external events.
///
/// Besides performing async actions on demand with [`Command`], most
/// applications also need to listen to external events passively.
///
/// A [`Subscription`] is normally provided to some runtime, like a [`Command`],
/// and it will generate events as long as the user keeps requesting it.
///
/// For instance, you can use a [`Subscription`] to listen to a WebSocket
/// connection, keyboard presses, mouse events, time ticks, etc.
///
/// [`Command`]: crate::Command
pub type Subscription<T> =
iced_futures::Subscription<Hasher, (Event, event::Status), T>;
/// A stream of runtime events.
///
/// It is the input of a [`Subscription`] in the native runtime.
pub type EventStream = BoxStream<(Event, event::Status)>;
/// A native [`Subscription`] tracker.
pub type Tracker =
iced_futures::subscription::Tracker<Hasher, (Event, event::Status)>;
pub use iced_futures::subscription::Recipe;
/// Returns a [`Subscription`] to all the runtime events.
///
/// This subscription will notify your application of any [`Event`] that was
/// not captured by any widget.
pub fn events() -> Subscription<Event> {
events_with(|event, status| match status {
event::Status::Ignored => Some(event),
event::Status::Captured => None,
})
}
/// Returns a [`Subscription`] that filters all the runtime events with the
/// provided function, producing messages accordingly.
///
/// This subscription will call the provided function for every [`Event`]
/// handled by the runtime. If the function:
///
/// - Returns `None`, the [`Event`] will be discarded.
/// - Returns `Some` message, the `Message` will be produced.
pub fn events_with<Message>(
f: fn(Event, event::Status) -> Option<Message>,
) -> Subscription<Message>
where
Message: 'static + Send,
{
#[derive(Debug, Clone, Copy, Hash)]
struct Events(u64);
let hash = {
use std::hash::Hasher as _;
let mut hasher = Hasher::default();
f.hash(&mut hasher);
hasher.finish()
};
Subscription::from_recipe(Runner {
initial: Events(hash),
spawn: move |_, events| {
use futures::future;
use futures::stream::StreamExt;
events.filter_map(move |(event, status)| {
future::ready(f(event, status))
})
},
})
}
/// Returns a [`Subscription`] that will create and asynchronously run the
/// [`Stream`] returned by the provided closure.
///
/// The `initial` state will be used to uniquely identify the [`Subscription`].
pub fn run<T, S, Message>(
initial: T,
f: impl FnOnce(T) -> S + 'static,
) -> Subscription<Message>
where
Message: 'static,
T: Clone + Hash + 'static,
S: Stream<Item = Message> + Send + 'static,
{
Subscription::from_recipe(Runner {
initial,
spawn: move |initial, _| f(initial),
})
}
/// Returns a [`Subscription`] that will create and asynchronously run a
/// [`Stream`] that will call the provided closure to produce every `Message`.
///
/// The `initial` state will be used to uniquely identify the [`Subscription`].
pub fn unfold<T, Fut, Message>(
initial: T,
mut f: impl FnMut(T) -> Fut + Send + Sync + 'static,
) -> Subscription<Message>
where
Message: 'static,
T: Clone + Hash + Send + 'static,
Fut: Future<Output = (Message, T)> + Send + 'static,
{
use futures::future::FutureExt;
run(initial, move |initial| {
futures::stream::unfold(initial, move |state| f(state).map(Some))
})
}
/// Returns a [`Subscription`] that will open a channel and asynchronously run a
/// [`Stream`] that will call the provided closure to produce every `Message`.
///
/// When the [`Subscription`] starts, an `on_ready` message will be produced
/// containing the [`mpsc::Sender`] end of the channel, which can be used by
/// the parent application to send `Input` to the running [`Subscription`].
///
/// The provided closure should use the [`mpsc::Receiver`] argument to await for
/// any `Input`.
///
/// This function is really useful to create asynchronous workers with
/// bidirectional communication with a parent application.
///
/// The `initial` state will be used to uniquely identify the [`Subscription`].
pub fn worker<T, Fut, Message, Input>(
initial: T,
on_ready: impl FnOnce(mpsc::Sender<Input>) -> Message + 'static,
f: impl FnMut(T, &mut mpsc::Receiver<Input>) -> Fut + Send + Sync + 'static,
) -> Subscription<Message>
where
T: Clone + Hash + Send + 'static,
Fut: Future<Output = (Message, T)> + Send + 'static,
Message: Send + 'static,
Input: Send + 'static,
{
use futures::future;
use futures::stream::StreamExt;
run(initial, move |initial| {
let (sender, receiver) = mpsc::channel(100);
futures::stream::once(future::ready(on_ready(sender))).chain(
futures::stream::unfold(
(f, initial, receiver),
move |(mut f, state, mut receiver)| async {
let (message, state) = f(state, &mut receiver).await;
Some((message, (f, state, receiver)))
},
),
)
})
}
struct Runner<T, F, S, Message>
where
F: FnOnce(T, EventStream) -> S,
S: Stream<Item = Message>,
{
initial: T,
spawn: F,
}
impl<T, S, F, Message> Recipe<Hasher, (Event, event::Status)>
for Runner<T, F, S, Message>
where
T: Clone + Hash + 'static,
F: FnOnce(T, EventStream) -> S,
S: Stream<Item = Message> + Send + 'static,
{
type Output = Message;
fn hash(&self, state: &mut Hasher) {
std::any::TypeId::of::<T>().hash(state);
self.initial.hash(state);
}
fn stream(self: Box<Self>, input: EventStream) -> BoxStream<Self::Output> {
use futures::stream::StreamExt;
(self.spawn)(self.initial, input).boxed()
}
}
|
