Merge pull request #117 from actix/new-connect

Release actix-http v2.0.0-alpha.2

.github/workflows/windows-mingw.yml

@@ -1,54 +0,0 @@
-name: CI (Windows-mingw)
-
-on: [push, pull_request]
-
-env:
-  OPENSSL_DIR: d:\a\_temp\msys\msys64\usr
-
-jobs:
-  build_and_test:
-    strategy:
-      fail-fast: false
-      matrix:
-        version:
-          - stable
-          - nightly
-
-    name: ${{ matrix.version }} - x86_64-pc-windows-gnu
-    runs-on: windows-latest
-
-    steps:
-      - uses: actions/checkout@master
-
-      - name: Install ${{ matrix.version }}
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: ${{ matrix.version }}-x86_64-pc-windows-gnu
-          profile: minimal
-          override: true
-
-      - name: Install MSYS2
-        uses: numworks/setup-msys2@v1
-
-      - name: Install OpenSSL
-        run: |
-          msys2do pacman --noconfirm -S openssl-devel pkg-config
-
-      - name: Copy and check libs
-        run: |
-          Copy-Item d:\a\_temp\msys\msys64\usr\lib\libssl.dll.a d:\a\_temp\msys\msys64\usr\lib\libssl.dll
-          Copy-Item d:\a\_temp\msys\msys64\usr\lib\libcrypto.dll.a d:\a\_temp\msys\msys64\usr\lib\libcrypto.dll
-          Get-ChildItem d:\a\_temp\msys\msys64\usr\lib
-          Get-ChildItem d:\a\_temp\msys\msys64\usr
-
-      - name: check build
-        uses: actions-rs/cargo@v1
-        with:
-          command: check
-          args: --all --bins --examples --tests
-
-      - name: tests
-        uses: actions-rs/cargo@v1
-        with:
-          command: test
-          args: --all --all-features --no-fail-fast -- --nocapture

actix-codec/Cargo.toml

@@ -23,4 +23,5 @@ futures-core = "0.3.1"
 futures-sink = "0.3.1"
 tokio = { version = "0.2.4", default-features=false }
 tokio-util = { version = "0.2.0", default-features=false, features=["codec"] }
-log = "0.4"
+log = "0.4"
+pin-project = "0.4.8"

actix-codec/src/framed.rs

@@ -5,6 +5,7 @@ use std::{fmt, io};
 use bytes::{Buf, BytesMut};
 use futures_core::{ready, Stream};
 use futures_sink::Sink;
+use pin_project::pin_project;
 
 use crate::{AsyncRead, AsyncWrite, Decoder, Encoder};
 
