From 309c81d05e8973d1d74e9574e551f11d8bb63097 Mon Sep 17 00:00:00 2001 From: Philip Degarmo Date: Sat, 24 Jul 2021 14:06:46 -0700 Subject: [PATCH 1/2] Switch core to use futures_channel instead of futures --- core/Cargo.toml | 2 +- core/src/distill_signal.rs | 12 ++++++------ 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/core/Cargo.toml b/core/Cargo.toml index 8bca43c3..1bedf631 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -15,7 +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 = { version = "0.3", default-features = false, features = ["std", "async-await"] } +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 } diff --git a/core/src/distill_signal.rs b/core/src/distill_signal.rs index e0907465..a893b0e8 100644 --- a/core/src/distill_signal.rs +++ b/core/src/distill_signal.rs @@ -2,20 +2,20 @@ 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. +/// Thin wrapper around `futures_channel::oneshot` to match `tokio::sync::oneshot` interface. pub fn oneshot() -> (Sender, Receiver) { - let (sender, receiver) = futures::channel::oneshot::channel(); + let (sender, receiver) = futures_channel::oneshot::channel(); (Sender::new(sender), Receiver::new(receiver)) } #[derive(Debug)] pub struct Receiver { - inner: futures::channel::oneshot::Receiver, + inner: futures_channel::oneshot::Receiver, } impl Receiver { #[inline(always)] - pub(crate) fn new(inner: futures::channel::oneshot::Receiver) -> Self { + pub(crate) fn new(inner: futures_channel::oneshot::Receiver) -> Self { Receiver { inner } } @@ -43,12 +43,12 @@ impl Future for Receiver { #[derive(Debug)] pub struct Sender { - inner: futures::channel::oneshot::Sender, + inner: futures_channel::oneshot::Sender, } impl Sender { #[inline(always)] - pub(crate) fn new(inner: futures::channel::oneshot::Sender) -> Self { + pub(crate) fn new(inner: futures_channel::oneshot::Sender) -> Self { Sender { inner } } From ae274ea3130dcf937e0f9f19e95638c09ca199c7 Mon Sep 17 00:00:00 2001 From: Philip Degarmo Date: Sat, 24 Jul 2021 14:54:58 -0700 Subject: [PATCH 2/2] Remove futures --- Cargo.toml | 4 +++- cli/Cargo.toml | 2 +- cli/src/lib.rs | 2 +- cli/src/shell.rs | 2 +- daemon/Cargo.toml | 5 ++++- daemon/src/asset_hub_service.rs | 2 +- daemon/src/daemon.rs | 6 +++--- daemon/src/file_asset_source.rs | 3 ++- daemon/src/file_tracker.rs | 12 ++++++------ daemon/src/source_pair_import.rs | 7 ++++--- daemon/src/watcher.rs | 2 +- importer/Cargo.toml | 4 +++- importer/src/boxed_importer.rs | 3 ++- importer/src/lib.rs | 4 +++- src/lib.rs | 4 +++- 15 files changed, 38 insertions(+), 24 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 95203d1e..34a8221c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,7 +32,9 @@ 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" serial_test = "0.5.1" diff --git a/cli/Cargo.toml b/cli/Cargo.toml index e615ddba..9000db18 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -12,7 +12,7 @@ distill-schema = { version = "=0.0.3", path = "../schema" } capnp = "0.14.0" capnp-rpc = "0.14.0" -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"] } diff --git a/cli/src/lib.rs b/cli/src/lib.rs index 15978604..267ccb4b 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -9,7 +9,7 @@ use distill_schema::{ }; pub mod shell; -use futures::AsyncReadExt; +use futures_util::AsyncReadExt; use shell::Autocomplete; pub use shell::Command; diff --git a/cli/src/shell.rs b/cli/src/shell.rs index 3ed310ef..52d379ec 100644 --- a/cli/src/shell.rs +++ b/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, diff --git a/daemon/Cargo.toml b/daemon/Cargo.toml index d4741f9a..d722c24c 100644 --- a/daemon/Cargo.toml +++ b/daemon/Cargo.toml @@ -20,7 +20,10 @@ notify = "4.0.15" distill-downstream-lmdb-rkv = "0.11.0-windows-fix" rayon = { version = "1.3", optional = true } log = { version = "0.4", features = ["serde", "std"] } -futures = { version = "0.3", default-features = false, features = ["std", "async-await"] } +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, features = ["std", "async-await", "async-await-macro"] } +futures-channel = { version = "0.3", default-features = false, features = ["std"] } event-listener = { version = "2.4.0" } serde = "1" serde_derive = "1.0" diff --git a/daemon/src/asset_hub_service.rs b/daemon/src/asset_hub_service.rs index 1305fda9..0f4579bb 100644 --- a/daemon/src/asset_hub_service.rs +++ b/daemon/src/asset_hub_service.rs @@ -18,7 +18,7 @@ use distill_schema::{ parse_artifact_metadata, parse_db_asset_ref, service::asset_hub, }; -use futures::{AsyncReadExt, TryFutureExt}; +use futures_util::{AsyncReadExt, TryFutureExt}; use crate::{ artifact_cache::ArtifactCache, diff --git a/daemon/src/daemon.rs b/daemon/src/daemon.rs index ec02db57..c5f63b99 100644 --- a/daemon/src/daemon.rs +++ b/daemon/src/daemon.rs @@ -14,7 +14,7 @@ use distill_core::distill_signal; use distill_importer::{BoxedImporter, ImporterContext}; use distill_schema::data; use file_asset_source::FileAssetSource; -use futures::future::FutureExt; +use futures_util::future::FutureExt; use std::rc::Rc; use crate::{ @@ -252,11 +252,11 @@ impl AssetDaemon { let rx_fuse = rx.fuse(); - futures::pin_mut!(service_handle, tracker_handle, asset_source_handle, rx_fuse); + futures_util::pin_mut!(service_handle, tracker_handle, asset_source_handle, rx_fuse); log::info!("Starting Daemon Loop"); loop { - futures::select! { + futures_util::select! { _done = &mut service_handle => panic!("ServiceHandle panicked"), _done = &mut tracker_handle => panic!("FileTracker panicked"), _done = &mut asset_source_handle => panic!("AssetSource panicked"), diff --git a/daemon/src/file_asset_source.rs b/daemon/src/file_asset_source.rs index e03fd129..8edb94c6 100644 --- a/daemon/src/file_asset_source.rs +++ b/daemon/src/file_asset_source.rs @@ -19,7 +19,8 @@ use distill_schema::{ data::{self, path_refs, source_metadata}, parse_db_metadata, }; -use futures::{channel::mpsc::unbounded, lock::Mutex, stream::StreamExt}; +use futures_channel::mpsc::unbounded; +use futures_util::{lock::Mutex, stream::StreamExt}; use log::{debug, error, info}; #[cfg(feature = "rayon")] use rayon::prelude::*; diff --git a/daemon/src/file_tracker.rs b/daemon/src/file_tracker.rs index 570570ec..ce50ef7d 100644 --- a/daemon/src/file_tracker.rs +++ b/daemon/src/file_tracker.rs @@ -17,8 +17,8 @@ use std::{ use distill_core::utils::{self, canonicalize_path}; use distill_schema::data::{self, dirty_file_info, rename_file_event, source_file_info, FileType}; use event_listener::Event; -use futures::{ - channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender}, +use futures_channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender}; +use futures_util::{ lock::Mutex, stream::StreamExt, FutureExt, @@ -705,17 +705,17 @@ impl FileTracker { let listener_rx = listener_rx_guard.get_mut().fuse(); let watcher_rx = watcher_rx.fuse(); - futures::pin_mut!(watcher_rx, listener_rx, stopping); + futures_util::pin_mut!(watcher_rx, listener_rx, stopping); let mut dirty = false; #[allow(unused_mut)] loop { - let delay = futures::FutureExt::fuse(async_io::Timer::after(Duration::from_millis(40))); + let delay = futures_util::future::FutureExt::fuse(async_io::Timer::after(Duration::from_millis(40))); - futures::pin_mut!(delay); + futures_util::pin_mut!(delay); - futures::select! { + futures_util::select! { new_listener = listener_rx.next() => listeners.register(new_listener), _ = delay => { if dirty { diff --git a/daemon/src/source_pair_import.rs b/daemon/src/source_pair_import.rs index 1190101b..d823699e 100644 --- a/daemon/src/source_pair_import.rs +++ b/daemon/src/source_pair_import.rs @@ -14,7 +14,8 @@ use distill_importer::{ SourceMetadata as ImporterSourceMetadata, SOURCEMETADATA_VERSION, }; use distill_schema::data; -use futures::future::{BoxFuture, Future}; +use futures_core::future::BoxFuture; +use std::future::Future; use log::{debug, error}; use serde::{Deserialize, Serialize}; @@ -25,7 +26,7 @@ use crate::{ file_tracker::FileState, watcher::file_metadata, }; -use futures::AsyncReadExt; +use futures_util::AsyncReadExt; pub type SourceMetadata = ImporterSourceMetadata, Box>; @@ -596,7 +597,7 @@ impl<'a> SourcePairImport<'a> { let mut f = std::fs::File::open(source)?; let mut contents = vec![]; f.read_to_end(&mut contents)?; - let mut cursor = futures::io::Cursor::new(contents); + let mut cursor = futures_util::io::Cursor::new(contents); importer .import_boxed( diff --git a/daemon/src/watcher.rs b/daemon/src/watcher.rs index 40505617..79d17cd9 100644 --- a/daemon/src/watcher.rs +++ b/daemon/src/watcher.rs @@ -10,7 +10,7 @@ use distill_core::utils::canonicalize_path; use notify::{watcher, DebouncedEvent, RecommendedWatcher, RecursiveMode, Watcher}; use crate::error::{Error, Result}; -use futures::channel::mpsc::UnboundedSender; +use futures_channel::mpsc::UnboundedSender; /// The purpose of DirWatcher is to provide enough information to /// determine which files may be candidates for going through the asset import process. diff --git a/importer/Cargo.toml b/importer/Cargo.toml index cc8e0a01..7fbbb084 100644 --- a/importer/Cargo.toml +++ b/importer/Cargo.toml @@ -15,7 +15,9 @@ serde = "1" erased-serde = "0.3" ron = { version = "0.6.4", optional = true } typetag = { version = "0.1", optional = true } -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 } log = { version = "0.4", features = ["serde"] } [features] diff --git a/importer/src/boxed_importer.rs b/importer/src/boxed_importer.rs index 05f66575..d9f25c55 100644 --- a/importer/src/boxed_importer.rs +++ b/importer/src/boxed_importer.rs @@ -1,6 +1,7 @@ use distill_core::TypeUuidDynamic; use erased_serde::Deserializer; -use futures::{future::BoxFuture, AsyncRead, AsyncWrite}; +use futures_core::future::BoxFuture; +use futures_io::{AsyncRead, AsyncWrite}; use serde::{Deserialize, Serialize}; use crate::{error::Result, AsyncImporter, ExportAsset, ImportOp, ImporterValue, SerdeObj}; diff --git a/importer/src/lib.rs b/importer/src/lib.rs index 06943a7c..b65f192f 100644 --- a/importer/src/lib.rs +++ b/importer/src/lib.rs @@ -14,7 +14,9 @@ pub use distill_core::{ use distill_core::{AssetRef, AssetUuid}; #[cfg(feature = "serde_importers")] pub use distill_serde_importable_derive::*; -use futures::{future::BoxFuture, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use futures_core::future::BoxFuture; +use futures_io::{AsyncRead, AsyncWrite}; +use futures_util::{AsyncReadExt, AsyncWriteExt}; pub use serde; use serde::Serialize; diff --git a/src/lib.rs b/src/lib.rs index 52722c6f..5bae48a8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -56,7 +56,9 @@ mod tests { }, LoadHandle, Loader, }; - use futures::{future::BoxFuture, io::AsyncReadExt, AsyncRead}; + use futures_core::future::BoxFuture; + use futures_io::AsyncRead; + use futures_util::io::AsyncReadExt; use serde::{Deserialize, Serialize}; use serial_test::serial; use uuid::Uuid;