Add when_read_async, cleanup when_read

README.md

@@ -102,9 +102,10 @@ async fn get_cat_url(id: u32) -> String {
 
 ### Readers
 
-| Modifier    | Description            |
-| ----------- | ---------------------- |
-| `when_read` | reads into byte buffer |
+| Modifier          | Description                           | Requires feature |
+| ----------------- | ------------------------------------- | ---------------- |
+| `when_read`       | reads into byte buffer                |                  |
+| `when_read_async` | asynchronously reads into byte buffer | `futures`        |
 
 ### Futures
 

src/assertions.rs

@@ -227,7 +227,8 @@
 //! [`to_be_some`]: crate::prelude::OptionAssertions::to_be_some
 //! [`to_equal`]: crate::prelude::GeneralAssertions::to_equal
 
-// pub mod functions;
+#[cfg(feature = "futures")]
+pub mod async_read;
 #[cfg(feature = "futures")]
 pub mod futures;
 pub mod general;

src/assertions/async_read.rs

@@ -0,0 +1,9 @@
+//! Modifiers for types that can be read asynchronously.
+
+mod extensions;
+mod modifiers;
+mod outputs;
+
+pub use extensions::*;
+pub use modifiers::*;
+pub use outputs::*;

src/assertions/async_read/extensions.rs

@@ -0,0 +1,75 @@
+use futures::AsyncRead;
+
+use crate::assertions::AssertionBuilder;
+
+use super::WhenReadAsyncModifier;
+
+/// Modifiers for types that implement [`futures::AsyncRead`].
+pub trait AsyncReadAssertions<T, M>
+where
+    T: AsyncRead,
+{
+    /// Asynchronously reads the subject into a buffer, then executes the
+    /// assertion on it.
+    ///
+    /// ```
+    /// # use expecters::prelude::*;
+    /// use futures::io::Cursor;
+    /// # #[tokio::main(flavor = "current_thread")]
+    /// # async fn main() {
+    /// expect!(
+    ///     Cursor::new("Hello, world!"),
+    ///     when_read_async,
+    ///     as_utf8,
+    ///     to_equal("Hello, world!"),
+    /// )
+    /// .await;
+    /// # }
+    /// ```
+    ///
+    /// The assertion fails if reading the subject fails:
+    ///
+    /// ```should_panic
+    /// # use expecters::prelude::*;
+    /// use std::{
+    ///     pin::Pin,
+    ///     task::{Context, Poll},
+    /// };
+    ///
+    /// use futures::io::{Error, ErrorKind, AsyncRead};
+    ///
+    /// struct MyReader;
+    ///
+    /// impl AsyncRead for MyReader {
+    ///     fn poll_read(
+    ///         self: Pin<&mut Self>,
+    ///         _cx: &mut Context,
+    ///         _buf: &mut [u8],
+    ///     ) -> Poll<std::io::Result<usize>> {
+    ///         Poll::Ready(Err(Error::new(ErrorKind::Other, "always fail")))
+    ///     }
+    /// }
+    ///
+    /// # #[tokio::main(flavor = "current_thread")]
+    /// # async fn main() {
+    /// expect!(
+    ///     MyReader,
+    ///     when_read_async,
+    ///     count,
+    ///     to_be_greater_than_or_equal_to(0),
+    /// )
+    /// .await;
+    /// # }
+    /// ```
+    fn when_read_async(self) -> AssertionBuilder<Vec<u8>, WhenReadAsyncModifier<M>>;
+}
+
+impl<T, M> AsyncReadAssertions<T, M> for AssertionBuilder<T, M>
+where
+    T: AsyncRead,
+{
+    #[inline]
+    fn when_read_async(self) -> AssertionBuilder<Vec<u8>, WhenReadAsyncModifier<M>> {
+        AssertionBuilder::modify(self, WhenReadAsyncModifier::new)
+    }
+}

src/assertions/async_read/modifiers.rs

@@ -0,0 +1,3 @@
+mod when_read;
+
+pub use when_read::*;

src/assertions/async_read/modifiers/when_read.rs

@@ -0,0 +1,51 @@
+use futures::AsyncRead;
+
+use crate::assertions::{
+    async_read::WhenReadAsyncFuture, general::IntoInitializableOutput, Assertion, AssertionContext,
+    AssertionContextBuilder, AssertionModifier,
+};
+
+/// Reads a subject into a buffer asynchronously.
+#[derive(Clone, Debug)]
+pub struct WhenReadAsyncModifier<M> {
+    prev: M,
+}
+
+impl<M> WhenReadAsyncModifier<M> {
+    #[inline]
+    pub(crate) fn new(prev: M) -> Self {
+        Self { prev }
+    }
+}
+
+impl<M, A> AssertionModifier<A> for WhenReadAsyncModifier<M>
+where
+    M: AssertionModifier<WhenReadAsyncAssertion<A>>,
+{
+    type Output = M::Output;
+
+    #[inline]
+    fn apply(self, cx: AssertionContextBuilder, next: A) -> Self::Output {
+        self.prev.apply(cx, WhenReadAsyncAssertion { next })
+    }
+}
+
+/// Reads the subject into a buffer asynchronously and executes the inner
+/// assertion on it.
+#[derive(Clone, Debug)]
+pub struct WhenReadAsyncAssertion<A> {
+    next: A,
+}
+
+impl<A, T> Assertion<T> for WhenReadAsyncAssertion<A>
+where
+    A: Assertion<Vec<u8>, Output: IntoInitializableOutput>,
+    T: AsyncRead,
+{
+    type Output = WhenReadAsyncFuture<T, A>;
+
+    #[inline]
+    fn execute(self, cx: AssertionContext, subject: T) -> Self::Output {
+        WhenReadAsyncFuture::new(cx, subject, self.next)
+    }
+}

src/assertions/async_read/outputs.rs

@@ -0,0 +1,3 @@
+mod when_read;
+
+pub use when_read::*;

src/assertions/async_read/outputs/when_read.rs

@@ -0,0 +1,94 @@
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{ready, Context, Poll},
+};
+
+use futures::AsyncRead;
+use pin_project_lite::pin_project;
+
+use crate::assertions::{general::IntoInitializableOutput, Assertion, AssertionContext};
+
+pin_project! {
+    /// Asynchronously reads a subject and executes an assertion on it.
+    #[derive(Clone, Debug)]
+    pub struct WhenReadAsyncFuture<T, A> {
+        #[pin]
+        subject: T,
+        buffer: Vec<u8>,
+        result: Vec<u8>,
+        next: Option<(AssertionContext, A)>
+    }
+}
+
+impl<T, A> WhenReadAsyncFuture<T, A> {
+    #[inline]
+    pub(crate) fn new(cx: AssertionContext, subject: T, next: A) -> Self {
+        WhenReadAsyncFuture {
+            subject,
+            buffer: vec![0; 32],
+            result: Vec::new(),
+            next: Some((cx, next)),
+        }
+    }
+}
+
+impl<T, A> Future for WhenReadAsyncFuture<T, A>
+where
+    T: AsyncRead,
+    A: Assertion<Vec<u8>, Output: IntoInitializableOutput>,
+{
+    type Output = <A::Output as IntoInitializableOutput>::Initialized;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        let mut projected = self.project();
+
+        // Read the subject
+        loop {
+            let result = ready!(projected.subject.as_mut().poll_read(cx, projected.buffer));
+            match result {
+                Ok(0) => break,
+                Ok(n) => {
+                    projected.result.extend(&projected.buffer[..n]);
+
+                    // Check if we can grow the buffer for the next read
+                    if n == projected.buffer.len() {
+                        projected.buffer.reserve(32);
+                        projected.buffer.resize(projected.buffer.capacity(), 0);
+                    }
+                }
+                Err(error) => {
+                    let (mut cx, _) = projected.next.take().expect("poll after ready");
+                    cx.annotate("error", error);
+                    return Poll::Ready(cx.fail("failed to read"));
+                }
+            };
+        }
+
+        let (mut cx, next) = projected.next.take().expect("poll after ready");
+        cx.annotate("read bytes", projected.result.len());
+        Poll::Ready(
+            next.execute(cx, std::mem::take(projected.result))
+                .into_initialized(),
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use futures::io::Cursor;
+
+    use crate::prelude::*;
+
+    #[tokio::test]
+    async fn long_data() {
+        let subject = "Hello, world! ".repeat(100);
+        expect!(
+            Cursor::new(subject.clone()),
+            when_read_async,
+            as_utf8,
+            to_equal(subject),
+        )
+        .await;
+    }
+}

src/assertions/read/extensions.rs

@@ -2,7 +2,7 @@ use std::io::Read;
 
 use crate::assertions::AssertionBuilder;
 
-use super::WhenReadAsBytesModifier;
+use super::WhenReadModifier;
 
 /// Modifiers for types that implement [`Read`].
 pub trait ReadExtensions<T, M>
@@ -31,7 +31,7 @@ where
     /// struct MyReader;
     ///
     /// impl Read for MyReader {
-    ///     fn read(&mut self, _: &mut [u8]) -> std::io::Result<usize> {
+    ///     fn read(&mut self, _buf: &mut [u8]) -> std::io::Result<usize> {
     ///         Err(Error::new(ErrorKind::Other, "always fail"))
     ///     }
     /// }
@@ -43,15 +43,15 @@ where
     ///     to_be_greater_than_or_equal_to(0),
     /// );
     /// ```
-    fn when_read(self) -> AssertionBuilder<Vec<u8>, WhenReadAsBytesModifier<M>>;
+    fn when_read(self) -> AssertionBuilder<Vec<u8>, WhenReadModifier<M>>;
 }
 
 impl<T, M> ReadExtensions<T, M> for AssertionBuilder<T, M>
 where
     T: Read,
 {
     #[inline]
-    fn when_read(self) -> AssertionBuilder<Vec<u8>, WhenReadAsBytesModifier<M>> {
-        AssertionBuilder::modify(self, WhenReadAsBytesModifier::new)
+    fn when_read(self) -> AssertionBuilder<Vec<u8>, WhenReadModifier<M>> {
+        AssertionBuilder::modify(self, WhenReadModifier::new)
     }
 }

src/assertions/read/modifiers.rs

@@ -1,3 +1,3 @@
-mod as_bytes;
+mod when_read;
 
-pub use as_bytes::*;
+pub use when_read::*;

src/assertions/read/modifiers/as_bytes.rs → src/assertions/read/modifiers/when_read.rs

@@ -7,35 +7,35 @@ use crate::assertions::{
 
 /// Reads a subject into a buffer.
 #[derive(Clone, Debug)]
-pub struct WhenReadAsBytesModifier<M> {
+pub struct WhenReadModifier<M> {
     prev: M,
 }
 
-impl<M> WhenReadAsBytesModifier<M> {
+impl<M> WhenReadModifier<M> {
     #[inline]
     pub(crate) fn new(prev: M) -> Self {
-        WhenReadAsBytesModifier { prev }
+        WhenReadModifier { prev }
     }
 }
 
-impl<M, A> AssertionModifier<A> for WhenReadAsBytesModifier<M>
+impl<M, A> AssertionModifier<A> for WhenReadModifier<M>
 where
-    M: AssertionModifier<WhenReadAsBytesAssertion<A>>,
+    M: AssertionModifier<WhenReadAssertion<A>>,
 {
     type Output = M::Output;
 
     fn apply(self, cx: AssertionContextBuilder, next: A) -> Self::Output {
-        self.prev.apply(cx, WhenReadAsBytesAssertion { next })
+        self.prev.apply(cx, WhenReadAssertion { next })
     }
 }
 
 /// Reads the subject into a buffer and executes the inner assertion on it.
 #[derive(Clone, Debug)]
-pub struct WhenReadAsBytesAssertion<A> {
+pub struct WhenReadAssertion<A> {
     next: A,
 }
 
-impl<T, A> Assertion<T> for WhenReadAsBytesAssertion<A>
+impl<T, A> Assertion<T> for WhenReadAssertion<A>
 where
     T: Read,
     A: Assertion<Vec<u8>, Output: IntoInitializableOutput>,
@@ -51,6 +51,7 @@ where
             }
         };
 
+        cx.annotate("read bytes", bytes.len());
         self.next.execute(cx, bytes).into_initialized()
     }
 }

src/prelude.rs

@@ -24,4 +24,4 @@ pub use crate::{
 };
 
 #[cfg(feature = "futures")]
-pub use crate::assertions::futures::FutureAssertions;
+pub use crate::assertions::{async_read::AsyncReadAssertions, futures::FutureAssertions};