Skip to content

Commit

Permalink
Merge pull request #1786 from iced-rs/feature/subscription-channel
Browse files Browse the repository at this point in the history 
`channel` helper for `subscription`
  • Loading branch information
@hecrj
hecrj authored Apr 11, 2023
2 parents ff24f90 + 0ed5434 commit ca828f0
Show file tree
Hide file tree
Showing 3 changed files with 122 additions and 86 deletions.
17 changes: 7 additions & 10 deletions examples/download_progress/src/download.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -30,18 +27,18 @@ 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,
downloaded: 0,
},
)
} 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 {
Expand All @@ -55,16 +52,16 @@ 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,
downloaded,
},
)
}
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
Expand Down
100 changes: 52 additions & 48 deletions examples/websocket/src/echo.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
}
}
}
Expand Down
91 changes: 63 additions & 28 deletions native/src/subscription.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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};

Expand Down Expand Up @@ -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
Expand All @@ -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>),
Expand All @@ -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;
/// }
/// }
/// }
/// }
Expand All @@ -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>
Expand Down

0 comments on commit ca828f0

Please sign in to comment.