feat: add tests for closeRead and closeWrite on streams

src/stream-muxer/tests/close-test.js

@@ -2,6 +2,10 @@
 /* eslint max-nested-callbacks: ["error", 8] */
 'use strict'
 
+const chai = require('chai')
+const expect = chai.expect
+chai.use(require('dirty-chai'))
+
 const pair = require('it-pair/duplex')
 const pipe = require('it-pipe')
 const { consume } = require('streaming-iterables')
@@ -109,5 +113,69 @@ module.exports = (common) => {
       // These should now all resolve without error
       await Promise.all(streamResults)
     })
+
+    it('can close a stream for writing', (done) => {
+      const p = pair()
+      const dialer = new Muxer()
+      const data = [randomBuffer(), randomBuffer()]
+
+      const listener = new Muxer(async stream => {
+        // Immediate close for write
+        await stream.closeWrite()
+        
+        const results = await pipe(stream, async (source) => {
+          const data = []
+          for await (const chunk of source) {
+            data.push(chunk.slice())
+          }
+          return data
+        })        
+        expect(results).to.eql(data)
+
+        try {
+          await stream.sink([randomBuffer()])
+        } catch (err) {
+          expect(err).to.exist()
+          return done()
+        }
+        expect.fail('should not support writing to closed writer')
+      })
+
+      pipe(p[0], dialer, p[0])
+      pipe(p[1], listener, p[1])
+
+      const stream = dialer.newStream()
+      stream.sink(data)
+    })
+
+    it('can close a stream for reading', (done) => {
+      const p = pair()
+      const dialer = new Muxer()
+      const data = [randomBuffer(), randomBuffer()]
+
+      const listener = new Muxer(async stream => {
+        const results = await pipe(stream, async (source) => {
+          const data = []
+          for await (const chunk of source) {
+            data.push(chunk.slice())
+          }
+          return data
+        })        
+        expect(results).to.eql(data)
+        done()
+      })
+
+      pipe(p[0], dialer, p[0])
+      pipe(p[1], listener, p[1])
+
+      const stream = dialer.newStream()  
+      stream.closeRead() 
+
+      // Source should be done
+      ;(async () => {
+        expect(await stream.source.next()).to.eql({ done: true })  
+        stream.sink(data)
+      })()        
+    })
   })
 }