fix: emit uint8arrays (#47)

Ensure that only uint8arrays are yielded

package.json

@@ -173,10 +173,9 @@
   "dependencies": {
     "err-code": "^3.0.1",
     "it-stream-types": "^1.0.4",
-    "uint8-varint": "^1.0.0",
+    "uint8-varint": "^1.0.1",
     "uint8arraylist": "^2.0.0",
-    "uint8arrays": "^3.0.0",
-    "varint": "^6.0.0"
+    "uint8arrays": "^3.0.0"
   },
   "devDependencies": {
     "@types/varint": "^6.0.0",
@@ -191,6 +190,7 @@
     "it-pushable": "^3.0.0",
     "it-reader": "^6.0.1",
     "p-defer": "^4.0.0",
-    "random-int": "^3.0.0"
+    "random-int": "^3.0.0",
+    "varint": "^6.0.0"
   }
 }

src/decode.ts

@@ -100,14 +100,14 @@ export function decode (options?: DecoderOptions): Transform<Uint8ArrayList | Ui
           }
 
           const data = buffer.sublist(0, dataLength)
+          buffer.consume(dataLength)
 
           if (options?.onData != null) {
             options.onData(data)
           }
 
           yield data
 
-          buffer.consume(dataLength)
           mode = ReadMode.LENGTH
         }
       }

src/encode.ts

@@ -2,14 +2,15 @@ import { Uint8ArrayList } from 'uint8arraylist'
 import { unsigned } from 'uint8-varint'
 import type { LengthEncoderFunction } from './index.js'
 import type { Source, Transform } from 'it-stream-types'
+import { allocUnsafe } from './alloc.js'
 
 interface EncoderOptions {
   lengthEncoder?: LengthEncoderFunction
 }
 
 const defaultEncoder: LengthEncoderFunction = (length) => {
   const lengthLength = unsigned.encodingLength(length)
-  const lengthBuf = new Uint8Array(lengthLength)
+  const lengthBuf = allocUnsafe(lengthLength)
 
   unsigned.encode(length, lengthBuf)
 
@@ -27,8 +28,21 @@ export function encode (options?: EncoderOptions): Transform<Uint8ArrayList | Ui
   const encoder = async function * (source: Source<Uint8ArrayList | Uint8Array>): Source<Uint8Array> {
     for await (const chunk of source) {
       // length + data
-      yield encodeLength(chunk.byteLength).subarray()
-      yield chunk
+      const length = encodeLength(chunk.byteLength)
+
+      // yield only Uint8Arrays
+      if (length instanceof Uint8Array) {
+        yield length
+      } else {
+        yield * length
+      }
+
+      // yield only Uint8Arrays
+      if (chunk instanceof Uint8Array) {
+        yield chunk
+      } else {
+        yield * chunk
+      }
     }
   }
 

test/decode.spec.ts

@@ -11,6 +11,8 @@ import { times } from './helpers/index.js'
 import * as lp from '../src/index.js'
 import { MAX_LENGTH_LENGTH, MAX_DATA_LENGTH } from '../src/decode.js'
 import { int32BEDecode } from './helpers/int32BE-decode.js'
+import { block } from 'it-block'
+import { unsigned } from 'uint8-varint'
 
 describe('decode', () => {
   it('should decode single message', async () => {
@@ -26,6 +28,25 @@ describe('decode', () => {
     expect(output.slice(-byteLength)).to.deep.equal(bytes)
   })
 
+  it('should decode single message sent in small chunks', async () => {
+    const byteLength = 512
+    const bytes = await randomBytes(byteLength)
+
+    const input = new Uint8ArrayList(
+      unsigned.encode(byteLength),
+      bytes
+    )
+
+    const [output] = await pipe(
+      [input],
+      block(1),
+      lp.decode(),
+      async (source) => await all(source)
+    )
+
+    expect(output.slice(-byteLength)).to.deep.equal(bytes)
+  })
+
   it('should decode empty single message', async () => {
     const input = uint8ArrayConcat([
       Uint8Array.from(varint.encode(0)),

test/encode.spec.ts

@@ -6,6 +6,7 @@ import { unsigned } from 'uint8-varint'
 import { times, someBytes } from './helpers/index.js'
 import * as lp from '../src/index.js'
 import { int32BEEncode } from './helpers/int32BE-encode.js'
+import { Uint8ArrayList } from 'uint8arraylist'
 
 describe('encode', () => {
   it('should encode length as prefix', async () => {
@@ -64,4 +65,19 @@ describe('encode', () => {
       expect(length).to.equal(input[inputIndex].length)
     }
   })
+
+  it('should only yield uint8arrays', async () => {
+    const output = await pipe(
+      [new Uint8Array(10), new Uint8ArrayList(new Uint8Array(20))],
+      lp.encode(),
+      async (source) => await all(source)
+    )
+
+    // length, data, length, data
+    expect(output).to.have.lengthOf(4)
+
+    for (let i = 0; i < output.length; i++) {
+      expect(output[i]).to.be.an.instanceOf(Uint8Array)
+    }
+  })
 })