Create `iced_futures` and wire everything up

23 files changed, 196 insertions, 195 deletions

diff --git a/Cargo.toml b/Cargo.toml
index aeb8382e..fbe3b9f2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -23,6 +23,7 @@ maintenance = { status = "actively-developed" }
 [workspace]
 members = [
     "core",
+    "futures",
     "native",
     "style",
     "web",
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 5e1a5532..22bc7ceb 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -7,14 +7,4 @@ description = "The essential concepts of Iced"
 license = "MIT"
 repository = "https://github.com/hecrj/iced"
 
-[features]
-# Exposes a future-based `Command` type
-command = ["futures"]
-# Exposes a future-based `Subscription` type
-subscription = ["futures", "log"]
-# Exposes a `runtime` module meant to abstract over different future executors
-runtime = ["command", "subscription"]
-
 [dependencies]
-futures = { version = "0.3", optional = true }
-log = { version = "0.4", optional = true }
diff --git a/core/src/lib.rs b/core/src/lib.rs
index 760acefe..bec307ad 100644
--- a/core/src/lib.rs
+++ b/core/src/lib.rs
@@ -32,21 +32,3 @@ pub use length::Length;
 pub use point::Point;
 pub use rectangle::Rectangle;
 pub use vector::Vector;
-
-#[cfg(feature = "command")]
-mod command;
-
-#[cfg(feature = "command")]
-pub use command::Command;
-
-#[cfg(feature = "subscription")]
-pub mod subscription;
-
-#[cfg(feature = "subscription")]
-pub use subscription::Subscription;
-
-#[cfg(feature = "runtime")]
-mod runtime;
-
-#[cfg(feature = "runtime")]
-pub use runtime::Runtime;
diff --git a/core/src/runtime/executor.rs b/core/src/runtime/executor.rs
deleted file mode 100644
index d171c6d5..00000000
--- a/core/src/runtime/executor.rs
+++ /dev/null
@@ -1,11 +0,0 @@
-use futures::Future;
-
-pub trait Executor {
-    fn new() -> Self;
-
-    fn spawn(&self, future: impl Future<Output = ()> + Send + 'static);
-
-    fn enter<R>(&self, f: impl FnOnce() -> R) -> R {
-        f()
-    }
-}
diff --git a/futures/Cargo.toml b/futures/Cargo.toml
new file mode 100644
index 00000000..fe0d378c
--- /dev/null
+++ b/futures/Cargo.toml
@@ -0,0 +1,23 @@
+[package]
+name = "iced_futures"
+version = "0.1.0-alpha"
+authors = ["Héctor Ramón Jiménez <hector0193@gmail.com>"]
+edition = "2018"
+description = "Commands, subscriptions, and runtimes for Iced"
+license = "MIT"
+repository = "https://github.com/hecrj/iced"
+documentation = "https://docs.rs/iced_futures"
+keywords = ["gui", "ui", "graphics", "interface", "futures"]
+categories = ["gui"]
+
+[dependencies]
+log = "0.4"
+
+[dependencies.futures]
+version = "0.3"
+features = ["thread-pool"]
+
+[dependencies.tokio]
+version = "0.2"
+optional = true
+features = ["rt-core"]
diff --git a/core/src/command.rs b/futures/src/command.rs
index e7885fb8..e7885fb8 100644
--- a/core/src/command.rs
+++ b/futures/src/command.rs
diff --git a/futures/src/lib.rs b/futures/src/lib.rs
new file mode 100644
index 00000000..f6bcf85a
--- /dev/null
+++ b/futures/src/lib.rs
@@ -0,0 +1,8 @@
+mod command;
+
+pub mod runtime;
+pub mod subscription;
+
+pub use command::Command;
+pub use runtime::Runtime;
+pub use subscription::Subscription;
diff --git a/core/src/runtime.rs b/futures/src/runtime.rs
index 31234d11..bc1ad8ac 100644
--- a/core/src/runtime.rs
+++ b/futures/src/runtime.rs
@@ -1,3 +1,4 @@
+//! Run commands and subscriptions.
 mod executor;
 
 pub use executor::Executor;
@@ -10,8 +11,8 @@ use std::marker::PhantomData;
 #[derive(Debug)]
 pub struct Runtime<Hasher, Event, Executor, Receiver, Message> {
     executor: Executor,
-    subscriptions: subscription::Tracker<Hasher, Event>,
     receiver: Receiver,
+    subscriptions: subscription::Tracker<Hasher, Event>,
     _message: PhantomData<Message>,
 }
 
@@ -28,15 +29,19 @@ where
         + 'static,
     Message: Send + 'static,
 {
-    pub fn new(receiver: Receiver) -> Self {
+    pub fn new(executor: Executor, receiver: Receiver) -> Self {
         Self {
-            executor: Executor::new(),
-            subscriptions: subscription::Tracker::new(),
+            executor,
             receiver,
+            subscriptions: subscription::Tracker::new(),
             _message: PhantomData,
         }
     }
 
+    pub fn enter<R>(&self, f: impl FnOnce() -> R) -> R {
+        self.executor.enter(f)
+    }
+
     pub fn spawn(&mut self, command: Command<Message>) {
         use futures::{FutureExt, SinkExt};
 
diff --git a/futures/src/runtime/executor.rs b/futures/src/runtime/executor.rs
new file mode 100644
index 00000000..855aa105
--- /dev/null
+++ b/futures/src/runtime/executor.rs
@@ -0,0 +1,26 @@
+use futures::Future;
+
+pub trait Executor {
+    fn spawn(&self, future: impl Future<Output = ()> + Send + 'static);
+
+    fn enter<R>(&self, f: impl FnOnce() -> R) -> R {
+        f()
+    }
+}
+
+impl Executor for futures::executor::ThreadPool {
+    fn spawn(&self, future: impl Future<Output = ()> + Send + 'static) {
+        self.spawn_ok(future);
+    }
+}
+
+#[cfg(feature = "tokio")]
+impl Executor for tokio::runtime::Runtime {
+    fn spawn(&self, future: impl Future<Output = ()> + Send + 'static) {
+        let _ = tokio::runtime::Runtime::spawn(self, future);
+    }
+
+    fn enter<R>(&self, f: impl FnOnce() -> R) -> R {
+        tokio::runtime::Runtime::enter(self, f)
+    }
+}
diff --git a/core/src/subscription.rs b/futures/src/subscription.rs
index 87e51e48..87e51e48 100644
--- a/core/src/subscription.rs
+++ b/futures/src/subscription.rs
diff --git a/core/src/subscription/tracker.rs b/futures/src/subscription/tracker.rs
index a942b619..a942b619 100644
--- a/core/src/subscription/tracker.rs
+++ b/futures/src/subscription/tracker.rs
diff --git a/native/Cargo.toml b/native/Cargo.toml
index a31b6627..57a869e2 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -8,8 +8,15 @@ license = "MIT"
 repository = "https://github.com/hecrj/iced"
 
 [dependencies]
-iced_core = { version = "0.1.0", path = "../core", features = ["command", "subscription"] }
 twox-hash = "1.5"
 raw-window-handle = "0.3"
 unicode-segmentation = "1.6"
 futures = "0.3"
+
+[dependencies.iced_core]
+version = "0.1.0"
+path = "../core"
+
+[dependencies.iced_futures]
+version = "0.1.0-alpha"
+path = "../futures"
diff --git a/native/src/lib.rs b/native/src/lib.rs
index 340b9ea7..7730c6a3 100644
--- a/native/src/lib.rs
+++ b/native/src/lib.rs
@@ -42,6 +42,7 @@
 pub mod input;
 pub mod layout;
 pub mod renderer;
+pub mod runtime;
 pub mod subscription;
 pub mod widget;
 pub mod window;
@@ -55,9 +56,10 @@ mod size;
 mod user_interface;
 
 pub use iced_core::{
-    Align, Background, Color, Command, Font, HorizontalAlignment, Length,
-    Point, Rectangle, Vector, VerticalAlignment,
+    Align, Background, Color, Font, HorizontalAlignment, Length, Point,
+    Rectangle, Vector, VerticalAlignment,
 };
+pub use iced_futures::Command;
 
 pub use clipboard::Clipboard;
 pub use element::Element;
@@ -66,6 +68,7 @@ pub use hasher::Hasher;
 pub use layout::Layout;
 pub use mouse_cursor::MouseCursor;
 pub use renderer::Renderer;
+pub use runtime::Runtime;
 pub use size::Size;
 pub use subscription::Subscription;
 pub use user_interface::{Cache, UserInterface};
diff --git a/native/src/runtime.rs b/native/src/runtime.rs
new file mode 100644
index 00000000..2b3abbf1
--- /dev/null
+++ b/native/src/runtime.rs
@@ -0,0 +1,14 @@
+//! Run commands and subscriptions.
+use crate::{Event, Hasher};
+
+/// A native runtime with a generic executor and receiver of results.
+///
+/// It can be used by shells to easily spawn a [`Command`] or track a
+/// [`Subscription`].
+///
+/// [`Command`]: ../struct.Command.html
+/// [`Subscription`]: ../struct.Subscription.html
+pub type Runtime<Executor, Receiver, Message> =
+    iced_futures::Runtime<Hasher, Event, Executor, Receiver, Message>;
+
+pub use iced_futures::runtime::Executor;
diff --git a/native/src/subscription.rs b/native/src/subscription.rs
index cd0822c1..43f1758a 100644
--- a/native/src/subscription.rs
+++ b/native/src/subscription.rs
@@ -15,7 +15,7 @@ use futures::stream::BoxStream;
 ///
 /// [`Command`]: ../struct.Command.html
 /// [`Subscription`]: struct.Subscription.html
-pub type Subscription<T> = iced_core::Subscription<Hasher, Event, T>;
+pub type Subscription<T> = iced_futures::Subscription<Hasher, Event, T>;
 
 /// A stream of runtime events.
 ///
@@ -27,9 +27,9 @@ pub type EventStream = BoxStream<'static, Event>;
 /// A native [`Subscription`] tracker.
 ///
 /// [`Subscription`]: type.Subscription.html
-pub type Tracker = iced_core::subscription::Tracker<Hasher, Event>;
+pub type Tracker = iced_futures::subscription::Tracker<Hasher, Event>;
 
-pub use iced_core::subscription::Recipe;
+pub use iced_futures::subscription::Recipe;
 
 mod events;
 
diff --git a/web/Cargo.toml b/web/Cargo.toml
index 605c7462..ea092575 100644
--- a/web/Cargo.toml
+++ b/web/Cargo.toml
@@ -15,12 +15,19 @@ categories = ["web-programming"]
 maintenance = { status = "actively-developed" }
 
 [dependencies]
-iced_core = { version = "0.1.0", path = "../core", features = ["command", "subscription"] }
 dodrio = "0.1.0"
 wasm-bindgen = "0.2.51"
 wasm-bindgen-futures = "0.4"
 futures = "0.3"
 
+[dependencies.iced_core]
+version = "0.1.0"
+path = "../core"
+
+[dependencies.iced_futures]
+version = "0.1.0-alpha"
+path = "../futures"
+
 [dependencies.web-sys]
 version = "0.3.27"
 features = [
diff --git a/web/src/lib.rs b/web/src/lib.rs
index 7ea22e85..b183c390 100644
--- a/web/src/lib.rs
+++ b/web/src/lib.rs
@@ -72,9 +72,10 @@ pub use dodrio;
 pub use element::Element;
 pub use hasher::Hasher;
 pub use iced_core::{
-    Align, Background, Color, Command, Font, HorizontalAlignment, Length,
+    Align, Background, Color, Font, HorizontalAlignment, Length,
     VerticalAlignment,
 };
+pub use iced_futures::Command;
 pub use style::Style;
 pub use subscription::Subscription;
 pub use widget::*;
@@ -148,7 +149,6 @@ pub trait Application {
     }
 }
 
-
 struct Instance<Message> {
     title: String,
     ui: Rc<RefCell<Box<dyn Application<Message = Message>>>>,
@@ -167,7 +167,7 @@ impl<Message> Clone for Instance<Message> {
 
 impl<Message> Instance<Message>
 where
-    Message: 'static
+    Message: 'static,
 {
     fn new(ui: impl Application<Message = Message> + 'static) -> Self {
         Self {
diff --git a/web/src/subscription.rs b/web/src/subscription.rs
index 4638c8ab..6b8415c0 100644
--- a/web/src/subscription.rs
+++ b/web/src/subscription.rs
@@ -14,6 +14,6 @@ use crate::Hasher;
 ///
 /// [`Command`]: ../struct.Command.html
 /// [`Subscription`]: struct.Subscription.html
-pub type Subscription<T> = iced_core::Subscription<Hasher, (), T>;
+pub type Subscription<T> = iced_futures::Subscription<Hasher, (), T>;
 
-pub use iced_core::subscription::Recipe;
+pub use iced_futures::subscription::Recipe;
diff --git a/winit/Cargo.toml b/winit/Cargo.toml
index 5727f8cf..3ed37dd5 100644
--- a/winit/Cargo.toml
+++ b/winit/Cargo.toml
@@ -14,11 +14,17 @@ categories = ["gui"]
 debug = []
 
 [dependencies]
-iced_native = { version = "0.1.0-alpha", path = "../native" }
 winit = { version = "0.20.0-alpha3", git = "https://github.com/hecrj/winit", rev = "709808eb4e69044705fcb214bcc30556db761405"}
-window_clipboard = { git = "https://github.com/hecrj/window_clipboard", rev = "22c6dd6c04cd05d528029b50a30c56417cd4bebf" }
-futures = { version = "0.3", features = ["thread-pool"] }
 log = "0.4"
+futures = { version = "0.3", features = ["thread-pool"] }
+
+[dependencies.iced_native]
+version = "0.1.0-alpha"
+path = "../native"
+
+[dependencies.window_clipboard]
+git = "https://github.com/hecrj/window_clipboard"
+rev = "22c6dd6c04cd05d528029b50a30c56417cd4bebf"
 
 [target.'cfg(target_os = "windows")'.dependencies.winapi]
 version = "0.3.6"
diff --git a/winit/src/application.rs b/winit/src/application.rs
index a14924ac..076ac092 100644
--- a/winit/src/application.rs
+++ b/winit/src/application.rs
@@ -1,9 +1,10 @@
 use crate::{
     conversion,
     input::{keyboard, mouse},
-    subscription, window, Cache, Clipboard, Command, Debug, Element, Event,
-    Mode, MouseCursor, Settings, Size, Subscription, UserInterface,
+    window, Cache, Clipboard, Command, Debug, Element, Event, Mode,
+    MouseCursor, Proxy, Settings, Size, Subscription, UserInterface,
 };
+use iced_native::Runtime;
 
 /// An interactive, native cross-platform application.
 ///
@@ -109,17 +110,19 @@ pub trait Application: Sized {
 
         debug.startup_started();
         let event_loop = EventLoop::with_user_event();
-        let proxy = event_loop.create_proxy();
-        let mut thread_pool =
-            futures::executor::ThreadPool::new().expect("Create thread pool");
-        let mut subscription_pool = subscription::Pool::new();
+        let mut runtime = {
+            let thread_pool = futures::executor::ThreadPool::new()
+                .expect("Create thread pool");
+
+            Runtime::new(thread_pool, Proxy::new(event_loop.create_proxy()))
+        };
         let mut external_messages = Vec::new();
 
         let (mut application, init_command) = Self::new();
-        spawn(init_command, &mut thread_pool, &proxy);
+        runtime.spawn(init_command);
 
         let subscription = application.subscription();
-        subscription_pool.update(subscription, &mut thread_pool, &proxy);
+        runtime.track(subscription);
 
         let mut title = application.title();
         let mut mode = application.mode();
@@ -212,7 +215,7 @@ pub trait Application: Sized {
                 events
                     .iter()
                     .cloned()
-                    .for_each(|event| subscription_pool.broadcast_event(event));
+                    .for_each(|event| runtime.broadcast(event));
 
                 let mut messages = user_interface.update(
                     &renderer,
@@ -241,17 +244,15 @@ pub trait Application: Sized {
                         debug.log_message(&message);
 
                         debug.update_started();
-                        let command = application.update(message);
-                        spawn(command, &mut thread_pool, &proxy);
+                        let command =
+                            runtime.enter(|| application.update(message));
+                        runtime.spawn(command);
                         debug.update_finished();
                     }
 
-                    let subscription = application.subscription();
-                    subscription_pool.update(
-                        subscription,
-                        &mut thread_pool,
-                        &proxy,
-                    );
+                    let subscription =
+                        runtime.enter(|| application.subscription());
+                    runtime.track(subscription);
 
                     // Update window title
                     let new_title = application.title();
@@ -463,28 +464,6 @@ fn to_physical(size: winit::dpi::LogicalSize, dpi: f64) -> (u16, u16) {
     )
 }
 
-fn spawn<Message: Send>(
-    command: Command<Message>,
-    thread_pool: &mut futures::executor::ThreadPool,
-    proxy: &winit::event_loop::EventLoopProxy<Message>,
-) {
-    use futures::FutureExt;
-
-    let futures = command.futures();
-
-    for future in futures {
-        let proxy = proxy.clone();
-
-        let future = future.map(move |message| {
-            proxy
-                .send_event(message)
-                .expect("Send command result to event loop");
-        });
-
-        thread_pool.spawn_ok(future);
-    }
-}
-
 // As defined in: http://www.unicode.org/faq/private_use.html
 // TODO: Remove once https://github.com/rust-windowing/winit/pull/1254 lands
 fn is_private_use_character(c: char) -> bool {
diff --git a/winit/src/lib.rs b/winit/src/lib.rs
index 9000f977..056ae8f0 100644
--- a/winit/src/lib.rs
+++ b/winit/src/lib.rs
@@ -31,7 +31,7 @@ pub mod settings;
 mod application;
 mod clipboard;
 mod mode;
-mod subscription;
+mod proxy;
 
 // We disable debug capabilities on release builds unless the `debug` feature
 // is explicitly enabled.
@@ -48,3 +48,4 @@ pub use settings::Settings;
 
 use clipboard::Clipboard;
 use debug::Debug;
+use proxy::Proxy;
diff --git a/winit/src/proxy.rs b/winit/src/proxy.rs
new file mode 100644
index 00000000..7e8dee98
--- /dev/null
+++ b/winit/src/proxy.rs
@@ -0,0 +1,57 @@
+use futures::{
+    task::{Context, Poll},
+    Sink,
+};
+use std::pin::Pin;
+
+pub struct Proxy<Message: 'static> {
+    raw: winit::event_loop::EventLoopProxy<Message>,
+}
+
+impl<Message: 'static> Clone for Proxy<Message> {
+    fn clone(&self) -> Self {
+        Self {
+            raw: self.raw.clone(),
+        }
+    }
+}
+
+impl<Message: 'static> Proxy<Message> {
+    pub fn new(raw: winit::event_loop::EventLoopProxy<Message>) -> Self {
+        Self { raw }
+    }
+}
+
+impl<Message: 'static> Sink<Message> for Proxy<Message> {
+    type Error = core::convert::Infallible;
+
+    fn poll_ready(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn start_send(
+        self: Pin<&mut Self>,
+        message: Message,
+    ) -> Result<(), Self::Error> {
+        let _ = self.raw.send_event(message);
+
+        Ok(())
+    }
+
+    fn poll_flush(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_close(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+}
diff --git a/winit/src/subscription.rs b/winit/src/subscription.rs
deleted file mode 100644
index bad68d55..00000000
--- a/winit/src/subscription.rs
+++ /dev/null
@@ -1,97 +0,0 @@
-use iced_native::{Event, Hasher, Subscription};
-use std::collections::HashMap;
-
-pub struct Pool {
-    alive: HashMap<u64, Handle>,
-}
-
-pub struct Handle {
-    _cancel: futures::channel::oneshot::Sender<()>,
-    listener: Option<futures::channel::mpsc::Sender<Event>>,
-}
-
-impl Pool {
-    pub fn new() -> Self {
-        Self {
-            alive: HashMap::new(),
-        }
-    }
-
-    pub fn update<Message: Send>(
-        &mut self,
-        subscription: Subscription<Message>,
-        thread_pool: &mut futures::executor::ThreadPool,
-        proxy: &winit::event_loop::EventLoopProxy<Message>,
-    ) {
-        use futures::{future::FutureExt, stream::StreamExt};
-
-        let recipes = subscription.recipes();
-        let mut alive = std::collections::HashSet::new();
-
-        for recipe in recipes {
-            let id = {
-                use std::hash::Hasher as _;
-
-                let mut hasher = Hasher::default();
-                recipe.hash(&mut hasher);
-
-                hasher.finish()
-            };
-
-            let _ = alive.insert(id);
-
-            if !self.alive.contains_key(&id) {
-                let (cancel, cancelled) = futures::channel::oneshot::channel();
-
-                // TODO: Use bus if/when it supports async
-                let (event_sender, event_receiver) =
-                    futures::channel::mpsc::channel(100);
-
-                let stream = recipe.stream(event_receiver.boxed());
-                let proxy = proxy.clone();
-
-                let future = futures::future::select(
-                    cancelled,
-                    stream.for_each(move |message| {
-                        proxy
-                            .send_event(message)
-                            .expect("Send subscription result to event loop");
-
-                        futures::future::ready(())
-                    }),
-                )
-                .map(|_| ());
-
-                thread_pool.spawn_ok(future);
-
-                let _ = self.alive.insert(
-                    id,
-                    Handle {
-                        _cancel: cancel,
-                        listener: if event_sender.is_closed() {
-                            None
-                        } else {
-                            Some(event_sender)
-                        },
-                    },
-                );
-            }
-        }
-
-        self.alive.retain(|id, _| alive.contains(&id));
-    }
-
-    pub fn broadcast_event(&mut self, event: Event) {
-        self.alive
-            .values_mut()
-            .filter_map(|connection| connection.listener.as_mut())
-            .for_each(|listener| {
-                if let Err(error) = listener.try_send(event.clone()) {
-                    log::error!(
-                        "Error sending event to subscription: {:?}",
-                        error
-                    );
-                }
-            });
-    }
-}