Implement close() from Sink (#492)

core/src/transport/memory.rs

@@ -177,6 +177,11 @@ impl<T> Sink for Chan<T> {
     fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
         self.outgoing.poll_complete().map_err(|_| io::ErrorKind::ConnectionReset.into())
     }
+
+    #[inline]
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.outgoing.close().map_err(|_| io::ErrorKind::ConnectionReset.into())
+    }
 }
 
 impl<T: IntoBuf> Into<RwStreamSink<Chan<T>>> for Chan<T> {

misc/multistream-select/src/length_delimited.rs

@@ -211,6 +211,11 @@ where
     fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
         self.inner.poll_complete()
     }
+
+    #[inline]
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.close()
+    }
 }
 
 fn decode_length_prefix(buf: &[u8]) -> u16 {

misc/multistream-select/src/protocol/dialer.rs

@@ -106,6 +106,11 @@ where
     fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
         Ok(self.inner.poll_complete()?)
     }
+
+    #[inline]
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        Ok(self.inner.close()?)
+    }
 }
 
 impl<R> Stream for Dialer<R>

misc/multistream-select/src/protocol/listener.rs

@@ -132,6 +132,11 @@ where
     fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
         Ok(self.inner.poll_complete()?)
     }
+
+    #[inline]
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        Ok(self.inner.close()?)
+    }
 }
 
 impl<R> Stream for Listener<R>

misc/rw-stream-sink/src/lib.rs

@@ -196,6 +196,9 @@ mod tests {
         fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
             self.1.poll_complete()
         }
+        fn close(&mut self) -> Poll<(), Self::SinkError> {
+            self.1.close()
+        }
     }
 
     #[test]

protocols/kad/src/kad_server.rs

@@ -438,6 +438,9 @@ mod tests {
         fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
             self.1.poll_complete()
         }
+        fn close(&mut self) -> Poll<(), Self::SinkError> {
+            self.1.close()
+        }
     }
 
     fn build_test() -> (KadConnecController, impl Stream<Item = KadIncomingRequest, Error = IoError>, KadConnecController, impl Stream<Item = KadIncomingRequest, Error = IoError>) {

protocols/secio/src/codec/decode.rs

@@ -125,4 +125,9 @@ where
     fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
         self.raw_stream.poll_complete()
     }
+
+    #[inline]
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.raw_stream.close()
+    }
 }

protocols/secio/src/codec/encode.rs

@@ -63,7 +63,6 @@ where
     type SinkError = S::SinkError;
 
     fn start_send(&mut self, mut data_buf: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
-
         // TODO if SinkError gets refactor to SecioError,
         // then use try_apply_keystream
         self.cipher_state.apply_keystream(&mut data_buf[..]);
@@ -77,6 +76,11 @@ where
     fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
         self.raw_sink.poll_complete()
     }
+
+    #[inline]
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.raw_sink.close()
+    }
 }
 
 impl<S> Stream for EncoderMiddleware<S>

protocols/secio/src/lib.rs

@@ -436,6 +436,11 @@ where
     fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
         self.inner.poll_complete()
     }
+
+    #[inline]
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        self.inner.close()
+    }
 }
 
 impl<S> Stream for SecioMiddleware<S>