fix: byte stream should return null when remote closes

packages/interface/src/message-stream.ts

@@ -134,6 +134,16 @@ export interface MessageStream<Timeline extends MessageStreamTimeline = MessageS
    */
   writableNeedsDrain: boolean
 
+  /**
+   * Returns the number of bytes that are queued to be read
+   */
+  readBufferLength: number
+
+  /**
+   * Returns the number of bytes that are queued to be written
+   */
+  writeBufferLength: number
+
   /**
    * Write data to the stream. If the method returns false it means the
    * internal buffer is now full and the caller should wait for the 'drain'

packages/transport-webrtc/src/stream.ts

@@ -207,20 +207,30 @@ export class WebRTCStream extends AbstractStream {
     options?.signal?.throwIfAborted()
     this.receivedFinAck = Promise.withResolvers<void>()
 
+    // don't wait for FIN_ACK forever
+    const signal = options?.signal ?? AbortSignal.timeout(this.finAckTimeout)
+
+    // allow cleaning up event promises
+    const eventPromises = [
+      pEvent(this.channel, 'close', {
+        signal
+      }),
+      pEvent(this.channel, 'error', {
+        signal
+      })
+    ]
+
     // wait for either:
     // 1. the FIN_ACK to be received
     // 2. the datachannel to close
     // 3. timeout
     await Promise.any([
-      raceSignal(this.receivedFinAck.promise, options?.signal),
-      pEvent(this.channel, 'close'),
-      new Promise<void>(resolve => {
-        AbortSignal.timeout(this.finAckTimeout)
-          .addEventListener('abort', () => {
-            resolve()
-          })
-      })
+      raceSignal(this.receivedFinAck.promise, signal),
+      ...eventPromises
     ])
+      .finally(() => {
+        eventPromises.forEach(p => p.cancel())
+      })
   }
 
   async sendCloseRead (options?: AbortOptions): Promise<void> {

packages/utils/src/abstract-message-stream.ts

@@ -100,6 +100,14 @@ export abstract class AbstractMessageStream<Timeline extends MessageStreamTimeli
     this.addEventListener('drain', continueSendingOnDrain)
   }
 
+  get readBufferLength (): number {
+    return this.readBuffer.byteLength
+  }
+
+  get writeBufferLength (): number {
+    return this.writeBuffer.byteLength
+  }
+
   async * [Symbol.asyncIterator] (): AsyncGenerator<Uint8Array | Uint8ArrayList> {
     if (this.readStatus !== 'readable' && this.readStatus !== 'paused') {
       return

packages/utils/src/stream-utils.ts

@@ -96,7 +96,7 @@ function isMultiaddrConnection (obj?: any): obj is MultiaddrConnection {
 
 function isEOF (obj?: any): boolean {
   if (isStream(obj)) {
-    return obj.readStatus === 'closing' || obj.readStatus === 'closed'
+    return obj.remoteWriteStatus !== 'writable' && obj.readBufferLength === 0
   }
 
   if (isMultiaddrConnection(obj)) {

packages/utils/test/stream-utils-test.spec.ts

@@ -181,6 +181,21 @@ describe('byte-stream', () => {
     expect(readIncoming).to.deep.equal(writtenOutgoing)
     expect(readOutgoing).to.deep.equal(writtenIncoming)
   })
+
+  it('should return null when the remote closes it\'s writable end', async () => {
+    const [outgoing, incoming] = await streamPair()
+
+    const incomingBytes = byteStream(incoming)
+
+    const [
+      bytes
+    ] = await Promise.all([
+      incomingBytes.read(),
+      outgoing.close()
+    ])
+
+    expect(bytes).to.be.null()
+  })
 })
 
 describe('stream-pair', () => {