Implement request cancellation

1 files changed, 17 insertions, 1 deletions

diff --git a/src/recv.rs b/src/recv.rs
index ac93c4b..8909190 100644
--- a/src/recv.rs
+++ b/src/recv.rs
@@ -53,6 +53,7 @@ impl Drop for Sender {
 #[async_trait]
 pub(crate) trait RecvLoop: Sync + 'static {
 	fn recv_handler(self: &Arc<Self>, id: RequestID, stream: ByteStream);
+	fn cancel_handler(self: &Arc<Self>, _id: RequestID) {}
 
 	async fn recv_loop<R>(self: Arc<Self>, mut read: R, debug_name: String) -> Result<(), Error>
 	where
@@ -78,6 +79,18 @@ pub(crate) trait RecvLoop: Sync + 'static {
 			read.read_exact(&mut header_size[..]).await?;
 			let size = ChunkLength::from_be_bytes(header_size);
 
+			if size == CANCEL_REQUEST {
+				if let Some(mut stream) = streams.remove(&id) {
+					let _ = stream.send(Err(std::io::Error::new(
+						std::io::ErrorKind::Other,
+						"netapp: cancel requested",
+					)));
+					stream.end();
+				}
+				self.cancel_handler(id);
+				continue;
+			}
+
 			let has_cont = (size & CHUNK_HAS_CONTINUATION) != 0;
 			let is_error = (size & ERROR_MARKER) != 0;
 			let size = (size & CHUNK_LENGTH_MASK) as usize;
@@ -88,7 +101,10 @@ pub(crate) trait RecvLoop: Sync + 'static {
 				let kind = u8_to_io_errorkind(next_slice[0]);
 				let msg =
 					std::str::from_utf8(&next_slice[1..]).unwrap_or("<invalid utf8 error message>");
-				debug!("recv_loop({}): got id {}, error {:?}: {}", debug_name, id, kind, msg);
+				debug!(
+					"recv_loop({}): got id {}, error {:?}: {}",
+					debug_name, id, kind, msg
+				);
 				Some(Err(std::io::Error::new(kind, msg.to_string())))
 			} else {
 				trace!(