PR review: added cancellation safety note and id()

Author: rklaehn

src/util/stream.rs

@@ -13,6 +13,8 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 /// An abstract `iroh::endpoint::SendStream`.
 pub trait SendStream: Send {
     /// Send bytes to the stream. This takes a `Bytes` because iroh can directly use them.
+    ///
+    /// This method is not cancellation safe. Even if this does not resolve, some bytes may have been written when previously polled.
     fn send_bytes(&mut self, bytes: Bytes) -> impl Future<Output = io::Result<()>> + Send;
     /// Send that sends a fixed sized buffer.
     fn send<const L: usize>(
@@ -25,6 +27,8 @@ pub trait SendStream: Send {
     fn reset(&mut self, code: VarInt) -> io::Result<()>;
     /// Wait for the stream to be stopped, returning the error code if it was.
     fn stopped(&mut self) -> impl Future<Output = io::Result<Option<VarInt>>> + Send;
+    /// Get the stream id.
+    fn id(&self) -> u64;
 }
 
 /// An abstract `iroh::endpoint::RecvStream`.
@@ -65,6 +69,10 @@ impl SendStream for iroh::endpoint::SendStream {
     async fn stopped(&mut self) -> io::Result<Option<VarInt>> {
         Ok(self.stopped().await?)
     }
+
+    fn id(&self) -> u64 {
+        self.id().index()
+    }
 }
 
 impl RecvStream for iroh::endpoint::RecvStream {
@@ -153,6 +161,10 @@ impl<W: SendStream> SendStream for &mut W {
     async fn stopped(&mut self) -> io::Result<Option<VarInt>> {
         self.deref_mut().stopped().await
     }
+
+    fn id(&self) -> u64 {
+        self.deref().id()
+    }
 }
 
 #[derive(Debug)]
@@ -289,6 +301,10 @@ impl<W: SendStreamSpecific> SendStream for AsyncWriteSendStream<W> {
         let res = self.0.stopped().await?;
         Ok(res)
     }
+
+    fn id(&self) -> u64 {
+        0
+    }
 }
 
 #[derive(Debug)]