Remove futures-channel dependency in tests

.github/workflows/ci.yml

@@ -22,6 +22,12 @@ jobs:
           command: check
           args: --target=wasm32-unknown-unknown
 
+      - name: Run cargo test
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --target=x86_64-unknown-linux-gnu
+
       - name: Install wasm-pack
         run: curl https://rustwasm.github.io/wasm-pack/installer/init.sh -sSf | sh
 
@@ -54,4 +60,4 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: clippy
-          args: --target wasm32-unknown-unknown
+          args: --target=wasm32-unknown-unknown

Cargo.toml

@@ -2,7 +2,7 @@
 name = "wasm-streams"
 version = "0.2.2"
 authors = ["Mattias Buelens <mattias@buelens.com>"]
-edition = "2018"
+edition = "2021"
 license = "MIT OR Apache-2.0"
 readme = "README.md"
 repository = "https://github.com/MattiasBuelens/wasm-streams/"
@@ -29,12 +29,6 @@ features = [
 ]
 
 [dev-dependencies]
-# TODO: Investigate why using `futures-channel` does not work.
-# In theory, there is no reason why `futures` is required here.
-# The `futures-channel` is sufficient for the test cases, but
-# the code does not compile unless we use the channels from the
-# `futures` crate, which are just re-exported from `futures-channel`!
-futures = { version = "0.3.21", default-features = false, features = ["std"] }
 wasm-bindgen-test = "0.3.20"
 tokio = { version = "^1", features = ["macros", "rt"] }
 pin-project = "^1.0.6"

tests/tests/pipe.rs

@@ -1,4 +1,3 @@
-use futures::channel::mpsc;
 use futures_util::stream::iter;
 use futures_util::{SinkExt, StreamExt};
 use wasm_bindgen::prelude::*;
@@ -8,6 +7,7 @@ use wasm_streams::readable::*;
 use wasm_streams::writable::*;
 
 use crate::js::*;
+use crate::util::*;
 
 #[wasm_bindgen_test]
 async fn test_pipe_js_to_rust() {
@@ -16,7 +16,7 @@ async fn test_pipe_js_to_rust() {
         chunks.clone().into_boxed_slice(),
     ));
 
-    let (sink, stream) = mpsc::unbounded::<JsValue>();
+    let (sink, stream) = SimpleChannel::<JsValue>::new().split();
     let sink = sink.sink_map_err(|_| JsValue::from_str("cannot happen"));
     let mut writable = WritableStream::from_sink(sink);
 

tests/tests/writable_stream.rs

@@ -1,6 +1,5 @@
 use std::pin::Pin;
 
-use futures::channel::mpsc;
 use futures_util::stream::iter;
 use futures_util::{AsyncReadExt, AsyncWriteExt, SinkExt, StreamExt};
 use js_sys::Uint8Array;
