Repository: rust-lang-nursery/futures-rs
Pull request #1880 (2019-09-21 10:02:18)

With: Ensure inner sink is ready before sending item
Previously the `With` implementation never consulted the inner sink
`poll_ready` before initiating a send operation.

After changes the `With` combinator is considered ready when processing
of previous item has been completed and the inner sink is ready.

Additionally remove buffered state since it isn't actually used.

Issue: #1834

cramertj: Nice, thanks for the PR! Can you add a test to ensure that this behavior doesn't regress?
tmiasko: Added a regression test & rebased.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6a04db2f..16911057 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,26 @@
+# 0.3.0-alpha.19 - 2019-9-25
+* Stabilized the `async-await` feature (#1816)
+* Made `async-await` feature no longer require `std` feature (#1815)
+* Updated `proc-macro2`, `syn`, and `quote` to 1.0 (#1798)
+* Exposed unstable `BiLock` (#1827)
+* Renamed "nightly" feature to "unstable" (#1823)
+* Moved to our own `io::{Empty, Repeat, Sink}` (#1829)
+* Made `AsyncRead::initializer` API unstable (#1845)
+* Moved the `Never` type from `futures-core` to `futures-util` (#1836)
+* Fixed use-after-free on panic in `ArcWake::wake_by_ref` (#1797)
+* Added `AsyncReadExt::chain` (#1810)
+* Added `Stream::size_hint` (#1853)
+* Added some missing `FusedFuture` (#1868) and `FusedStream` implementations (#1831)
+* Added a `From` impl for `Mutex` (#1839)
+* Added `Mutex::{get_mut, into_inner}` (#1839)
+* Re-exported `TryConcat` and `TryFilter` (#1814)
+* Lifted `Unpin` bound and implemented `AsyncBufRead` for `io::Take` (#1821)
+* Lifted `Unpin` bounds on `get_pin_mut` (#1820)
+* Changed `SendAll` to flush the `Sink` when the source `Stream` is pending (#1877)
+* Set default threadpool size to one if `num_cpus::get()` returns zero (#1835)
+* Removed dependency on `rand` by using our own PRNG (#1837)
+* Removed `futures-core` dependency from `futures-sink` (#1832)
+
 # 0.3.0-alpha.18 - 2019-8-9
 * Rewrote `join!` and `try_join!` as procedural macros to allow passing expressions (#1783)
 * Banned manual implementation of `TryFuture` and `TryStream` for forward compatibility. See #1776 for more details. (#1777)
diff --git a/README.md b/README.md
index 50496782..4616ea3a 100644
--- a/README.md
+++ b/README.md
@@ -21,7 +21,7 @@
 </p>
 
 <p align="center">
-  <a href="https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures/">
+  <a href="https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures/">
     Documentation
   </a> | <a href="https://rust-lang-nursery.github.io/futures-rs/">
     Website
@@ -34,7 +34,7 @@ Add this to your `Cargo.toml`:
 
 ```toml
 [dependencies]
-futures-preview = "=0.3.0-alpha.18"
+futures-preview = "=0.3.0-alpha.19"
 ```
 
 Now, you can use futures-rs:
@@ -53,7 +53,7 @@ a `#[no_std]` environment, use:
 
 ```toml
 [dependencies]
-futures-preview = { version = "=0.3.0-alpha.18", default-features = false }
+futures-preview = { version = "=0.3.0-alpha.19", default-features = false }
 ```
 
 ### Feature `async-await`
@@ -62,7 +62,7 @@ The `async-await` feature provides several convenient features using async/await
 
 ```toml
 [dependencies]
-futures-preview = { version = "=0.3.0-alpha.18", features = ["async-await"] }
+futures-preview = { version = "=0.3.0-alpha.19", features = ["async-await"] }
 ```
 
 The current `async-await` feature requires Rust nightly 2019-08-21 or later.
diff --git a/futures-channel/Cargo.toml b/futures-channel/Cargo.toml
index 1f3b3cb0..247ce224 100644
--- a/futures-channel/Cargo.toml
+++ b/futures-channel/Cargo.toml
@@ -1,12 +1,12 @@
 [package]
 name = "futures-channel-preview"
 edition = "2018"
-version = "0.3.0-alpha.18"
+version = "0.3.0-alpha.19"
 authors = ["Alex Crichton <alex@alexcrichton.com>"]
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang-nursery/futures-rs"
 homepage = "https://rust-lang-nursery.github.io/futures-rs"
-documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_channel"
+documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_channel"
 description = """
 Channels for asynchronous communication using futures-rs.
 """
@@ -27,9 +27,9 @@ unstable = ["futures-core-preview/unstable"]
 cfg-target-has-atomic = ["futures-core-preview/cfg-target-has-atomic"]
 
 [dependencies]
-futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.18", default-features = false }
-futures-sink-preview = { path = "../futures-sink", version = "=0.3.0-alpha.18", default-features = false, optional = true }
+futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.19", default-features = false }
+futures-sink-preview = { path = "../futures-sink", version = "=0.3.0-alpha.19", default-features = false, optional = true }
 
 [dev-dependencies]
-futures-preview = { path = "../futures", version = "=0.3.0-alpha.18", default-features = true }
-futures-test-preview = { path = "../futures-test", version = "=0.3.0-alpha.18", default-features = true }
+futures-preview = { path = "../futures", version = "=0.3.0-alpha.19", default-features = true }
+futures-test-preview = { path = "../futures-test", version = "=0.3.0-alpha.19", default-features = true }
diff --git a/futures-channel/src/lib.rs b/futures-channel/src/lib.rs
index d421f15b..3ac13e4e 100644
--- a/futures-channel/src/lib.rs
+++ b/futures-channel/src/lib.rs
@@ -17,7 +17,7 @@
 
 #![doc(test(attr(deny(warnings), allow(dead_code, unused_assignments, unused_variables))))]
 
-#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_channel")]
+#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_channel")]
 
 #[cfg(all(feature = "cfg-target-has-atomic", not(feature = "unstable")))]
 compile_error!("The `cfg-target-has-atomic` feature requires the `unstable` feature as an explicit opt-in to unstable features");
diff --git a/futures-core/Cargo.toml b/futures-core/Cargo.toml
index a810628c..bf0d29a0 100644
--- a/futures-core/Cargo.toml
+++ b/futures-core/Cargo.toml
@@ -1,12 +1,12 @@
 [package]
 name = "futures-core-preview"
 edition = "2018"
-version = "0.3.0-alpha.18"
+version = "0.3.0-alpha.19"
 authors = ["Alex Crichton <alex@alexcrichton.com>"]
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang-nursery/futures-rs"
 homepage = "https://rust-lang-nursery.github.io/futures-rs"
-documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_core"
+documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_core"
 description = """
 The core traits and types in for the `futures` library.
 """
@@ -28,4 +28,4 @@ cfg-target-has-atomic = []
 [dependencies]
 
 [dev-dependencies]
-futures-preview = { path = "../futures", version = "=0.3.0-alpha.18" }
+futures-preview = { path = "../futures", version = "=0.3.0-alpha.19" }
diff --git a/futures-core/src/lib.rs b/futures-core/src/lib.rs
index a7d5299f..009ba0eb 100644
--- a/futures-core/src/lib.rs
+++ b/futures-core/src/lib.rs
@@ -11,7 +11,7 @@
 
 #![doc(test(attr(deny(warnings), allow(dead_code, unused_assignments, unused_variables))))]
 
-#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_core")]
+#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_core")]
 
 #[cfg(all(feature = "cfg-target-has-atomic", not(feature = "unstable")))]
 compile_error!("The `cfg-target-has-atomic` feature requires the `unstable` feature as an explicit opt-in to unstable features");
diff --git a/futures-executor/Cargo.toml b/futures-executor/Cargo.toml
index 3c43632e..3aa15175 100644
--- a/futures-executor/Cargo.toml
+++ b/futures-executor/Cargo.toml
@@ -1,12 +1,12 @@
 [package]
 name = "futures-executor-preview"
 edition = "2018"
-version = "0.3.0-alpha.18"
+version = "0.3.0-alpha.19"
 authors = ["Alex Crichton <alex@alexcrichton.com>"]
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang-nursery/futures-rs"
 homepage = "https://rust-lang-nursery.github.io/futures-rs"
-documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_executor"
+documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_executor"
 description = """
 Executors for asynchronous tasks based on the futures-rs library.
 """
@@ -19,9 +19,9 @@ default = ["std"]
 std = ["futures-core-preview/std", "futures-util-preview/std", "num_cpus"]
 
 [dependencies]
-futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.18", default-features = false }
-futures-util-preview = { path = "../futures-util", version = "=0.3.0-alpha.18", default-features = false }
+futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.19", default-features = false }
+futures-util-preview = { path = "../futures-util", version = "=0.3.0-alpha.19", default-features = false }
 num_cpus = { version = "1.8.0", optional = true }
 
 [dev-dependencies]
-futures-preview = { path = "../futures", version = "=0.3.0-alpha.18" }
+futures-preview = { path = "../futures", version = "=0.3.0-alpha.19" }
diff --git a/futures-executor/src/lib.rs b/futures-executor/src/lib.rs
index 55379592..0cc02930 100644
--- a/futures-executor/src/lib.rs
+++ b/futures-executor/src/lib.rs
@@ -12,7 +12,7 @@
 
 #![doc(test(attr(deny(warnings), allow(dead_code, unused_assignments, unused_variables))))]
 
-#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_executor")]
+#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_executor")]
 
 #[cfg(feature = "std")]
 mod local_pool;
diff --git a/futures-io/Cargo.toml b/futures-io/Cargo.toml
index 6af7c24c..dd84a6e6 100644
--- a/futures-io/Cargo.toml
+++ b/futures-io/Cargo.toml
@@ -1,12 +1,12 @@
 [package]
 name = "futures-io-preview"
 edition = "2018"
-version = "0.3.0-alpha.18"
+version = "0.3.0-alpha.19"
 authors = ["Alex Crichton <alex@alexcrichton.com>"]
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang-nursery/futures-rs"
 homepage = "https://rust-lang-nursery.github.io/futures-rs"
-documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_io"
+documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_io"
 description = """
 The `AsyncRead` and `AsyncWrite` traits for the futures-rs library.
 """
diff --git a/futures-io/src/lib.rs b/futures-io/src/lib.rs
index 551ee070..ca07fb0d 100644
--- a/futures-io/src/lib.rs
+++ b/futures-io/src/lib.rs
@@ -19,7 +19,7 @@
 
 #![doc(test(attr(deny(warnings), allow(dead_code, unused_assignments, unused_variables))))]
 
-#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_io")]
+#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_io")]
 
 #[cfg(all(feature = "read_initializer", not(feature = "unstable")))]
 compile_error!("The `read_initializer` feature requires the `unstable` feature as an explicit opt-in to unstable features");
diff --git a/futures-join-macro/Cargo.toml b/futures-join-macro/Cargo.toml
index caba4692..fd4b89ae 100644
--- a/futures-join-macro/Cargo.toml
+++ b/futures-join-macro/Cargo.toml
@@ -1,12 +1,12 @@
 [package]
 name = "futures-join-macro-preview"
 edition = "2018"
-version = "0.3.0-alpha.18"
+version = "0.3.0-alpha.19"
 authors = ["Taiki Endo <te316e89@gmail.com>"]
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang-nursery/futures-rs"
 homepage = "https://rust-lang-nursery.github.io/futures-rs"
-documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_join_macro"
+documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_join_macro"
 description = """
 Definition of the `join!` macro and the `try_join!` macro.
 """
diff --git a/futures-select-macro/Cargo.toml b/futures-select-macro/Cargo.toml
index 78ef7c6a..e40b041d 100644
--- a/futures-select-macro/Cargo.toml
+++ b/futures-select-macro/Cargo.toml
@@ -1,12 +1,12 @@
 [package]
 name = "futures-select-macro-preview"
 edition = "2018"
-version = "0.3.0-alpha.18"
+version = "0.3.0-alpha.19"
 authors = ["Taylor Cramer <cramertj@google.com>"]
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang-nursery/futures-rs"
 homepage = "https://rust-lang-nursery.github.io/futures-rs"
-documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_select_macro"
+documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_select_macro"
 description = """
 The `select!` macro for waiting on multiple different `Future`s at once and handling the first one to complete.
 """
diff --git a/futures-sink/Cargo.toml b/futures-sink/Cargo.toml
index ee751631..109c87a3 100644
--- a/futures-sink/Cargo.toml
+++ b/futures-sink/Cargo.toml
@@ -1,12 +1,12 @@
 [package]
 name = "futures-sink-preview"
 edition = "2018"
-version = "0.3.0-alpha.18"
+version = "0.3.0-alpha.19"
 authors = ["Alex Crichton <alex@alexcrichton.com>"]
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang-nursery/futures-rs"
 homepage = "https://rust-lang-nursery.github.io/futures-rs"
-documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_sink"
+documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_sink"
 description = """
 The asynchronous `Sink` trait for the futures-rs library.
 """
diff --git a/futures-sink/src/lib.rs b/futures-sink/src/lib.rs
index 8400325d..d4cb82a4 100644
--- a/futures-sink/src/lib.rs
+++ b/futures-sink/src/lib.rs
@@ -11,7 +11,7 @@
 
 #![doc(test(attr(deny(warnings), allow(dead_code, unused_assignments, unused_variables))))]
 
-#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_sink")]
+#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_sink")]
 
 #[cfg(feature = "alloc")]
 extern crate alloc;
diff --git a/futures-util/Cargo.toml b/futures-util/Cargo.toml
index 1825ea83..4cc51799 100644
--- a/futures-util/Cargo.toml
+++ b/futures-util/Cargo.toml
@@ -1,12 +1,12 @@
 [package]
 name = "futures-util-preview"
 edition = "2018"
-version = "0.3.0-alpha.18"
+version = "0.3.0-alpha.19"
 authors = ["Alex Crichton <alex@alexcrichton.com>"]
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang-nursery/futures-rs"
 homepage = "https://rust-lang-nursery.github.io/futures-rs"
-documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_util"
+documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_util"
 description = """
 Common utilities and extension traits for the futures-rs library.
 """
@@ -36,12 +36,12 @@ bilock = []
 read_initializer = ["io", "futures-io-preview/read_initializer", "futures-io-preview/unstable"]
 
 [dependencies]
-futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.18", default-features = false }
-futures-channel-preview = { path = "../futures-channel", version = "=0.3.0-alpha.18", default-features = false, features = ["std"], optional = true }
-futures-io-preview = { path = "../futures-io", version = "=0.3.0-alpha.18", default-features = false, features = ["std"], optional = true }
-futures-sink-preview = { path = "../futures-sink", version = "=0.3.0-alpha.18", default-features = false, optional = true }
-futures-join-macro-preview = { path = "../futures-join-macro", version = "=0.3.0-alpha.18", default-features = false, optional = true }
-futures-select-macro-preview = { path = "../futures-select-macro", version = "=0.3.0-alpha.18", default-features = false, optional = true }
+futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.19", default-features = false }
+futures-channel-preview = { path = "../futures-channel", version = "=0.3.0-alpha.19", default-features = false, features = ["std"], optional = true }
+futures-io-preview = { path = "../futures-io", version = "=0.3.0-alpha.19", default-features = false, features = ["std"], optional = true }
+futures-sink-preview = { path = "../futures-sink", version = "=0.3.0-alpha.19", default-features = false, optional = true }
+futures-join-macro-preview = { path = "../futures-join-macro", version = "=0.3.0-alpha.19", default-features = false, optional = true }
+futures-select-macro-preview = { path = "../futures-select-macro", version = "=0.3.0-alpha.19", default-features = false, optional = true }
 proc-macro-hack = { version = "0.5.9", optional = true }
 proc-macro-nested = { version = "0.1.2", optional = true }
 slab = { version = "0.4", optional = true }
@@ -51,8 +51,8 @@ tokio-io = { version = "0.1.9", optional = true }
 pin-utils = "0.1.0-alpha.4"
 
 [dev-dependencies]
-futures-preview = { path = "../futures", version = "=0.3.0-alpha.18", features = ["async-await"] }
-futures-test-preview = { path = "../futures-test", version = "=0.3.0-alpha.18" }
+futures-preview = { path = "../futures", version = "=0.3.0-alpha.19", features = ["async-await"] }
+futures-test-preview = { path = "../futures-test", version = "=0.3.0-alpha.19" }
 tokio = "0.1.11"
 
 [package.metadata.docs.rs]
diff --git a/futures-util/src/lib.rs b/futures-util/src/lib.rs
index 3f91a9eb..bd404f2a 100644
--- a/futures-util/src/lib.rs
+++ b/futures-util/src/lib.rs
@@ -12,7 +12,7 @@
 
 #![doc(test(attr(deny(warnings), allow(dead_code, unused_assignments, unused_variables))))]
 
-#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_util")]
+#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_util")]
 
 #[cfg(all(feature = "cfg-target-has-atomic", not(feature = "unstable")))]
 compile_error!("The `cfg-target-has-atomic` feature requires the `unstable` feature as an explicit opt-in to unstable features");
diff --git a/futures-util/src/sink/with.rs b/futures-util/src/sink/with.rs
index 9ea47a73..5acac017 100644
--- a/futures-util/src/sink/with.rs
+++ b/futures-util/src/sink/with.rs
@@ -1,6 +1,5 @@
 use core::fmt;
 use core::marker::PhantomData;
-use core::mem;
 use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::stream::Stream;
@@ -13,8 +12,8 @@ use pin_utils::{unsafe_pinned, unsafe_unpinned};
 pub struct With<Si, Item, U, Fut, F> {
     sink: Si,
     f: F,
-    state: State<Fut, Item>,
-    _phantom: PhantomData<fn(U)>,
+    state: Option<Fut>,
+    _phantom: PhantomData<fn(U) -> Item>,
 }
 
 impl<Si, Item, U, Fut, F> Unpin for With<Si, Item, U, Fut, F>
@@ -27,7 +26,6 @@ impl<Si, Item, U, Fut, F> fmt::Debug for With<Si, Item, U, Fut, F>
 where
     Si: fmt::Debug,
     Fut: fmt::Debug,
-    Item: fmt::Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("With")
@@ -44,7 +42,7 @@ where Si: Sink<Item>,
 {
     unsafe_pinned!(sink: Si);
     unsafe_unpinned!(f: F);
-    unsafe_pinned!(state: State<Fut, Item>);
+    unsafe_pinned!(state: Option<Fut>);
 
     pub(super) fn new<E>(sink: Si, f: F) -> Self
         where
@@ -52,7 +50,7 @@ where Si: Sink<Item>,
             E: From<Si::Error>,
     {
         With {
-            state: State::Empty,
+            state: None,
             sink,
             f,
             _phantom: PhantomData,
@@ -60,29 +58,6 @@ where Si: Sink<Item>,
     }
 }
 
-#[derive(Debug)]
-enum State<Fut, T> {
-    Empty,
-    Process(Fut),
-    Buffered(T),
-}
-
-impl<Fut, T> State<Fut, T> {
-    #[allow(clippy::wrong_self_convention)]
-    fn as_pin_mut(self: Pin<&mut Self>) -> State<Pin<&mut Fut>, Pin<&mut T>> {
-        unsafe {
-            match self.get_unchecked_mut() {
-                State::Empty =>
-                    State::Empty,
-                State::Process(fut) =>
-                    State::Process(Pin::new_unchecked(fut)),
-                State::Buffered(item) =>
-                    State::Buffered(Pin::new_unchecked(item)),
-            }
-        }
-    }
-}
-
 // Forwarding impl of Stream from the underlying sink
 impl<S, Item, U, Fut, F> Stream for With<S, Item, U, Fut, F>
     where S: Stream + Sink<Item>,
@@ -132,23 +107,18 @@ impl<Si, Item, U, Fut, F, E> With<Si, Item, U, Fut, F>
         self.sink
     }
 
+    /// Completes the processing of previous item if any.
     fn poll(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), E>> {
-        let buffered = match self.as_mut().state().as_pin_mut() {
-            State::Empty => return Poll::Ready(Ok(())),
-            State::Process(fut) => Some(ready!(fut.poll(cx))?),
-            State::Buffered(_) => None,
+        let item = match self.as_mut().state().as_pin_mut() {
+            None => return Poll::Ready(Ok(())),
+            Some(fut) => ready!(fut.poll(cx))?,
         };
-        if let Some(buffered) = buffered {
-            self.as_mut().state().set(State::Buffered(buffered));
-        }
-        if let State::Buffered(item) = unsafe { mem::replace(self.as_mut().state().get_unchecked_mut(), State::Empty) } {
-            Poll::Ready(self.as_mut().sink().start_send(item).map_err(Into::into))
-        } else {
-            unreachable!()
-        }
+        self.as_mut().state().set(None);
+        self.as_mut().sink().start_send(item)?;
+        Poll::Ready(Ok(()))
     }
 }
 
@@ -161,18 +131,20 @@ impl<Si, Item, U, Fut, F, E> Sink<U> for With<Si, Item, U, Fut, F>
     type Error = E;
 
     fn poll_ready(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), Self::Error>> {
-        self.poll(cx)
+        ready!(self.as_mut().poll(cx))?;
+        ready!(self.as_mut().sink().poll_ready(cx)?);
+        Poll::Ready(Ok(()))
     }
 
     fn start_send(
         mut self: Pin<&mut Self>,
         item: U,
     ) -> Result<(), Self::Error> {
-        let item = (self.as_mut().f())(item);
-        self.as_mut().state().set(State::Process(item));
+        let future = (self.as_mut().f())(item);
+        self.as_mut().state().set(Some(future));
         Ok(())
     }
 
diff --git a/futures/Cargo.toml b/futures/Cargo.toml
index 71e99ea6..ed7c6b1c 100644
--- a/futures/Cargo.toml
+++ b/futures/Cargo.toml
@@ -1,14 +1,14 @@
 [package]
 name = "futures-preview"
 edition = "2018"
-version = "0.3.0-alpha.18"
+version = "0.3.0-alpha.19"
 authors = ["Alex Crichton <alex@alexcrichton.com>"]
 license = "MIT OR Apache-2.0"
 readme = "../README.md"
 keywords = ["futures", "async", "future"]
 repository = "https://github.com/rust-lang-nursery/futures-rs"
 homepage = "https://rust-lang-nursery.github.io/futures-rs"
-documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures"
+documentation = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures"
 description = """
 An implementation of futures and streams featuring zero allocations,
 composability, and iterator-like interfaces.
@@ -22,16 +22,16 @@ name = "futures"
 travis-ci = { repository = "rust-lang-nursery/futures-rs" }
 
 [dependencies]
-futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.18", default-features = false }
-futures-channel-preview = { path = "../futures-channel", version = "=0.3.0-alpha.18", default-features = false, features = ["sink"] }
-futures-executor-preview = { path = "../futures-executor", version = "=0.3.0-alpha.18", default-features = false }
-futures-io-preview = { path = "../futures-io", version = "=0.3.0-alpha.18", default-features = false }
-futures-sink-preview = { path = "../futures-sink", version = "=0.3.0-alpha.18", default-features = false }
-futures-util-preview = { path = "../futures-util", version = "=0.3.0-alpha.18", default-features = false, features = ["sink"] }
+futures-core-preview = { path = "../futures-core", version = "=0.3.0-alpha.19", default-features = false }
+futures-channel-preview = { path = "../futures-channel", version = "=0.3.0-alpha.19", default-features = false, features = ["sink"] }
+futures-executor-preview = { path = "../futures-executor", version = "=0.3.0-alpha.19", default-features = false }
+futures-io-preview = { path = "../futures-io", version = "=0.3.0-alpha.19", default-features = false }
+futures-sink-preview = { path = "../futures-sink", version = "=0.3.0-alpha.19", default-features = false }
+futures-util-preview = { path = "../futures-util", version = "=0.3.0-alpha.19", default-features = false, features = ["sink"] }
 
 [dev-dependencies]
 pin-utils = "0.1.0-alpha.4"
-futures-test-preview = { path = "../futures-test", version = "=0.3.0-alpha.18" }
+futures-test-preview = { path = "../futures-test", version = "=0.3.0-alpha.19" }
 tokio = "0.1.11"
 assert_matches = "1.3.0"
 
diff --git a/futures/src/lib.rs b/futures/src/lib.rs
index 46158ca8..2ff3f474 100644
--- a/futures/src/lib.rs
+++ b/futures/src/lib.rs
@@ -33,7 +33,7 @@
 
 #![doc(test(attr(deny(warnings), allow(dead_code, unused_assignments, unused_variables))))]
 
-#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures")]
+#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures")]
 
 #[cfg(all(feature = "cfg-target-has-atomic", not(feature = "unstable")))]
 compile_error!("The `cfg-target-has-atomic` feature requires the `unstable` feature as an explicit opt-in to unstable features");


Test patch:
diff --git a/futures-test/Cargo.toml b/futures-test/Cargo.toml
index 31575ee0..ecc9fb51 100644
--- a/futures-test/Cargo.toml
+++ b/futures-test/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "futures-test-preview"
 edition = "2018"
-version = "0.3.0-alpha.18"
+version = "0.3.0-alpha.19"
 authors = ["Wim Looman <wim@nemo157.com>"]
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/rust-lang-nursery/futures-rs"
@@ -15,14 +15,14 @@ Common utilities for testing components built off futures-rs.
 name = "futures_test"
 
 [dependencies]
-futures-core-preview = { version = "=0.3.0-alpha.18", path = "../futures-core", default-features = false }
-futures-io-preview = { version = "=0.3.0-alpha.18", path = "../futures-io", default-features = false }
-futures-util-preview = { version = "=0.3.0-alpha.18", path = "../futures-util", default-features = false }
-futures-executor-preview = { version = "=0.3.0-alpha.18", path = "../futures-executor", default-features = false }
+futures-core-preview = { version = "=0.3.0-alpha.19", path = "../futures-core", default-features = false }
+futures-io-preview = { version = "=0.3.0-alpha.19", path = "../futures-io", default-features = false }
+futures-util-preview = { version = "=0.3.0-alpha.19", path = "../futures-util", default-features = false }
+futures-executor-preview = { version = "=0.3.0-alpha.19", path = "../futures-executor", default-features = false }
 pin-utils = { version = "0.1.0-alpha.4", default-features = false }
 
 [dev-dependencies]
-futures-preview = { version = "=0.3.0-alpha.18", path = "../futures", default-features = false, features = ["std"] }
+futures-preview = { version = "=0.3.0-alpha.19", path = "../futures", default-features = false, features = ["std"] }
 
 [features]
 default = ["std"]
diff --git a/futures-test/src/lib.rs b/futures-test/src/lib.rs
index e95706b2..a783abfa 100644
--- a/futures-test/src/lib.rs
+++ b/futures-test/src/lib.rs
@@ -7,7 +7,7 @@
 
 #![doc(test(attr(deny(warnings), allow(dead_code, unused_assignments, unused_variables))))]
 
-#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.18/futures_test")]
+#![doc(html_root_url = "https://rust-lang-nursery.github.io/futures-api-docs/0.3.0-alpha.19/futures_test")]
 
 #[cfg(not(feature = "std"))]
 compile_error!("`futures-test` must have the `std` feature activated, this is a default-active feature");
diff --git a/futures/tests/sink.rs b/futures/tests/sink.rs
index be5e1453..9849af36 100644
--- a/futures/tests/sink.rs
+++ b/futures/tests/sink.rs
@@ -224,6 +224,32 @@ fn with_flat_map() {
     assert_eq!(sink.get_ref(), &[1, 2, 2, 3, 3, 3]);
 }
 
+// Check that `with` propagates `poll_ready` to the inner sink.
+// Regression test for the issue #1834.
+#[test]
+fn with_propagates_poll_ready() {
+    let (tx, mut rx) = mpsc::channel::<i32>(0);
+    let mut tx = tx.with(|item: i32| future::ok::<i32, mpsc::SendError>(item + 10));
+
+    block_on(future::lazy(|_| {
+        flag_cx(|flag, cx| {
+            let mut tx = Pin::new(&mut tx);
+
+            // Should be ready for the first item.
+            assert_eq!(tx.as_mut().poll_ready(cx), Poll::Ready(Ok(())));
+            assert_eq!(tx.as_mut().start_send(0), Ok(()));
+
+            // Should be ready for the second item only after the first one is received.
+            assert_eq!(tx.as_mut().poll_ready(cx), Poll::Pending);
+            assert!(!flag.get());
+            sassert_next(&mut rx, 10);
+            assert!(flag.get());
+            assert_eq!(tx.as_mut().poll_ready(cx), Poll::Ready(Ok(())));
+            assert_eq!(tx.as_mut().start_send(1), Ok(()));
+        })
+    }));
+}
+
 // Immediately accepts all requests to start pushing, but completion is managed
 // by manually flushing
 struct ManualFlush<T: Unpin> {