@@ -20,7 +21,9 @@ bitflags::bitflags! {
 
 /// A unified `Stream` and `Sink` interface to an underlying I/O object, using
 /// the `Encoder` and `Decoder` traits to encode and decode frames.
+#[pin_project]
 pub struct Framed<T, U> {
+    #[pin]
     io: T,
     codec: U,
     flags: Flags,
@@ -28,8 +31,6 @@ pub struct Framed<T, U> {
     write_buf: BytesMut,
 }
 
-impl<T, U> Unpin for Framed<T, U> {}
-
 impl<T, U> Framed<T, U>
 where
     T: AsyncRead + AsyncWrite,
@@ -185,17 +186,18 @@ impl<T, U> Framed<T, U> {
 
 impl<T, U> Framed<T, U> {
     /// Serialize item and Write to the inner buffer
-    pub fn write(&mut self, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error>
+    pub fn write(mut self: Pin<&mut Self>, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error>
     where
         T: AsyncWrite,
         U: Encoder,
     {
-        let remaining = self.write_buf.capacity() - self.write_buf.len();
+        let this = self.as_mut().project();
+        let remaining = this.write_buf.capacity() - this.write_buf.len();
         if remaining < LW {
-            self.write_buf.reserve(HW - remaining);
+            this.write_buf.reserve(HW - remaining);
         }
 
-        self.codec.encode(item, &mut self.write_buf)?;
+        this.codec.encode(item, this.write_buf)?;
         Ok(())
     }
 
@@ -207,21 +209,22 @@ impl<T, U> Framed<T, U> {
     }
 
     /// Try to read underlying I/O stream and decode item.
-    pub fn next_item(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
+    pub fn next_item(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
     where
         T: AsyncRead,
         U: Decoder,
     {
         loop {
+            let mut this = self.as_mut().project();
             // Repeatedly call `decode` or `decode_eof` as long as it is
             // "readable". Readable is defined as not having returned `None`. If
             // the upstream has returned EOF, and the decoder is no longer
             // readable, it can be assumed that the decoder will never become
             // readable again, at which point the stream is terminated.
 
-            if self.flags.contains(Flags::READABLE) {
-                if self.flags.contains(Flags::EOF) {
-                    match self.codec.decode_eof(&mut self.read_buf) {
+            if this.flags.contains(Flags::READABLE) {
+                if this.flags.contains(Flags::EOF) {
+                    match this.codec.decode_eof(&mut this.read_buf) {
                         Ok(Some(frame)) => return Poll::Ready(Some(Ok(frame))),
                         Ok(None) => return Poll::Ready(None),
                         Err(e) => return Poll::Ready(Some(Err(e))),
@@ -230,7 +233,7 @@ impl<T, U> Framed<T, U> {
 
                 log::trace!("attempting to decode a frame");
 
-                match self.codec.decode(&mut self.read_buf) {
+                match this.codec.decode(&mut this.read_buf) {
                     Ok(Some(frame)) => {
                         log::trace!("frame decoded from buffer");
                         return Poll::Ready(Some(Ok(frame)));
@@ -239,45 +242,44 @@ impl<T, U> Framed<T, U> {
                     _ => (), // Need more data
                 }
 
-                self.flags.remove(Flags::READABLE);
+                this.flags.remove(Flags::READABLE);
             }
 
-            debug_assert!(!self.flags.contains(Flags::EOF));
+            debug_assert!(!this.flags.contains(Flags::EOF));
 
             // Otherwise, try to read more data and try again. Make sure we've got room
-            let remaining = self.read_buf.capacity() - self.read_buf.len();
+            let remaining = this.read_buf.capacity() - this.read_buf.len();
             if remaining < LW {
-                self.read_buf.reserve(HW - remaining)
+                this.read_buf.reserve(HW - remaining)
             }
-            let cnt = match unsafe {
-                Pin::new_unchecked(&mut self.io).poll_read_buf(cx, &mut self.read_buf)
-            } {
+            let cnt = match this.io.poll_read_buf(cx, &mut this.read_buf) {
                 Poll::Pending => return Poll::Pending,
                 Poll::Ready(Err(e)) => return Poll::Ready(Some(Err(e.into()))),
                 Poll::Ready(Ok(cnt)) => cnt,
             };
 
             if cnt == 0 {
-                self.flags.insert(Flags::EOF);
+                this.flags.insert(Flags::EOF);
             }
-            self.flags.insert(Flags::READABLE);
+            this.flags.insert(Flags::READABLE);
         }
     }
 
     /// Flush write buffer to underlying I/O stream.
-    pub fn flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
+    pub fn flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
     where
         T: AsyncWrite,
         U: Encoder,
     {
+        let mut this = self.as_mut().project();
         log::trace!("flushing framed transport");
 
-        while !self.write_buf.is_empty() {
-            log::trace!("writing; remaining={}", self.write_buf.len());
+        while !this.write_buf.is_empty() {
+            log::trace!("writing; remaining={}", this.write_buf.len());
 
-            let n = ready!(unsafe {
-                Pin::new_unchecked(&mut self.io).poll_write(cx, &self.write_buf)
-            })?;
+            let n = ready!(
+                this.io.as_mut().poll_write(cx, this.write_buf)
+            )?;
 
             if n == 0 {
                 return Poll::Ready(Err(io::Error::new(
@@ -288,26 +290,25 @@ impl<T, U> Framed<T, U> {
             }
 
             // remove written data
-            self.write_buf.advance(n);
+            this.write_buf.advance(n);
         }
 
         // Try flushing the underlying IO
-        ready!(unsafe { Pin::new_unchecked(&mut self.io).poll_flush(cx) })?;
+        ready!(this.io.poll_flush(cx))?;
 
         log::trace!("framed transport flushed");
         Poll::Ready(Ok(()))
     }
 
     /// Flush write buffer and shutdown underlying I/O stream.
-    pub fn close(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
+    pub fn close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
     where
         T: AsyncWrite,
         U: Encoder,
     {
-        unsafe {
-            ready!(Pin::new_unchecked(&mut self.io).poll_flush(cx))?;
-            ready!(Pin::new_unchecked(&mut self.io).poll_shutdown(cx))?;
-        }
+        let mut this = self.as_mut().project();
+        ready!(this.io.as_mut().poll_flush(cx))?;
+        ready!(this.io.as_mut().poll_shutdown(cx))?;
         Poll::Ready(Ok(()))
     }
 }
@@ -319,7 +320,7 @@ where
 {
     type Item = Result<U::Item, U::Error>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         self.next_item(cx)
     }
 }
@@ -341,21 +342,21 @@ where
     }
 
     fn start_send(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         item: <U as Encoder>::Item,
     ) -> Result<(), Self::Error> {
         self.write(item)
     }
 
     fn poll_flush(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), Self::Error>> {
         self.flush(cx)
     }
 
     fn poll_close(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), Self::Error>> {
         self.close(cx)

actix-connect/CHANGES.md

@@ -1,5 +1,23 @@
 # Changes
 
+## [2.0.0-alpha.2] - 2020-03-08
+
+### Changed
+
+* Update `trust-dns-proto` dependency to 0.19. [#116]
+* Update `trust-dns-resolver` dependency to 0.19. [#116]
+* `Address` trait is now required to have static lifetime. [#116]
+* `start_resolver` and `start_default_resolver` are now `async` and may return a `ConnectError`. [#116]
+
+[#116]: https://github.com/actix/actix-net/pull/116
+
+## [2.0.0-alpha.1] - 2020-03-03
+
+### Changed
+
+* Update `rustls` dependency to 0.17
+* Update `tokio-rustls` dependency to 0.13
+
 ## [1.0.2] - 2020-01-15
 
 * Fix actix-service 1.0.3 compatibility

actix-connect/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-connect"
-version = "1.0.2"
+version = "2.0.0-alpha.2"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix connect - tcp connector service"
 keywords = ["network", "framework", "async", "futures"]
@@ -40,16 +40,16 @@ either = "1.5.3"
 futures = "0.3.1"
 http = { version = "0.2.0", optional = true }
 log = "0.4"
-trust-dns-proto = "=0.18.0-alpha.2"
-trust-dns-resolver = "=0.18.0-alpha.2"
+trust-dns-proto = { version = "0.19", default-features = false, features = ["tokio-runtime"] }
+trust-dns-resolver = { version = "0.19", default-features = false, features = ["tokio-runtime", "system-config"] }
 
 # openssl
 open-ssl = { version="0.10", package = "openssl", optional = true }
 tokio-openssl = { version = "0.4.0", optional = true }
 
 # rustls
-rust-tls = { version = "0.16.0", package = "rustls", optional = true }
-tokio-rustls = { version = "0.12.0", optional = true }
+rust-tls = { version = "0.17.0", package = "rustls", optional = true }
+tokio-rustls = { version = "0.13.0", optional = true }
 webpki = { version = "0.21", optional = true }
 
 [dev-dependencies]

actix-connect/src/connect.rs

@@ -6,7 +6,7 @@ use std::net::SocketAddr;
 use either::Either;
 
 /// Connect request
-pub trait Address: Unpin {
+pub trait Address: Unpin + 'static {
     /// Host name of the request
     fn host(&self) -> &str;
 

actix-connect/src/lib.rs

@@ -25,7 +25,7 @@ use actix_rt::{net::TcpStream, Arbiter};
 use actix_service::{pipeline, pipeline_factory, Service, ServiceFactory};
 use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
 use trust_dns_resolver::system_conf::read_system_conf;
-use trust_dns_resolver::AsyncResolver;
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
 
 pub mod resolver {
     pub use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
@@ -39,17 +39,18 @@ pub use self::error::ConnectError;
 pub use self::resolve::{Resolver, ResolverFactory};
 pub use self::service::{ConnectService, ConnectServiceFactory, TcpConnectService};
 
-pub fn start_resolver(cfg: ResolverConfig, opts: ResolverOpts) -> AsyncResolver {
-    let (resolver, bg) = AsyncResolver::new(cfg, opts);
-    actix_rt::spawn(bg);
-    resolver
+pub async fn start_resolver(
+    cfg: ResolverConfig,
+    opts: ResolverOpts,
+) -> Result<AsyncResolver, ConnectError> {
+    Ok(AsyncResolver::tokio(cfg, opts).await?)
 }
 
 struct DefaultResolver(AsyncResolver);
 
-pub(crate) fn get_default_resolver() -> AsyncResolver {
+pub(crate) async fn get_default_resolver() -> Result<AsyncResolver, ConnectError> {
     if Arbiter::contains_item::<DefaultResolver>() {
-        Arbiter::get_item(|item: &DefaultResolver| item.0.clone())
+        Ok(Arbiter::get_item(|item: &DefaultResolver| item.0.clone()))
     } else {
         let (cfg, opts) = match read_system_conf() {
             Ok((cfg, opts)) => (cfg, opts),
@@ -59,16 +60,15 @@ pub(crate) fn get_default_resolver() -> AsyncResolver {
             }
         };
 
-        let (resolver, bg) = AsyncResolver::new(cfg, opts);
-        actix_rt::spawn(bg);
+        let resolver = AsyncResolver::tokio(cfg, opts).await?;
 
         Arbiter::set_item(DefaultResolver(resolver.clone()));
-        resolver
+        Ok(resolver)
     }
 }
 
-pub fn start_default_resolver() -> AsyncResolver {
-    get_default_resolver()
+pub async fn start_default_resolver() -> Result<AsyncResolver, ConnectError> {
+    get_default_resolver().await
 }
 
 /// Create tcp connector service

actix-connect/src/resolve.rs

@@ -6,8 +6,8 @@ use std::task::{Context, Poll};
 
 use actix_service::{Service, ServiceFactory};
 use futures::future::{ok, Either, Ready};
-use trust_dns_resolver::lookup_ip::LookupIpFuture;
-use trust_dns_resolver::{AsyncResolver, Background};
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
+use trust_dns_resolver::{error::ResolveError, lookup_ip::LookupIp};
 
 use crate::connect::{Address, Connect};
 use crate::error::ConnectError;
@@ -106,7 +106,10 @@ impl<T: Address> Service for Resolver<T> {
     type Request = Connect<T>;
     type Response = Connect<T>;
     type Error = ConnectError;
-    type Future = Either<ResolverFuture<T>, Ready<Result<Connect<T>, Self::Error>>>;
+    type Future = Either<
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>,
+        Ready<Result<Connect<T>, Self::Error>>,
+    >;
 
     fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         Poll::Ready(Ok(()))
@@ -119,32 +122,48 @@ impl<T: Address> Service for Resolver<T> {
             req.addr = Some(either::Either::Left(SocketAddr::new(ip, req.port())));
             Either::Right(ok(req))
         } else {
-            trace!("DNS resolver: resolving host {:?}", req.host());
-            if self.resolver.is_none() {
-                self.resolver = Some(get_default_resolver());
-            }
-            Either::Left(ResolverFuture::new(req, self.resolver.as_ref().unwrap()))
+            let resolver = self.resolver.as_ref().map(AsyncResolver::clone);
+            Either::Left(Box::pin(async move {
+                trace!("DNS resolver: resolving host {:?}", req.host());
+                let resolver = if let Some(resolver) = resolver {
+                    resolver
+                } else {
+                    get_default_resolver()
+                        .await
+                        .expect("Failed to get default resolver")
+                };
+                ResolverFuture::new(req, &resolver).await
+            }))
         }
     }
 }
 
+type LookupIpFuture = Pin<Box<dyn Future<Output = Result<LookupIp, ResolveError>>>>;
+
 #[doc(hidden)]
 /// Resolver future
 pub struct ResolverFuture<T: Address> {
     req: Option<Connect<T>>,
-    lookup: Background<LookupIpFuture>,
+    lookup: LookupIpFuture,
 }
 
 impl<T: Address> ResolverFuture<T> {
     pub fn new(req: Connect<T>, resolver: &AsyncResolver) -> Self {
-        let lookup = if let Some(host) = req.host().splitn(2, ':').next() {
-            resolver.lookup_ip(host)
+        let host = if let Some(host) = req.host().splitn(2, ':').next() {
+            host
         } else {
-            resolver.lookup_ip(req.host())
+            req.host()
         };
 
+        // Clone data to be moved to the lookup future
+        let host_clone = host.to_owned();
+        let resolver_clone = resolver.clone();
+
         ResolverFuture {
-            lookup,
+            lookup: Box::pin(async move {
+                let resolver = resolver_clone;
+                resolver.lookup_ip(host_clone).await
+            }),
             req: Some(req),
         }
     }

actix-connect/src/service.rs

@@ -6,7 +6,7 @@ use actix_rt::net::TcpStream;
 use actix_service::{Service, ServiceFactory};
 use either::Either;
 use futures::future::{ok, Ready};
-use trust_dns_resolver::AsyncResolver;
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
 
 use crate::connect::{Address, Connect, Connection};
 use crate::connector::{TcpConnector, TcpConnectorFactory};

actix-connect/src/ssl/openssl.rs

@@ -11,7 +11,7 @@ use actix_codec::{AsyncRead, AsyncWrite};
 use actix_rt::net::TcpStream;
 use actix_service::{Service, ServiceFactory};
 use futures::future::{err, ok, Either, FutureExt, LocalBoxFuture, Ready};
-use trust_dns_resolver::AsyncResolver;
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
 
 use crate::{
     Address, Connect, ConnectError, ConnectService, ConnectServiceFactory, Connection,

actix-connect/tests/test_connect.rs

@@ -14,12 +14,10 @@ use actix_connect::Connect;
 #[actix_rt::test]
 async fn test_string() {
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 
@@ -33,12 +31,10 @@ async fn test_string() {
 #[actix_rt::test]
 async fn test_rustls_string() {
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 
@@ -51,16 +47,14 @@ async fn test_rustls_string() {
 #[actix_rt::test]
 async fn test_static_str() {
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 
-    let resolver = actix_connect::start_default_resolver();
+    let resolver = actix_connect::start_default_resolver().await.unwrap();
     let mut conn = actix_connect::new_connector(resolver.clone());
 
     let con = conn.call(Connect::with("10", srv.addr())).await.unwrap();
@@ -75,17 +69,17 @@ async fn test_static_str() {
 #[actix_rt::test]
 async fn test_new_service() {
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 
     let resolver =
-        actix_connect::start_resolver(ResolverConfig::default(), ResolverOpts::default());
+        actix_connect::start_resolver(ResolverConfig::default(), ResolverOpts::default())
+            .await
+            .unwrap();
 
     let factory = actix_connect::new_connector_factory(resolver);
 
@@ -100,12 +94,10 @@ async fn test_uri() {
     use std::convert::TryFrom;
 
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 
@@ -121,12 +113,10 @@ async fn test_rustls_uri() {
     use std::convert::TryFrom;
 
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 

actix-ioframe/Cargo.toml

@@ -27,5 +27,5 @@ pin-project = "0.4.6"
 log = "0.4"
 
 [dev-dependencies]
-actix-connect = "1.0.0"
+actix-connect = "2.0.0-alpha.2"
 actix-testing = "1.0.0"

actix-ioframe/src/connect.rs

@@ -42,6 +42,7 @@ where
 pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Out> {
     pub(crate) state: St,
     pub(crate) out: Option<Out>,
+    #[pin]
     pub(crate) framed: Framed<Io, Codec>,
 }
 
@@ -97,8 +98,8 @@ where
 {
     type Error = <Codec as Encoder>::Error;
 
-    fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        if self.framed.is_write_ready() {
+    fn poll_ready(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        if self.as_mut().project().framed.is_write_ready() {
             Poll::Ready(Ok(()))
         } else {
             Poll::Pending
@@ -112,11 +113,11 @@ where
         self.project().framed.write(item)
     }
 
-    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.get_mut().framed.flush(cx)
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.as_mut().project().framed.flush(cx)
     }
 
-    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.get_mut().framed.close(cx)
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.as_mut().project().framed.close(cx)
     }
 }

actix-ioframe/src/dispatcher.rs

@@ -6,6 +6,7 @@ use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::Service;
 use actix_utils::mpsc;
 use futures::Stream;
+use pin_project::pin_project;
 use log::debug;
 
 use crate::error::ServiceError;
@@ -15,6 +16,7 @@ type Response<U> = <U as Encoder>::Item;
 
 /// FramedTransport - is a future that reads frames from Framed object
 /// and pass then to the service.
+#[pin_project]
 pub(crate) struct Dispatcher<S, T, U, Out>
 where
     S: Service<Request = Request<U>, Response = Option<Response<U>>>,
@@ -29,6 +31,7 @@ where
     service: S,
     sink: Option<Out>,
     state: FramedState<S, U>,
+    #[pin]
     framed: Framed<T, U>,
     rx: mpsc::Receiver<Result<<U as Encoder>::Item, S::Error>>,
 }
@@ -90,26 +93,27 @@ where
     <U as Encoder>::Error: std::fmt::Debug,
     Out: Stream<Item = <U as Encoder>::Item> + Unpin,
 {
-    fn poll_read(&mut self, cx: &mut Context<'_>) -> bool {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool {
         loop {
-            match self.service.poll_ready(cx) {
+            let this = self.as_mut().project();
+            match this.service.poll_ready(cx) {
                 Poll::Ready(Ok(_)) => {
-                    let item = match self.framed.next_item(cx) {
+                    let item = match this.framed.next_item(cx) {
                         Poll::Ready(Some(Ok(el))) => el,
                         Poll::Ready(Some(Err(err))) => {
-                            self.state = FramedState::FramedError(ServiceError::Decoder(err));
+                            *this.state = FramedState::FramedError(ServiceError::Decoder(err));
                             return true;
                         }
                         Poll::Pending => return false,
                         Poll::Ready(None) => {
                             log::trace!("Client disconnected");
-                            self.state = FramedState::Stopping;
+                            *this.state = FramedState::Stopping;
                             return true;
                         }
                     };
 
-                    let tx = self.rx.sender();
-                    let fut = self.service.call(item);
+                    let tx = this.rx.sender();
+                    let fut = this.service.call(item);
                     actix_rt::spawn(async move {
                         let item = fut.await;
                         let item = match item {
@@ -122,7 +126,7 @@ where
                 }
                 Poll::Pending => return false,
                 Poll::Ready(Err(err)) => {
-                    self.state = FramedState::Error(ServiceError::Service(err));
+                    *this.state = FramedState::Error(ServiceError::Service(err));
                     return true;
                 }
             }
@@ -130,37 +134,38 @@ where
     }
 
     /// write to framed object
-    fn poll_write(&mut self, cx: &mut Context<'_>) -> bool {
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool {
         loop {
-            while !self.framed.is_write_buf_full() {
-                match Pin::new(&mut self.rx).poll_next(cx) {
+            let mut this = self.as_mut().project();
+            while !this.framed.is_write_buf_full() {
+                match Pin::new(&mut this.rx).poll_next(cx) {
                     Poll::Ready(Some(Ok(msg))) => {
-                        if let Err(err) = self.framed.write(msg) {
-                            self.state = FramedState::FramedError(ServiceError::Encoder(err));
+                        if let Err(err) = this.framed.as_mut().write(msg) {
+                            *this.state = FramedState::FramedError(ServiceError::Encoder(err));
                             return true;
                         }
                         continue;
                     }
                     Poll::Ready(Some(Err(err))) => {
-                        self.state = FramedState::Error(ServiceError::Service(err));
+                        *this.state = FramedState::Error(ServiceError::Service(err));
                         return true;
                     }
                     Poll::Ready(None) | Poll::Pending => (),
                 }
 
-                if self.sink.is_some() {
-                    match Pin::new(self.sink.as_mut().unwrap()).poll_next(cx) {
+                if this.sink.is_some() {
+                    match Pin::new(this.sink.as_mut().unwrap()).poll_next(cx) {
                         Poll::Ready(Some(msg)) => {
-                            if let Err(err) = self.framed.write(msg) {
-                                self.state =
+                            if let Err(err) = this.framed.as_mut().write(msg) {
+                                *this.state =
                                     FramedState::FramedError(ServiceError::Encoder(err));
                                 return true;
                             }
                             continue;
                         }
                         Poll::Ready(None) => {
-                            let _ = self.sink.take();
-                            self.state = FramedState::FlushAndStop;
+                            let _ = this.sink.take();
+                            *this.state = FramedState::FlushAndStop;
                             return true;
                         }
                         Poll::Pending => (),
@@ -169,13 +174,13 @@ where
                 break;
             }
 
-            if !self.framed.is_write_buf_empty() {
-                match self.framed.flush(cx) {
+            if !this.framed.is_write_buf_empty() {
+                match this.framed.as_mut().flush(cx) {
                     Poll::Pending => break,
                     Poll::Ready(Ok(_)) => (),
                     Poll::Ready(Err(err)) => {
                         debug!("Error sending data: {:?}", err);
-                        self.state = FramedState::FramedError(ServiceError::Encoder(err));
+                        *this.state = FramedState::FramedError(ServiceError::Encoder(err));
                         return true;
                     }
                 }
@@ -187,13 +192,14 @@ where
     }
 
     pub(crate) fn poll(
-        &mut self,
+        mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), ServiceError<S::Error, U>>> {
-        match self.state {
+        let mut this = self.as_mut().project();
+        match this.state {
             FramedState::Processing => loop {
-                let read = self.poll_read(cx);
-                let write = self.poll_write(cx);
+                let read = self.as_mut().poll_read(cx);
+                let write = self.as_mut().poll_write(cx);
                 if read || write {
                     continue;
                 } else {
@@ -202,18 +208,18 @@ where
             },
             FramedState::Error(_) => {
                 // flush write buffer
-                if !self.framed.is_write_buf_empty() {
-                    if let Poll::Pending = self.framed.flush(cx) {
+                if !this.framed.is_write_buf_empty() {
+                    if let Poll::Pending = this.framed.flush(cx) {
                         return Poll::Pending;
                     }
                 }
-                Poll::Ready(Err(self.state.take_error()))
+                Poll::Ready(Err(this.state.take_error()))
             }
             FramedState::FlushAndStop => {
                 // drain service responses
-                match Pin::new(&mut self.rx).poll_next(cx) {
+                match Pin::new(this.rx).poll_next(cx) {
                     Poll::Ready(Some(Ok(msg))) => {
-                        if self.framed.write(msg).is_err() {
+                        if this.framed.as_mut().write(msg).is_err() {
                             return Poll::Ready(Ok(()));
                         }
                     }
@@ -222,8 +228,8 @@ where
                 }
 
                 // flush io
-                if !self.framed.is_write_buf_empty() {
-                    match self.framed.flush(cx) {
+                if !this.framed.is_write_buf_empty() {
+                    match this.framed.flush(cx) {
                         Poll::Ready(Err(err)) => {
                             debug!("Error sending data: {:?}", err);
                         }
@@ -235,7 +241,7 @@ where
                 };
                 Poll::Ready(Ok(()))
             }
-            FramedState::FramedError(_) => Poll::Ready(Err(self.state.take_framed_error())),
+            FramedState::FramedError(_) => Poll::Ready(Err(this.state.take_framed_error())),
             FramedState::Stopping => Poll::Ready(Ok(())),
         }
     }

actix-ioframe/src/service.rs

@@ -357,7 +357,7 @@ where
 {
     Connect(#[pin] C::Future, Rc<T>),
     Handler(#[pin] T::Future, Option<Framed<Io, Codec>>, Option<Out>),
-    Dispatcher(Dispatcher<T::Service, Io, Codec, Out>),
+    Dispatcher(#[pin] Dispatcher<T::Service, Io, Codec, Out>),
 }
 
 impl<St, Io, Codec, Out, C, T> FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>
@@ -408,7 +408,7 @@ where
                     Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
                 }
             }
-            FramedServiceImplResponseInner::Dispatcher(ref mut fut) => {
+            FramedServiceImplResponseInner::Dispatcher(fut) => {
                 Either::Right(fut.poll(cx))
             }
         }

actix-macros/Cargo.toml

@@ -14,7 +14,7 @@ workspace = ".."
 proc-macro = true
 
 [dependencies]
-quote = "^1"
+quote = "1.0.3"
 syn = { version = "^1", features = ["full"] }
 
 [dev-dependencies]

actix-rt/CHANGES.md

@@ -4,6 +4,8 @@
 
 - Expose `System::is_set` to check if current system is running
 
+- Add `Arbiter::local_join` associated function to get be able to `await` for spawned futures
+
 ## [1.0.0] - 2019-12-11
 
 * Update dependencies

actix-rt/Cargo.toml

@@ -18,6 +18,7 @@ path = "src/lib.rs"
 [dependencies]
 actix-macros = "0.1.0"
 actix-threadpool = "0.3"
-futures = "0.3.1"
+futures-channel = { version = "0.3.1", default-features = false }
+futures-util = { version = "0.3.1", default-features = false }
 copyless = "0.1.4"
-tokio = { version = "0.2.6", default-features=false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] }
+tokio = { version = "0.2.6", default-features = false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] }

actix-rt/src/arbiter.rs

@@ -6,19 +6,22 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::task::{Context, Poll};
 use std::{fmt, thread};
 
-use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
-use futures::channel::oneshot::{channel, Canceled, Sender};
-use futures::{future, Future, FutureExt, Stream};
+use futures_channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
+use futures_channel::oneshot::{channel, Canceled, Sender};
+use futures_util::{future::{self, Future, FutureExt}, stream::Stream};
 
 use crate::runtime::Runtime;
 use crate::system::System;
 
 use copyless::BoxHelper;
 
+pub use tokio::task::JoinHandle;
+
 thread_local!(
     static ADDR: RefCell<Option<Arbiter>> = RefCell::new(None);
     static RUNNING: Cell<bool> = Cell::new(false);
     static Q: RefCell<Vec<Pin<Box<dyn Future<Output = ()>>>>> = RefCell::new(Vec::new());
+    static PENDING: RefCell<Vec<JoinHandle<()>>> = RefCell::new(Vec::new());
     static STORAGE: RefCell<HashMap<TypeId, Box<dyn Any>>> = RefCell::new(HashMap::new());
 );
 
@@ -170,13 +173,15 @@ impl Arbiter {
         RUNNING.with(move |cell| {
             if cell.get() {
                 // Spawn the future on running executor
-                tokio::task::spawn_local(future);
+                PENDING.with(move |cell| {
+                    cell.borrow_mut().push(tokio::task::spawn_local(future));
+                })
             } else {
                 // Box the future and push it to the queue, this results in double boxing
                 // because the executor boxes the future again, but works for now
                 Q.with(move |cell| {
                     cell.borrow_mut()
-                        .push(unsafe { Pin::new_unchecked(Box::alloc().init(future)) })
+                        .push(Pin::from(Box::alloc().init(future)))
                 });
             }
         });
@@ -294,6 +299,15 @@ impl Arbiter {
             Ok(())
         }
     }
+
+    /// Returns a future that will be completed once all currently spawned futures
+    /// have completed.
+    pub fn local_join() -> impl Future<Output = ()> {
+        PENDING.with(move |cell| {
+            let current = cell.replace(Vec::new());
+            future::join_all(current).map(|_| ())
+        })
+    }
 }
 
 struct ArbiterController {
@@ -329,7 +343,9 @@ impl Future for ArbiterController {
                         return Poll::Ready(());
                     }
                     ArbiterCommand::Execute(fut) => {
-                        tokio::task::spawn_local(fut);
+                        PENDING.with(move |cell| {
+                            cell.borrow_mut().push(tokio::task::spawn_local(fut));
+                        });
                     }
                     ArbiterCommand::ExecuteFn(f) => {
                         f.call_box();

actix-rt/src/builder.rs

@@ -1,9 +1,9 @@
 use std::borrow::Cow;
 use std::io;
 
-use futures::channel::mpsc::unbounded;
-use futures::channel::oneshot::{channel, Receiver};
-use futures::future::{lazy, Future, FutureExt};
+use futures_channel::mpsc::unbounded;
+use futures_channel::oneshot::{channel, Receiver};
+use futures_util::future::{lazy, Future, FutureExt};
 use tokio::task::LocalSet;
 
 use crate::arbiter::{Arbiter, SystemArbiter};

actix-rt/src/lib.rs

@@ -25,7 +25,7 @@ pub use actix_threadpool as blocking;
 /// This function panics if actix system is not running.
 pub fn spawn<F>(f: F)
 where
-    F: futures::Future<Output = ()> + 'static,
+    F: futures_util::future::Future<Output = ()> + 'static,
 {
     if !System::is_set() {
         panic!("System is not running");

actix-rt/src/system.rs

@@ -3,7 +3,7 @@ use std::future::Future;
 use std::io;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
-use futures::channel::mpsc::UnboundedSender;
+use futures_channel::mpsc::UnboundedSender;
 use tokio::task::LocalSet;
 
 use crate::arbiter::{Arbiter, SystemCommand};

actix-rt/tests/wait_spawned.rs

@@ -0,0 +1,100 @@
+use std::time::{Duration, Instant};
+
+#[test]
+fn await_for_timer() {
+    let time = Duration::from_secs(2);
+    let instant = Instant::now();
+    actix_rt::System::new("test_wait_timer").block_on(async move {
+        tokio::time::delay_for(time).await;
+    });
+    assert!(
+        instant.elapsed() >= time,
+        "Block on should poll awaited future to completion"
+    );
+}
+
+#[test]
+fn join_another_arbiter() {
+    let time = Duration::from_secs(2);
+    let instant = Instant::now();
+    actix_rt::System::new("test_join_another_arbiter").block_on(async move {
+        let mut arbiter = actix_rt::Arbiter::new();
+        arbiter.send(Box::pin(async move {
+            tokio::time::delay_for(time).await;
+            actix_rt::Arbiter::current().stop();
+        }));
+        arbiter.join().unwrap();
+    });
+    assert!(
+        instant.elapsed() >= time,
+        "Join on another arbiter should complete only when it calls stop"
+    );
+
+    let instant = Instant::now();
+    actix_rt::System::new("test_join_another_arbiter").block_on(async move {
+        let mut arbiter = actix_rt::Arbiter::new();
+        arbiter.exec_fn(move || {
+            actix_rt::spawn(async move {
+                tokio::time::delay_for(time).await;
+                actix_rt::Arbiter::current().stop();
+            });
+        });
+        arbiter.join().unwrap();
+    });
+    assert!(
+        instant.elapsed() >= time,
+        "Join on a arbiter that has used actix_rt::spawn should wait for said future"
+    );
+
+    let instant = Instant::now();
+    actix_rt::System::new("test_join_another_arbiter").block_on(async move {
+        let mut arbiter = actix_rt::Arbiter::new();
+        arbiter.send(Box::pin(async move {
+            tokio::time::delay_for(time).await;
+            actix_rt::Arbiter::current().stop();
+        }));
+        arbiter.stop();
+        arbiter.join().unwrap();
+    });
+    assert!(
+        instant.elapsed() < time,
+        "Premature stop of arbiter should conclude regardless of it's current state"
+    );
+}
+
+#[test]
+fn join_current_arbiter() {
+    let time = Duration::from_secs(2);
+
+    let instant = Instant::now();
+    actix_rt::System::new("test_join_current_arbiter").block_on(async move {
+        actix_rt::spawn(async move {
+            tokio::time::delay_for(time).await;
+            actix_rt::Arbiter::current().stop();
+        });
+        actix_rt::Arbiter::local_join().await;
+    });
+    assert!(
+        instant.elapsed() >= time,
+        "Join on current arbiter should wait for all spawned futures"
+    );
+
+    let large_timer = Duration::from_secs(20);
+    let instant = Instant::now();
+    actix_rt::System::new("test_join_current_arbiter").block_on(async move {
+        actix_rt::spawn(async move {
+            tokio::time::delay_for(time).await;
+            actix_rt::Arbiter::current().stop();
+        });
+        let f = actix_rt::Arbiter::local_join();
+        actix_rt::spawn(async move {
+            tokio::time::delay_for(large_timer).await;
+            actix_rt::Arbiter::current().stop();
+        });
+        f.await;
+    });
+    assert!(
+        instant.elapsed() < large_timer,
+        "local_join should await only for the already spawned futures"
+    );
+}

actix-tls/CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [2.0.0-alpha.1] - 2020-03-03
+
+### Changed
+
+* Update `rustls` dependency to 0.17
+* Update `tokio-rustls` dependency to 0.13
+* Update `webpki-roots` dependency to 0.19
+
 ## [1.0.0] - 2019-12-11
 
 * 1.0.0 release

actix-tls/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-tls"
-version = "1.0.0"
+version = "2.0.0-alpha.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix tls services"
 keywords = ["network", "framework", "async", "futures"]
@@ -46,10 +46,10 @@ open-ssl = { version="0.10", package = "openssl", optional = true }
 tokio-openssl = { version = "0.4.0", optional = true }
 
 # rustls
-rust-tls = { version = "0.16.0", package = "rustls", optional = true }
+rust-tls = { version = "0.17.0", package = "rustls", optional = true }
 webpki = { version = "0.21", optional = true }
-webpki-roots = { version = "0.17", optional = true }
-tokio-rustls = { version = "0.12.0", optional = true }
+webpki-roots = { version = "0.19", optional = true }
+tokio-rustls = { version = "0.13.0", optional = true }
 
 # native-tls
 native-tls = { version="0.2", optional = true }

actix-utils/src/framed.rs

@@ -77,6 +77,7 @@ where
 {
     service: S,
     state: State<S, U>,
+    #[pin]
     framed: Framed<T, U>,
     rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
     tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
@@ -169,7 +170,7 @@ where
         &mut self.framed
     }
 
-    fn poll_read(&mut self, cx: &mut Context<'_>) -> bool
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
     where
         S: Service<Request = Request<U>, Response = Response<U>>,
         S::Error: 'static,
@@ -180,29 +181,30 @@ where
         <U as Encoder>::Error: std::fmt::Debug,
     {
         loop {
-            match self.service.poll_ready(cx) {
+            let this = self.as_mut().project();
+            match this.service.poll_ready(cx) {
                 Poll::Ready(Ok(_)) => {
-                    let item = match self.framed.next_item(cx) {
+                    let item = match this.framed.next_item(cx) {
                         Poll::Ready(Some(Ok(el))) => el,
                         Poll::Ready(Some(Err(err))) => {
-                            self.state = State::FramedError(DispatcherError::Decoder(err));
+                            *this.state = State::FramedError(DispatcherError::Decoder(err));
                             return true;
                         }
                         Poll::Pending => return false,
                         Poll::Ready(None) => {
-                            self.state = State::Stopping;
+                            *this.state = State::Stopping;
                             return true;
                         }
                     };
 
-                    let tx = self.tx.clone();
-                    actix_rt::spawn(self.service.call(item).map(move |item| {
+                    let tx = this.tx.clone();
+                    actix_rt::spawn(this.service.call(item).map(move |item| {
                         let _ = tx.send(item.map(Message::Item));
                     }));
                 }
                 Poll::Pending => return false,
                 Poll::Ready(Err(err)) => {
-                    self.state = State::Error(DispatcherError::Service(err));
+                    *this.state = State::Error(DispatcherError::Service(err));
                     return true;
                 }
             }
@@ -210,7 +212,7 @@ where
     }
 
     /// write to framed object
-    fn poll_write(&mut self, cx: &mut Context<'_>) -> bool
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
     where
         S: Service<Request = Request<U>, Response = Response<U>>,
         S::Error: 'static,
@@ -221,33 +223,34 @@ where
         <U as Encoder>::Error: std::fmt::Debug,
     {
         loop {
-            while !self.framed.is_write_buf_full() {
-                match Pin::new(&mut self.rx).poll_next(cx) {
+            let mut this = self.as_mut().project();
+            while !this.framed.is_write_buf_full() {
+                match Pin::new(&mut this.rx).poll_next(cx) {
                     Poll::Ready(Some(Ok(Message::Item(msg)))) => {
-                        if let Err(err) = self.framed.write(msg) {
-                            self.state = State::FramedError(DispatcherError::Encoder(err));
+                        if let Err(err) = this.framed.as_mut().write(msg) {
+                            *this.state = State::FramedError(DispatcherError::Encoder(err));
                             return true;
                         }
                     }
                     Poll::Ready(Some(Ok(Message::Close))) => {
-                        self.state = State::FlushAndStop;
+                        *this.state = State::FlushAndStop;
                         return true;
                     }
                     Poll::Ready(Some(Err(err))) => {
-                        self.state = State::Error(DispatcherError::Service(err));
+                        *this.state = State::Error(DispatcherError::Service(err));
                         return true;
                     }
                     Poll::Ready(None) | Poll::Pending => break,
                 }
             }
 
-            if !self.framed.is_write_buf_empty() {
-                match self.framed.flush(cx) {
+            if !this.framed.is_write_buf_empty() {
+                match this.framed.flush(cx) {
                     Poll::Pending => break,
                     Poll::Ready(Ok(_)) => (),
                     Poll::Ready(Err(err)) => {
                         debug!("Error sending data: {:?}", err);
-                        self.state = State::FramedError(DispatcherError::Encoder(err));
+                        *this.state = State::FramedError(DispatcherError::Encoder(err));
                         return true;
                     }
                 }
@@ -279,7 +282,7 @@ where
 
             return match this.state {
                 State::Processing => {
-                    if self.poll_read(cx) || self.poll_write(cx) {
+                    if self.as_mut().poll_read(cx) || self.as_mut().poll_write(cx) {
                         continue;
                     } else {
                         Poll::Pending
@@ -287,12 +290,12 @@ where
                 }
                 State::Error(_) => {
                     // flush write buffer
-                    if !self.framed.is_write_buf_empty() {
-                        if let Poll::Pending = self.framed.flush(cx) {
+                    if !this.framed.is_write_buf_empty() {
+                        if let Poll::Pending = this.framed.flush(cx) {
                             return Poll::Pending;
                         }
                     }
-                    Poll::Ready(Err(self.state.take_error()))
+                    Poll::Ready(Err(this.state.take_error()))
                 }
                 State::FlushAndStop => {
                     if !this.framed.is_write_buf_empty() {