Remove crossbeam feature, switch to flume for all

Cargo.toml

@@ -14,11 +14,10 @@ homepage = "https://github.com/deib-polimi/noir"
 readme = "README.md"
 
 [features]
-default = ["flume", "clap", "ssh", "timestamp"]
-crossbeam = ["crossbeam-channel"]
+default = ["clap", "ssh", "timestamp"]
 timestamp = []
 ssh = ["ssh2", "whoami", "shell-escape", "sha2", "base64"]
-async-tokio = ["tokio", "flume", "futures", "tokio/net", "tokio/io-util", "tokio/time", "tokio/rt-multi-thread", "tokio/macros"]
+async-tokio = ["tokio", "futures", "tokio/net", "tokio/io-util", "tokio/time", "tokio/rt-multi-thread", "tokio/macros"]
 profiler = []
 
 [dependencies]
@@ -60,8 +59,7 @@ sha2 = { version = "0.10.8", optional = true }
 base64 = { version = "0.22.0", optional = true }
 
 # channel implementation
-crossbeam-channel = { version = "0.5.12", optional = true }
-flume = { version = "0.11.0", optional = true }
+flume = "0.11.0"
 
 # used for csv file source
 csv = "1.3.0"
@@ -100,10 +98,6 @@ micrometer = { version = "0.2.7", features = ["enable"]}
 # for the examples
 regex = "1.10.3"
 
-# used in the benchmarks
-crossbeam-channel = "0.5.12"
-flume = "0.11.0"
-
 kstring = { version = "2.0.0", features = ["serde"] }
 nexmark = { version = "0.2.0", features = ["serde"] }
 

src/channel.rs

@@ -5,13 +5,6 @@
 
 use std::time::Duration;
 
-#[cfg(feature = "crossbeam")]
-use crossbeam_channel::{
-    bounded as bounded_ext, select, unbounded as unbounded_ext, Receiver as ReceiverExt,
-    RecvError as ExtRecvError, RecvTimeoutError as ExtRecvTimeoutError, SendError as SendErrorExt,
-    Sender as SenderExt, TryRecvError as ExtTryRecvError,
-};
-#[cfg(feature = "flume")]
 use flume::{
     bounded as bounded_ext, unbounded as unbounded_ext, Receiver as ReceiverExt,
     RecvError as ExtRecvError, RecvTimeoutError as ExtRecvTimeoutError, SendError as SendErrorExt,
@@ -85,30 +78,6 @@ pub enum SelectResult<In1, In2> {
     B(Result<In2, RecvError>),
 }
 
-#[cfg(feature = "crossbeam")]
-#[macro_use]
-mod select_impl {
-    macro_rules! select_impl {
-        ($self:expr, $other:expr) => {
-            select! {
-                recv($self.0) -> el => SelectResult::A(el.map_err(RecvError::from)),
-                recv($other.0) -> el => SelectResult::B(el.map_err(RecvError::from)),
-            }
-        };
-    }
-
-    macro_rules! select_timeout_impl {
-        ($self:expr, $other:expr, $timeout:expr) => {
-            select! {
-                recv($self.0) -> el => Ok(SelectResult::A(el.map_err(RecvError::from))),
-                recv($other.0) -> el => Ok(SelectResult::B(el.map_err(RecvError::from))),
-                default($timeout) => Err(RecvTimeoutError::Timeout),
-            }
-        };
-    }
-}
-
-#[cfg(feature = "flume")]
 #[macro_use]
 mod select_impl {
     macro_rules! select_impl {
@@ -160,7 +129,7 @@ impl<T: ChannelItem> Receiver<T> {
     }
 
     #[inline]
-    #[cfg(feature = "flume")]
+
     pub async fn recv_async(&self) -> Result<T, RecvError> {
         self.0.recv_async().await.map_err(RecvError::from)
     }

src/operator/mod.rs

@@ -7,9 +7,6 @@ use std::fmt::Display;
 use std::hash::Hash;
 use std::ops::{AddAssign, Div};
 
-#[cfg(feature = "crossbeam")]
-use crossbeam_channel::{unbounded, Receiver, Sender};
-#[cfg(not(feature = "crossbeam"))]
 use flume::{unbounded, Receiver};
 #[cfg(feature = "async-tokio")]
 use futures::Future;

src/operator/sink/collect_channel.rs

@@ -5,9 +5,6 @@ use crate::operator::sink::Sink;
 use crate::operator::{ExchangeData, Operator, StreamElement};
 use crate::scheduler::ExecutionMetadata;
 
-#[cfg(feature = "crossbeam")]
-use crossbeam_channel::Sender;
-#[cfg(not(feature = "crossbeam"))]
 use flume::Sender;
 
 #[derive(Debug, Clone)]