@@ -96,7 +95,7 @@ async fn test_writable_stream_writer_into_sink() {
 
 #[wasm_bindgen_test]
 async fn test_writable_stream_from_sink() {
-    let (sink, stream) = mpsc::unbounded::<JsValue>();
+    let (sink, stream) = SimpleChannel::<JsValue>::new().split();
     let sink = sink.sink_map_err(|_| JsValue::from_str("cannot happen"));
     let mut writable = WritableStream::from_sink(sink);
 
@@ -115,7 +114,7 @@ async fn test_writable_stream_from_sink() {
 
 #[wasm_bindgen_test]
 async fn test_writable_stream_from_sink_then_into_sink() {
-    let (sink, stream) = mpsc::unbounded::<JsValue>();
+    let (sink, stream) = SimpleChannel::<JsValue>::new().split();
     let sink = sink.sink_map_err(|_| JsValue::from_str("cannot happen"));
     let writable = WritableStream::from_sink(sink);
     let mut sink = writable.into_sink();

tests/util/mod.rs

@@ -1,5 +1,7 @@
 pub use byte_channel::ByteChannel;
 pub use drop_observer::observe_drop;
+pub use simple_channel::SimpleChannel;
 
 pub mod byte_channel;
 pub mod drop_observer;
+pub mod simple_channel;

tests/util/simple_channel.rs

@@ -0,0 +1,157 @@
+use std::collections::VecDeque;
+use std::pin::Pin;
+use std::task::{Context, Poll, Waker};
+
+use futures_util::{Sink, Stream};
+use pin_project::pin_project;
+
+#[pin_project]
+#[derive(Debug)]
+pub struct SimpleChannel<T> {
+    queue: VecDeque<T>,
+    waker: Option<Waker>,
+    closed: bool,
+}
+
+impl<T> SimpleChannel<T> {
+    pub fn new() -> Self {
+        SimpleChannel {
+            queue: VecDeque::new(),
+            waker: None,
+            closed: false,
+        }
+    }
+}
+
+impl<T> Stream for SimpleChannel<T> {
+    type Item = T;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.project();
+        match this.queue.pop_front() {
+            Some(item) => Poll::Ready(Some(item)),
+            None if *this.closed => Poll::Ready(None),
+            None => {
+                *this.waker = Some(cx.waker().clone());
+                Poll::Pending
+            }
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (self.queue.len(), None)
+    }
+}
+
+impl<T> Sink<T> for SimpleChannel<T> {
+    type Error = ();
+
+    fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
+        let this = self.project();
+        this.queue.push_back(item);
+        if let Some(waker) = this.waker.take() {
+            waker.wake();
+        }
+        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>> {
+        let this = self.project();
+        *this.closed = true;
+        if let Some(waker) = this.waker.take() {
+            waker.wake();
+        }
+        Poll::Ready(Ok(()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use futures_util::future::join;
+    use futures_util::stream::iter;
+    use futures_util::{SinkExt, StreamExt};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn test_write_then_read() {
+        let channel = SimpleChannel::<u32>::new();
+        let (mut sink, mut stream) = channel.split();
+
+        send_many(&mut sink, vec![1, 2, 3]).await.unwrap();
+        assert_eq!(stream.next().await.unwrap(), 1);
+        assert_eq!(stream.next().await.unwrap(), 2);
+
+        send_many(&mut sink, vec![4, 5]).await.unwrap();
+        assert_eq!(stream.next().await.unwrap(), 3);
+        assert_eq!(stream.next().await.unwrap(), 4);
+        assert_eq!(stream.next().await.unwrap(), 5);
+
+        sink.close().await.unwrap();
+        assert_eq!(stream.next().await, None);
+    }
+
+    #[tokio::test]
+    async fn test_read_then_write() {
+        let channel = SimpleChannel::<u32>::new();
+        let (mut sink, mut stream) = channel.split();
+
+        join(
+            async {
+                assert_eq!(stream.next().await.unwrap(), 1);
+                assert_eq!(stream.next().await.unwrap(), 2);
+                assert_eq!(stream.next().await.unwrap(), 3);
+            },
+            async {
+                send_many(&mut sink, vec![1, 2, 3, 4]).await.unwrap();
+            },
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    async fn test_read_then_close() {
+        let channel = SimpleChannel::<u32>::new();
+        let (mut sink, mut stream) = channel.split();
+
+        join(
+            async {
+                assert_eq!(stream.next().await, None);
+            },
+            async {
+                sink.close().await.unwrap();
+            },
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    async fn test_close_then_read() {
+        let channel = SimpleChannel::<u32>::new();
+        let (mut sink, stream) = channel.split();
+
+        send_many(&mut sink, vec![1, 2, 3]).await.unwrap();
+        sink.close().await.unwrap();
+
+        // should still read items from queue
+        assert_eq!(stream.collect::<Vec<_>>().await, vec![1, 2, 3]);
+    }
+
+    async fn send_many<T, Si>(
+        sink: &mut Si,
+        values: impl IntoIterator<Item = T>,
+    ) -> Result<(), Si::Error>
+    where
+        Si: Sink<T> + Unpin,
+    {
+        sink.send_all(&mut iter(values).map(Ok)).await
+    }
+}