src/endpoint.rs

@@ -52,7 +52,10 @@ where
 	T: EndpointHandler<M>,
 	M: Message + 'static,
 {
-	async fn handle(self: &Arc<Self>, m: Req<M>, from: NodeID) -> Resp<M> {
+	async fn handle(self: &Arc<Self>, mut m: Req<M>, from: NodeID) -> Resp<M> {
+		// Immediately drop stream to avoid backpressure if a stream was sent
+		// (this will make all data sent to the stream be ignored immediately)
+		drop(m.take_stream());
 		Resp::new(EndpointHandler::handle(self, m.msg(), from).await)
 	}
 }

src/message.rs

@@ -81,6 +81,10 @@ impl<M: Message> Req<M> {
 		}
 	}
 
+	pub fn take_stream(&mut self) -> Option<ByteStream> {
+		std::mem::replace(&mut self.body, BodyData::None).into_stream()
+	}
+
 	pub(crate) fn into_enc(
 		self,
 		prio: RequestPriority,