summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--examples/download_progress/src/download.rs17
-rw-r--r--examples/websocket/src/echo.rs100
-rw-r--r--native/src/subscription.rs91
3 files changed, 122 insertions, 86 deletions
diff --git a/examples/download_progress/src/download.rs b/examples/download_progress/src/download.rs
index 39dd843f..cd7647e8 100644
--- a/examples/download_progress/src/download.rs
+++ b/examples/download_progress/src/download.rs
@@ -18,10 +18,7 @@ pub struct Download<I> {
url: String,
}
-async fn download<I: Copy>(
- id: I,
- state: State,
-) -> (Option<(I, Progress)>, State) {
+async fn download<I: Copy>(id: I, state: State) -> ((I, Progress), State) {
match state {
State::Ready(url) => {
let response = reqwest::get(&url).await;
@@ -30,7 +27,7 @@ async fn download<I: Copy>(
Ok(response) => {
if let Some(total) = response.content_length() {
(
- Some((id, Progress::Started)),
+ (id, Progress::Started),
State::Downloading {
response,
total,
@@ -38,10 +35,10 @@ async fn download<I: Copy>(
},
)
} else {
- (Some((id, Progress::Errored)), State::Finished)
+ ((id, Progress::Errored), State::Finished)
}
}
- Err(_) => (Some((id, Progress::Errored)), State::Finished),
+ Err(_) => ((id, Progress::Errored), State::Finished),
}
}
State::Downloading {
@@ -55,7 +52,7 @@ async fn download<I: Copy>(
let percentage = (downloaded as f32 / total as f32) * 100.0;
(
- Some((id, Progress::Advanced(percentage))),
+ (id, Progress::Advanced(percentage)),
State::Downloading {
response,
total,
@@ -63,8 +60,8 @@ async fn download<I: Copy>(
},
)
}
- Ok(None) => (Some((id, Progress::Finished)), State::Finished),
- Err(_) => (Some((id, Progress::Errored)), State::Finished),
+ Ok(None) => ((id, Progress::Finished), State::Finished),
+ Err(_) => ((id, Progress::Errored), State::Finished),
},
State::Finished => {
// We do not let the stream die, as it would start a
diff --git a/examples/websocket/src/echo.rs b/examples/websocket/src/echo.rs
index e74768a6..f9807172 100644
--- a/examples/websocket/src/echo.rs
+++ b/examples/websocket/src/echo.rs
@@ -13,63 +13,67 @@ use std::fmt;
pub fn connect() -> Subscription<Event> {
struct Connect;
- subscription::unfold(
+ subscription::channel(
std::any::TypeId::of::<Connect>(),
- State::Disconnected,
- |state| async move {
- match state {
- State::Disconnected => {
- const ECHO_SERVER: &str = "ws://localhost:3030";
-
- match async_tungstenite::tokio::connect_async(ECHO_SERVER)
+ 100,
+ |mut output| async move {
+ let mut state = State::Disconnected;
+
+ loop {
+ match &mut state {
+ State::Disconnected => {
+ const ECHO_SERVER: &str = "ws://127.0.0.1:3030";
+
+ match async_tungstenite::tokio::connect_async(
+ ECHO_SERVER,
+ )
.await
- {
- Ok((websocket, _)) => {
- let (sender, receiver) = mpsc::channel(100);
-
- (
- Some(Event::Connected(Connection(sender))),
- State::Connected(websocket, receiver),
- )
- }
- Err(_) => {
- tokio::time::sleep(
- tokio::time::Duration::from_secs(1),
- )
- .await;
+ {
+ Ok((websocket, _)) => {
+ let (sender, receiver) = mpsc::channel(100);
+
+ let _ = output
+ .send(Event::Connected(Connection(sender)))
+ .await;
- (Some(Event::Disconnected), State::Disconnected)
+ state = State::Connected(websocket, receiver);
+ }
+ Err(_) => {
+ tokio::time::sleep(
+ tokio::time::Duration::from_secs(1),
+ )
+ .await;
+
+ let _ = output.send(Event::Disconnected).await;
+ }
}
}
- }
- State::Connected(mut websocket, mut input) => {
- let mut fused_websocket = websocket.by_ref().fuse();
-
- futures::select! {
- received = fused_websocket.select_next_some() => {
- match received {
- Ok(tungstenite::Message::Text(message)) => {
- (
- Some(Event::MessageReceived(Message::User(message))),
- State::Connected(websocket, input)
- )
- }
- Ok(_) => {
- (None, State::Connected(websocket, input))
- }
- Err(_) => {
- (Some(Event::Disconnected), State::Disconnected)
+ State::Connected(websocket, input) => {
+ let mut fused_websocket = websocket.by_ref().fuse();
+
+ futures::select! {
+ received = fused_websocket.select_next_some() => {
+ match received {
+ Ok(tungstenite::Message::Text(message)) => {
+ let _ = output.send(Event::MessageReceived(Message::User(message))).await;
+ }
+ Err(_) => {
+ let _ = output.send(Event::Disconnected).await;
+
+ state = State::Disconnected;
+ }
+ Ok(_) => continue,
}
}
- }
- message = input.select_next_some() => {
- let result = websocket.send(tungstenite::Message::Text(message.to_string())).await;
+ message = input.select_next_some() => {
+ let result = websocket.send(tungstenite::Message::Text(message.to_string())).await;
+
+ if result.is_err() {
+ let _ = output.send(Event::Disconnected).await;
- if result.is_ok() {
- (None, State::Connected(websocket, input))
- } else {
- (Some(Event::Disconnected), State::Disconnected)
+ state = State::Disconnected;
+ }
}
}
}
diff --git a/native/src/subscription.rs b/native/src/subscription.rs
index 16e78e82..0ff5e320 100644
--- a/native/src/subscription.rs
+++ b/native/src/subscription.rs
@@ -3,6 +3,8 @@ use crate::event::{self, Event};
use crate::window;
use crate::Hasher;
+use iced_futures::futures::channel::mpsc;
+use iced_futures::futures::never::Never;
use iced_futures::futures::{self, Future, Stream};
use iced_futures::{BoxStream, MaybeSend};
@@ -133,6 +135,27 @@ where
/// [`Stream`] that will call the provided closure to produce every `Message`.
///
/// The `id` will be used to uniquely identify the [`Subscription`].
+pub fn unfold<I, T, Fut, Message>(
+ id: I,
+ initial: T,
+ mut f: impl FnMut(T) -> Fut + MaybeSend + Sync + 'static,
+) -> Subscription<Message>
+where
+ I: Hash + 'static,
+ T: MaybeSend + 'static,
+ Fut: Future<Output = (Message, T)> + MaybeSend + 'static,
+ Message: 'static + MaybeSend,
+{
+ use futures::future::FutureExt;
+
+ run_with_id(
+ id,
+ futures::stream::unfold(initial, move |state| f(state).map(Some)),
+ )
+}
+
+/// Creates a [`Subscription`] that publishes the events sent from a [`Future`]
+/// to an [`mpsc::Sender`] with the given bounds.
///
/// # Creating an asynchronous worker with bidirectional communication
/// You can leverage this helper to create a [`Subscription`] that spawns
@@ -145,6 +168,7 @@ where
/// ```
/// use iced_native::subscription::{self, Subscription};
/// use iced_native::futures::channel::mpsc;
+/// use iced_native::futures::sink::SinkExt;
///
/// pub enum Event {
/// Ready(mpsc::Sender<Input>),
@@ -165,27 +189,35 @@ where
/// fn some_worker() -> Subscription<Event> {
/// struct SomeWorker;
///
-/// subscription::unfold(std::any::TypeId::of::<SomeWorker>(), State::Starting, |state| async move {
-/// match state {
-/// State::Starting => {
-/// // Create channel
-/// let (sender, receiver) = mpsc::channel(100);
+/// subscription::channel(std::any::TypeId::of::<SomeWorker>(), 100, |mut output| async move {
+/// let mut state = State::Starting;
///
-/// (Some(Event::Ready(sender)), State::Ready(receiver))
-/// }
-/// State::Ready(mut receiver) => {
-/// use iced_native::futures::StreamExt;
+/// loop {
+/// match &mut state {
+/// State::Starting => {
+/// // Create channel
+/// let (sender, receiver) = mpsc::channel(100);
+///
+/// // Send the sender back to the application
+/// output.send(Event::Ready(sender)).await;
+///
+/// // We are ready to receive messages
+/// state = State::Ready(receiver);
+/// }
+/// State::Ready(receiver) => {
+/// use iced_native::futures::StreamExt;
///
-/// // Read next input sent from `Application`
-/// let input = receiver.select_next_some().await;
+/// // Read next input sent from `Application`
+/// let input = receiver.select_next_some().await;
///
-/// match input {
-/// Input::DoSomeWork => {
-/// // Do some async work...
+/// match input {
+/// Input::DoSomeWork => {
+/// // Do some async work...
///
-/// // Finally, we can optionally return a message to tell the
-/// // `Application` the work is done
-/// (Some(Event::WorkFinished), State::Ready(receiver))
+/// // Finally, we can optionally produce a message to tell the
+/// // `Application` the work is done
+/// output.send(Event::WorkFinished).await;
+/// }
/// }
/// }
/// }
@@ -198,25 +230,28 @@ where
/// connection open.
///
/// [`websocket`]: https://github.com/iced-rs/iced/tree/0.8/examples/websocket
-pub fn unfold<I, T, Fut, Message>(
+pub fn channel<I, Fut, Message>(
id: I,
- initial: T,
- mut f: impl FnMut(T) -> Fut + MaybeSend + Sync + 'static,
+ size: usize,
+ f: impl Fn(mpsc::Sender<Message>) -> Fut + MaybeSend + Sync + 'static,
) -> Subscription<Message>
where
I: Hash + 'static,
- T: MaybeSend + 'static,
- Fut: Future<Output = (Option<Message>, T)> + MaybeSend + 'static,
+ Fut: Future<Output = Never> + MaybeSend + 'static,
Message: 'static + MaybeSend,
{
- use futures::future::{self, FutureExt};
- use futures::stream::StreamExt;
+ use futures::stream::{self, StreamExt};
- run_with_id(
+ Subscription::from_recipe(Runner {
id,
- futures::stream::unfold(initial, move |state| f(state).map(Some))
- .filter_map(future::ready),
- )
+ spawn: move |_| {
+ let (sender, receiver) = mpsc::channel(size);
+
+ let runner = stream::once(f(sender)).map(|_| unreachable!());
+
+ stream::select(receiver, runner)
+ },
+ })
}
struct Runner<I, F, S, Message>
generated by cgit (git 2.34.1) at 2025-04-16 10:53:56 +0000