fix: use .close() to close streams (#3191)

Instead of the weird double-sink of `[]`, use the `.close` method
to explicitly close streams and muxers.

Author: achingbrain

packages/interface-compliance-tests/src/stream-muxer/base-test.ts

@@ -1,7 +1,6 @@
 import { expect } from 'aegir/chai'
 import all from 'it-all'
 import { byteStream } from 'it-byte-stream'
-import drain from 'it-drain'
 import map from 'it-map'
 import { duplexPair } from 'it-pair/duplex'
 import { pipe } from 'it-pipe'
@@ -12,22 +11,19 @@ import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
 import { isValidTick } from '../is-valid-tick.js'
 import type { TestSetup } from '../index.js'
 import type { Stream, StreamMuxerFactory } from '@libp2p/interface'
-import type { Source, Duplex } from 'it-stream-types'
+import type { Source } from 'it-stream-types'
 import type { DeferredPromise } from 'p-defer'
 
-async function drainAndClose (stream: Duplex<any>): Promise<void> {
-  await pipe([], stream, drain)
-}
-
 export default (common: TestSetup<StreamMuxerFactory>): void => {
   describe('base', () => {
-    it('Open a stream from the dialer', async () => {
+    it('should open a stream from the dialer', async () => {
       const p = duplexPair<Uint8Array | Uint8ArrayList>()
-      const dialerFactory = await common.setup()
-      const dialer = dialerFactory.createStreamMuxer({ direction: 'outbound' })
       const onStreamPromise: DeferredPromise<Stream> = defer()
       const onStreamEndPromise: DeferredPromise<Stream> = defer()
 
+      const dialerFactory = await common.setup()
+      const dialer = dialerFactory.createStreamMuxer({ direction: 'outbound' })
+
       const listenerFactory = await common.setup()
       const listener = listenerFactory.createStreamMuxer({
         direction: 'inbound',
@@ -42,18 +38,20 @@ export default (common: TestSetup<StreamMuxerFactory>): void => {
       void pipe(p[0], dialer, p[0])
       void pipe(p[1], listener, p[1])
 
-      const conn = await dialer.newStream()
-      expect(dialer.streams).to.include(conn)
-      expect(isValidTick(conn.timeline.open)).to.equal(true)
+      const dialerStream = await dialer.newStream()
+      expect(dialer.streams).to.include(dialerStream)
+      expect(isValidTick(dialerStream.timeline.open)).to.equal(true)
 
-      void drainAndClose(conn)
+      const dialerBytes = byteStream(dialerStream)
+      void dialerBytes.write(uint8ArrayFromString('hello'))
 
-      const stream = await onStreamPromise.promise
-      expect(isValidTick(stream.timeline.open)).to.equal(true)
+      const listenerStream = await onStreamPromise.promise
+      expect(isValidTick(listenerStream.timeline.open)).to.equal(true)
       // Make sure the stream is being tracked
-      expect(listener.streams).to.include(stream)
+      expect(listener.streams).to.include(listenerStream)
 
-      void drainAndClose(stream)
+      await dialerStream.close()
+      await listenerStream.close()
 
       // Make sure stream is closed properly
       const endedStream = await onStreamEndPromise.promise
@@ -66,22 +64,26 @@ export default (common: TestSetup<StreamMuxerFactory>): void => {
       // Make sure the stream is removed from tracking
       expect(isValidTick(endedStream.timeline.close)).to.equal(true)
 
-      await drainAndClose(dialer)
-      await drainAndClose(listener)
+      await dialer.close()
+      await listener.close()
 
       // ensure we have no streams left
       expect(dialer.streams).to.have.length(0)
       expect(listener.streams).to.have.length(0)
     })
 
-    it('Open a stream from the listener', async () => {
+    it('should open a stream from the listener', async () => {
       const p = duplexPair<Uint8Array | Uint8ArrayList>()
       const onStreamPromise: DeferredPromise<Stream> = defer()
+      const onStreamEndPromise: DeferredPromise<Stream> = defer()
       const dialerFactory = await common.setup()
       const dialer = dialerFactory.createStreamMuxer({
         direction: 'outbound',
         onIncomingStream: (stream: Stream) => {
           onStreamPromise.resolve(stream)
+        },
+        onStreamEnd: (stream) => {
+          onStreamEndPromise.resolve(stream)
         }
       })
 
@@ -91,21 +93,35 @@ export default (common: TestSetup<StreamMuxerFactory>): void => {
       void pipe(p[0], dialer, p[0])
       void pipe(p[1], listener, p[1])
 
-      const conn = await listener.newStream()
+      const listenerStream = await listener.newStream()
+      const listenerBytes = byteStream(listenerStream)
+      void listenerBytes.write(uint8ArrayFromString('hello'))
 
-      void drainAndClose(conn)
+      const dialerStream = await onStreamPromise.promise
 
-      const stream = await onStreamPromise.promise
-      expect(isValidTick(stream.timeline.open)).to.equal(true)
-      expect(listener.streams).to.include(conn)
-      expect(isValidTick(conn.timeline.open)).to.equal(true)
-      void drainAndClose(stream)
+      expect(isValidTick(dialerStream.timeline.open)).to.equal(true)
+      expect(listener.streams).to.include(listenerStream)
+      expect(isValidTick(listenerStream.timeline.open)).to.equal(true)
+
+      await dialerStream.close()
+      await listenerStream.close()
+
+      // Make sure stream is closed properly
+      const endedStream = await onStreamEndPromise.promise
+      expect(dialer.streams).to.not.include(endedStream)
+
+      if (endedStream.timeline.close == null) {
+        throw new Error('timeline had no close time')
+      }
+
+      // Make sure the stream is removed from tracking
+      expect(isValidTick(endedStream.timeline.close)).to.equal(true)
 
-      await drainAndClose(dialer)
-      await drainAndClose(listener)
+      await dialer.close()
+      await listener.close()
     })
 
-    it('Open a stream on both sides', async () => {
+    it('should open a stream on both sides', async () => {
       const p = duplexPair<Uint8Array | Uint8ArrayList>()
       const onDialerStreamPromise: DeferredPromise<Stream> = defer()
       const onListenerStreamPromise: DeferredPromise<Stream> = defer()
@@ -132,19 +148,19 @@ export default (common: TestSetup<StreamMuxerFactory>): void => {
       const listenerInitiatorStream = await listener.newStream()
 
       await Promise.all([
-        drainAndClose(dialerInitiatorStream),
-        drainAndClose(listenerInitiatorStream),
-        onDialerStreamPromise.promise.then(async stream => { await drainAndClose(stream) }),
-        onListenerStreamPromise.promise.then(async stream => { await drainAndClose(stream) })
+        dialerInitiatorStream.close(),
+        listenerInitiatorStream.close(),
+        onDialerStreamPromise.promise.then(async stream => { await stream.close() }),
+        onListenerStreamPromise.promise.then(async stream => { await stream.close() })
       ])
 
       await Promise.all([
-        drainAndClose(dialer),
-        drainAndClose(listener)
+        dialer.close(),
+        listener.close()
       ])
     })
 
-    it('Open a stream on one side, write, open a stream on the other side', async () => {
+    it('should open a stream on one side, write, open a stream on the other side', async () => {
       const toString = (source: Source<Uint8ArrayList>): AsyncGenerator<string> => map(source, (u) => uint8ArrayToString(u.subarray()))
       const p = duplexPair<Uint8Array | Uint8ArrayList>()
       const onDialerStreamPromise: DeferredPromise<Stream> = defer()