Repository: paritytech/substrate
File: client/network/src/protocol/notifications/upgrade/notifications.rs (Rust)

Code before:
	pub negotiated_fallback: Option<ProtocolName>,
	/// Implementation of `Sink` that allows sending messages on the substream.
	pub substream: NotificationsOutSubstream<TSubstream>,
}

impl<TSubstream> Sink<Vec<u8>> for NotificationsOutSubstream<TSubstream>
where
	TSubstream: AsyncRead + AsyncWrite + Unpin,
{
	type Error = NotificationsOutError;

	fn poll_ready(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
		let mut this = self.project();
		Sink::poll_ready(this.socket.as_mut(), cx).map_err(NotificationsOutError::Io)
	}

	fn start_send(self: Pin<&mut Self>, item: Vec<u8>) -> Result<(), Self::Error> {
		let mut this = self.project();
		Sink::start_send(this.socket.as_mut(), io::Cursor::new(item))
			.map_err(NotificationsOutError::Io)
	}

	fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
		let mut this = self.project();

		// `Sink::poll_flush` does not expose stream closed error untill we write something into
		// the stream, so the code below makes sure we detect that the substream was closed
		// even if we don't write anything into it.
		match Stream::poll_next(this.socket.as_mut(), cx) {
			Poll::Pending => {},
			Poll::Ready(Some(_)) => {
				error!(
					target: "sub-libp2p",
					"Unexpected incoming data in `NotificationsOutSubstream`",
				);
			},
			Poll::Ready(None) =>
				return Poll::Ready(Err(NotificationsOutError::Io(io::Error::new(
					io::ErrorKind::ConnectionAborted,
					"substream was closed by the remote",
				)))),
		}

		Sink::poll_flush(this.socket.as_mut(), cx).map_err(NotificationsOutError::Io)
	}

	fn poll_close(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
		let mut this = self.project();
		Sink::poll_close(this.socket.as_mut(), cx).map_err(NotificationsOutError::Io)
	}
}

Review comment (suggestion):
```suggestion
		// `Sink::poll_flush` does not expose stream closed error until we write something into
```

Code after:
	pub substream: NotificationsOutSubstream<TSubstream>,
}

impl<TSubstream> Sink<Vec<u8>> for NotificationsOutSubstream<TSubstream>
where
	TSubstream: AsyncRead + AsyncWrite + Unpin,
{
	type Error = NotificationsOutError;

	fn poll_ready(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
		let mut this = self.project();
		Sink::poll_ready(this.socket.as_mut(), cx).map_err(NotificationsOutError::Io)
	}

	fn start_send(self: Pin<&mut Self>, item: Vec<u8>) -> Result<(), Self::Error> {
		let mut this = self.project();
		Sink::start_send(this.socket.as_mut(), io::Cursor::new(item))
			.map_err(NotificationsOutError::Io)
	}

	fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
		let mut this = self.project();

		// `Sink::poll_flush` does not expose stream closed error until we write something into
		// the stream, so the code below makes sure we detect that the substream was closed
		// even if we don't write anything into it.
		match Stream::poll_next(this.socket.as_mut(), cx) {
			Poll::Pending => {},
			Poll::Ready(Some(_)) => {
				error!(
					target: "sub-libp2p",
					"Unexpected incoming data in `NotificationsOutSubstream`",
				);
			},
			Poll::Ready(None) => return Poll::Ready(Err(NotificationsOutError::Terminated)),
		}

		Sink::poll_flush(this.socket.as_mut(), cx).map_err(NotificationsOutError::Io)
	}

	fn poll_close(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
		let mut this = self.project();
		Sink::poll_close(this.socket.as_mut(), cx).map_err(NotificationsOutError::Io)
	}
}

/// Error generated by sending on a notifications out substream.
#[derive(Debug, thiserror::Error)]
pub enum NotificationsHandshakeError {
	/// I/O error on the substream.
	#[error(transparent)]