Merge pull request #1349 from iced-rs/fix/subscription-cancelation

Fix `Subscription` cancelation when never awaiting
This commit is contained in:
Héctor Ramón 2022-05-30 17:57:29 -04:00 committed by GitHub
commit d26b3f0402
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23

View file

@ -1,6 +1,9 @@
use crate::{BoxFuture, MaybeSend, Subscription}; use crate::{BoxFuture, MaybeSend, Subscription};
use futures::{channel::mpsc, sink::Sink}; use futures::{
channel::mpsc,
sink::{Sink, SinkExt},
};
use std::{collections::HashMap, marker::PhantomData}; use std::{collections::HashMap, marker::PhantomData};
/// A registry of subscription streams. /// A registry of subscription streams.
@ -64,7 +67,7 @@ where
+ MaybeSend + MaybeSend
+ Clone, + Clone,
{ {
use futures::{future::FutureExt, stream::StreamExt}; use futures::stream::StreamExt;
let mut futures: Vec<BoxFuture<()>> = Vec::new(); let mut futures: Vec<BoxFuture<()>> = Vec::new();
@ -85,19 +88,29 @@ where
continue; continue;
} }
let (cancel, cancelled) = futures::channel::oneshot::channel(); let (cancel, mut canceled) = futures::channel::oneshot::channel();
// TODO: Use bus if/when it supports async // TODO: Use bus if/when it supports async
let (event_sender, event_receiver) = let (event_sender, event_receiver) =
futures::channel::mpsc::channel(100); futures::channel::mpsc::channel(100);
let stream = recipe.stream(event_receiver.boxed()); let mut receiver = receiver.clone();
let mut stream = recipe.stream(event_receiver.boxed());
let future = futures::future::select( let future = async move {
cancelled, loop {
stream.map(Ok).forward(receiver.clone()), let select =
) futures::future::select(&mut canceled, stream.next());
.map(|_| ());
match select.await {
futures::future::Either::Left(_)
| futures::future::Either::Right((None, _)) => break,
futures::future::Either::Right((Some(message), _)) => {
let _ = receiver.send(message).await;
}
}
}
};
let _ = self.subscriptions.insert( let _ = self.subscriptions.insert(
id, id,