Remove futures crate dependency

Cargo.toml

@@ -18,7 +18,6 @@ members = [
     "importer",
     "loader",
     "schema",
-    "client",
     "cli",
     "daemon",
     "core",
@@ -33,10 +32,11 @@ distill-importer = { version = "=0.0.3", path = "importer", optional = true }
 distill-loader = { version = "=0.0.3", path = "loader", optional = true }
 
 [dev-dependencies]
-futures = "0.3"
+futures-core = { version = "0.3", default-features = false, features = ["alloc"] }
+futures-io = { version = "0.3", default-features = false }
+futures-util = { version = "0.3", default-features = false }
 serde = "1"
 uuid = "0.8.2"
-tokio = { version = "1.2", features = ["sync"] }
 serial_test = "0.5.1"
 
 [features]

README.md

@@ -101,3 +101,8 @@ Licensed under either of
 at your option.
 
 PLEASE NOTE that some dependencies may be licensed under other terms. These are listed in [deny.toml](deny.toml) under licenses.exceptions on a best-effort basis, and are validated in every CI run using [cargo-deny](https://github.com/EmbarkStudios/cargo-deny).
+
+## Vendored Code
+
+In addition to crate dependencies, this project contains some vendored code:
+ * [daemon/src/timout.rs](daemon/src/timout.rs) - Used under Apache 2.0/MIT license. (Only used in unit tests)

cli/Cargo.toml

@@ -12,11 +12,11 @@ distill-schema = { version = "=0.0.3", path = "../schema" }
 
 capnp = "0.14.0"
 capnp-rpc = "0.14.0"
-tokio = { version = "1.2", features = ["io-std", "rt", "rt-multi-thread", "net", "io-util"] }
-tokio-util = { version = "0.6.1", features = ["codec", "compat"] }
-futures = { version = "0.3", default-features = false, features = ["std", "async-await"] }
+futures-util = { version = "0.3", default-features = false }
 uuid = "0.8.2"
 async-trait = "0.1.22"
 crossterm = { version = "0.17", features = ["event-stream"] }
 defer = "0.1.0"
-tokio-stream = { version = "0.1.2", features = ["io-util"] }
+async-io = "1.4.1"
+async-net = "1.6.0"
+async-executor = "1.4.1"

cli/src/lib.rs

@@ -9,6 +9,7 @@ use distill_schema::{
 };
 
 pub mod shell;
+use futures_util::AsyncReadExt;
 use shell::Autocomplete;
 pub use shell::Command;
 
@@ -52,16 +53,18 @@ pub struct Context {
     snapshot: Rc<RefCell<Snapshot>>,
 }
 
-pub async fn create_context() -> Result<Context, Box<dyn std::error::Error>> {
+pub async fn create_context(
+    local: &async_executor::LocalExecutor<'_>,
+) -> Result<Context, Box<dyn std::error::Error>> {
     use std::net::ToSocketAddrs;
     let addr = "127.0.0.1:9999".to_socket_addrs()?.next().unwrap();
-    let stream = tokio::net::TcpStream::connect(&addr).await?;
+    let stream = async_net::TcpStream::connect(&addr).await?;
     stream.set_nodelay(true).unwrap();
-    use tokio_util::compat::*;
-    let (reader, writer) = stream.into_split();
+
+    let (reader, writer) = stream.split();
     let rpc_network = Box::new(twoparty::VatNetwork::new(
-        reader.compat(),
-        writer.compat_write(),
+        reader,
+        writer,
         rpc_twoparty_capnp::Side::Client,
         *ReaderOptions::new()
             .nesting_limit(64)
@@ -72,14 +75,18 @@ pub async fn create_context() -> Result<Context, Box<dyn std::error::Error>> {
 
     let hub: asset_hub::Client = rpc_system.bootstrap(rpc_twoparty_capnp::Side::Server);
     let _disconnector = rpc_system.get_disconnector();
-    tokio::task::spawn_local(rpc_system);
+
+    local.spawn(rpc_system).detach();
+
     let snapshot = Rc::new(RefCell::new({
         let request = hub.get_snapshot_request();
         request.send().promise.await?.get()?.get_snapshot()?
     }));
+
     let listener: asset_hub::listener::Client = capnp_rpc::new_client(ListenerImpl {
         snapshot: snapshot.clone(),
     });
+
     let mut request = hub.register_listener_request();
     request.get().set_listener(listener);
 

cli/src/main.rs

@@ -1,13 +1,14 @@
 use distill_cli::{shell::Shell, *};
 
 pub fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let runtime = tokio::runtime::Runtime::new().unwrap();
-    let local = tokio::task::LocalSet::new();
-    runtime.block_on(local.run_until(async_main()))
+    let local = async_executor::LocalExecutor::new();
+    async_io::block_on(local.run(async_main(&local)))
 }
 
-async fn async_main() -> Result<(), Box<dyn std::error::Error>> {
-    let ctx = create_context().await?;
+async fn async_main(
+    local: &async_executor::LocalExecutor<'_>,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let ctx = create_context(local).await?;
 
     let mut shell = Shell::new(ctx);
 

cli/src/shell.rs

@@ -8,7 +8,7 @@ use crossterm::{
     style::{Color, Print, ResetColor, SetBackgroundColor, SetForegroundColor},
     terminal::{disable_raw_mode, enable_raw_mode, Clear, ClearType},
 };
-use futures::{
+use futures_util::{
     future::{pending, FusedFuture, FutureExt},
     select,
     stream::StreamExt,
@@ -425,8 +425,7 @@ impl<C> Shell<C> {
                                     break 'event_loop;
                                 }
                                 _ => {
-                                    let items =
-                                        std::mem::replace(&mut autocomplete.items, Vec::new());
+                                    let items = std::mem::take(&mut autocomplete.items);
                                     state = State::Select {
                                         overlap: autocomplete.overlap,
                                         items,

client/Cargo.toml

@@ -1,16 +0,0 @@
-[package]
-name = "distill-client"
-version = "0.0.1"
-authors = ["Karl Bergström <[email protected]>"]
-edition = "2018"
-license = "MIT OR Apache-2.0"
-publish = false
-
-[dependencies]
-distill-schema = { path = "../schema" }
-
-capnp = "0.14.0"
-capnp-rpc = "0.14.0"
-tokio = "1.2"
-tokio-util = { version = "0.6.1", features = ["compat"] }
-futures-util = { version = "0.3", default-features = false }

core/Cargo.toml

@@ -15,6 +15,7 @@ path_utils = ["dunce", "path-clean", "path-slash"]
 uuid = { version = "0.8.2", features = ["v4"] }
 serde = { version = "1", optional = true, features = ["derive"] }
 futures-core = { version = "0.3", default-features = false, features = ["alloc"] }
+futures-channel = { version = "0.3", default-features = false, features = ["std"] }
 type-uuid = { version = "0.1.2", optional = true, default-features = false }
 dunce = { version = "1.0", optional = true }
 path-clean = { version = "0.1", optional = true }

core/src/distill_signal.rs

@@ -0,0 +1,87 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// Thin wrapper around `futures_channel::oneshot` to match `tokio::sync::oneshot` interface.
+pub fn oneshot<T>() -> (Sender<T>, Receiver<T>) {
+    let (sender, receiver) = futures_channel::oneshot::channel();
+    (Sender::new(sender), Receiver::new(receiver))
+}
+
+#[derive(Debug)]
+pub struct Receiver<T> {
+    inner: futures_channel::oneshot::Receiver<T>,
+}
+
+impl<T> Receiver<T> {
+    #[inline(always)]
+    pub(crate) fn new(inner: futures_channel::oneshot::Receiver<T>) -> Self {
+        Receiver { inner }
+    }
+
+    #[inline]
+    pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
+        match self.inner.try_recv() {
+            Ok(Some(x)) => Ok(x),
+            Ok(None) => Err(TryRecvError::Empty),
+            Err(_canceled) => Err(TryRecvError::Closed),
+        }
+    }
+}
+
+impl<T> Future for Receiver<T> {
+    type Output = Result<T, RecvError>;
+
+    fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<T, RecvError>> {
+        match self.try_recv() {
+            Ok(value) => Poll::Ready(Ok(value)),
+            Err(TryRecvError::Closed) => Poll::Ready(Err(RecvError { 0: () })),
+            Err(TryRecvError::Empty) => Poll::Pending,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct Sender<T> {
+    inner: futures_channel::oneshot::Sender<T>,
+}
+
+impl<T> Sender<T> {
+    #[inline(always)]
+    pub(crate) fn new(inner: futures_channel::oneshot::Sender<T>) -> Self {
+        Sender { inner }
+    }
+
+    #[inline]
+    pub fn send(self, value: T) -> Result<(), RecvError> {
+        match self.inner.send(value) {
+            Ok(_) => Ok(()),
+            Err(_) => Err(RecvError { 0: () }),
+        }
+    }
+}
+
+use self::error::*;
+pub mod error {
+    use std::fmt;
+
+    #[derive(Debug, Eq, PartialEq)]
+    pub struct RecvError(pub(super) ());
+
+    #[derive(Debug, Eq, PartialEq)]
+    pub enum TryRecvError {
+        Empty,
+        Closed,
+    }
+
+    impl fmt::Display for TryRecvError {
+        fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+            match self {
+                TryRecvError::Empty => write!(fmt, "channel empty"),
+                TryRecvError::Closed => write!(fmt, "channel closed"),
+            }
+        }
+    }
+
+    impl std::error::Error for TryRecvError {}
+}

core/src/lib.rs

@@ -13,6 +13,8 @@ use uuid::Uuid;
 pub mod importer_context;
 pub mod utils;
 
+pub mod distill_signal;
+
 /// A universally unique identifier for an asset.
 /// An asset can be a value of any Rust type that implements
 /// [`TypeUuidDynamic`] + [serde::Serialize] + [Send].