Don't call poll_complete once Sink01::close has been called

Author: LucioFranco

futures-util/src/compat/compat01as03.rs

@@ -9,7 +9,6 @@ use futures_01::{
 use futures_core::{task as task03, Future as Future03, Stream as Stream03};
 use std::pin::Pin;
 use futures_sink::Sink as Sink03;
-use std::pin::{Pin, Unpin};
 use std::task::LocalWaker;
 
 /// Converts a futures 0.1 Future, Stream, AsyncRead, or AsyncWrite
@@ -127,6 +126,7 @@ impl<St: Stream01> Stream03 for Compat01As03<St> {
 pub struct Compat01As03Sink<S, SinkItem> {
     pub(crate) inner: Spawn01<S>,
     pub(crate) buffer: Option<SinkItem>,
+    pub(crate) close_started: bool,
 }
 
 impl<S, SinkItem> Unpin for Compat01As03Sink<S, SinkItem> {}
@@ -137,6 +137,7 @@ impl<S, SinkItem> Compat01As03Sink<S, SinkItem> {
         Compat01As03Sink {
             inner: spawn01(inner),
             buffer: None,
+            close_started: false
         }
     }
 
@@ -231,39 +232,49 @@ where
         lw: &LocalWaker,
     ) -> task03::Poll<Result<(), Self::SinkError>> {
         let item = self.buffer.take();
+        let close_started = self.close_started;
+
         match self.in_notify(lw, |f| match item {
             Some(i) => match f.start_send(i) {
                 Ok(AsyncSink01::Ready) => {
                     match f.poll_complete() {
                         Ok(Async01::Ready(_)) => {
                             match <S as Sink01>::close(f) {
-                                Ok(i) => Ok((i, None)),
+                                Ok(i) => Ok((i, None, true)),
                                 Err(e) => Err(e)
                             }
                         },
-                        Ok(Async01::NotReady) => Ok((Async01::NotReady, None)),
+                        Ok(Async01::NotReady) => Ok((Async01::NotReady, None, false)),
                         Err(e) => Err(e)
                     }
                 },
                 Ok(AsyncSink01::NotReady(t)) => {
-                    Ok((Async01::NotReady, Some(t)))
+                    Ok((Async01::NotReady, Some(t), close_started))
                 }
                 Err(e) => Err(e),
             },
-            None => match f.poll_complete() {
-                Ok(Async01::Ready(_)) => {
-                    match <S as Sink01>::close(f) {
-                        Ok(i) => Ok((i, None)),
-                        Err(e) => Err(e)
-                    }
-                },
-                Ok(Async01::NotReady) => Ok((Async01::NotReady, None)),
-                Err(e) => Err(e)
+            None => if close_started {
+                match <S as Sink01>::close(f) {
+                    Ok(i) => Ok((i, None, true)),
+                    Err(e) => Err(e)
+                }
+            } else {
+                match f.poll_complete() {
+                    Ok(Async01::Ready(_)) => {
+                        match <S as Sink01>::close(f) {
+                            Ok(i) => Ok((i, None, true)),
+                            Err(e) => Err(e)
+                        }
+                    },
+                    Ok(Async01::NotReady) => Ok((Async01::NotReady, None, close_started)),
+                    Err(e) => Err(e)
+                }
             },
         }) {
-            Ok((Async01::Ready(_), _)) => task03::Poll::Ready(Ok(())),
-            Ok((Async01::NotReady, item)) => {
+            Ok((Async01::Ready(_), _, _)) => task03::Poll::Ready(Ok(())),
+            Ok((Async01::NotReady, item, close_started)) => {
                 self.buffer = item;
+                self.close_started = close_started;
                 task03::Poll::Pending
             }
             Err(e) => task03::Poll::Ready(Err(e)),