feat: add support for chained streams in decoders

Author: melchor629

lib/decoder/file-decoder.d.ts

@@ -13,6 +13,7 @@ export interface FileDecoderOptions extends DecoderOptions {
  * @emits metadata When a metadata block is received, the event will be fired
  * @emits format When the decoder knows the exact format of the flac
  * @emits flac-error When there is an error on the decoding
+ * @emits end-link When chained decoding is enabled and a link finishes
  */
 export default class FileDecoder extends Readable implements BaseDecoder {
   constructor(options: FileDecoderOptions);

lib/decoder/file-decoder.js

@@ -31,6 +31,10 @@ class FileDecoder extends Readable {
         this._builder.setMetadataRespond(type)
       }
     }
+
+    if (options.isChainedStream) {
+      this._builder.setDecodeChainedStream(true)
+    }
   }
 
   get processedSamples() {
@@ -148,6 +152,10 @@ class FileDecoder extends Readable {
           await this._dec.finishAsync()
           this.push(null)
           this._dec = null
+        } else if (decState === flac.Decoder.State.END_OF_LINK) {
+          this._debug('End of link reached')
+          await this._dec.finishLinkAsync()
+          this.emit('end-link')
         } else if (!couldProcess && !this.destroyed) {
           this._throwDecoderError()
           return

lib/decoder/interfaces.d.ts

@@ -4,6 +4,8 @@ import { format } from '../api.js'
 export interface DecoderOptions {
   /** If set to true, the input must be an Ogg/FLAC stream. */
   isOggStream?: boolean
+  /** Enables chaining decoding. Emits `end-link` when a link is finished. */
+  isChainedStream?: boolean
   /**
   * If set to true, it will emit the `metadata` for each metadata block.
   * If set to an array, it will only emit `metadata` for the types

lib/decoder/stream-decoder.d.ts

@@ -7,6 +7,7 @@ import { BaseDecoder, DecoderOptions, DecoderPosition } from './interfaces.js'
  * @emits metadata When a metadata block is received, the event will be fired
  * @emits format When the decoder knows the exact format of the flac
  * @emits flac-error When there is an error on the decoding
+ * @emits end-link When chained decoding is enabled and a link finishes
  */
 export default class StreamDecoder extends Transform implements BaseDecoder {
   constructor(props: DecoderOptions)

lib/decoder/stream-decoder.js

@@ -31,6 +31,10 @@ class StreamDecoder extends stream.Transform {
         this._builder.setMetadataRespond(type)
       }
     }
+
+    if (options.isChainedStream) {
+      this._builder.setDecodeChainedStream(true)
+    }
   }
 
   get processedSamples() {
@@ -166,6 +170,14 @@ class StreamDecoder extends stream.Transform {
         }
         this._readCallback = null
 
+        // if chained is enabled and the decoder is in end of link, then finish link
+        // and prepare the decoder for the next link
+        if (this._dec.getState() === flac.Decoder.State.END_OF_LINK) {
+          this._debug('End of link reached')
+          await this._dec.finishLinkAsync()
+          this.emit('end-link')
+        }
+
         // if the call has been blocked, then it does not need to do anything else
         if (this._hasBeenBlocked) {
           this._debug('processSingle was blocked, stopping loop for this processing')

test/encode-decode/js-streams.test.js

@@ -133,6 +133,29 @@ describe('encode & decode: js streams', () => {
       comparePCM(okData, raw, 24)
     })
 
+    it('decode using stream (chained ogg)', async () => {
+      const input = fs.createReadStream(pathForFile('chained.oga'))
+      const dec = new StreamDecoder({ outputAs32: false, isOggStream: true, isChainedStream: true })
+      const chunks = []
+      let endLinks = 0
+      let metadata = 0
+
+      input.pipe(dec)
+      dec.on('data', (chunk) => chunks.push(chunk))
+      dec.on('end-link', () => { endLinks += 1 })
+      dec.on('metadata', () => { metadata += 1 })
+      await events.once(dec, 'end')
+
+      const raw = Buffer.concat(chunks)
+      expect(raw).toHaveLength(totalSamples * 3 * 2 * 3)
+      expect(dec.processedSamples).toStrictEqual(totalSamples * 3)
+      expect(endLinks).toBe(2)
+      expect(metadata).toBe(3)
+      comparePCM(okData, raw.subarray(0, okData.byteLength), 24)
+      comparePCM(okData, raw.subarray(okData.byteLength, okData.byteLength * 2), 24)
+      comparePCM(okData, raw.subarray(okData.byteLength * 2), 24)
+    })
+
     it('encode using stream and file-bit input', async () => {
       const output = fs.createWriteStream(tmpFile.path)
       const enc = new StreamEncoder({
@@ -442,6 +465,32 @@ describe('encode & decode: js streams', () => {
       comparePCM(okData, raw, 24)
     })
 
+    it('decode using file (chained ogg)', async () => {
+      const dec = new FileDecoder({
+        outputAs32: false,
+        file: pathForFile('chained.oga'),
+        isOggStream: true,
+        isChainedStream: true,
+      })
+      const chunks = []
+      let endLinks = 0
+      let metadata = 0
+
+      dec.on('data', (chunk) => chunks.push(chunk))
+      dec.on('end-link', () => { endLinks += 1 })
+      dec.on('metadata', () => { metadata += 1 })
+      await events.once(dec, 'end')
+
+      const raw = Buffer.concat(chunks)
+      expect(raw).toHaveLength(totalSamples * 3 * 2 * 3)
+      expect(dec.processedSamples).toStrictEqual(totalSamples * 3)
+      expect(endLinks).toBe(2)
+      expect(metadata).toBe(3)
+      comparePCM(okData, raw.subarray(0, okData.byteLength), 24)
+      comparePCM(okData, raw.subarray(okData.byteLength, okData.byteLength * 2), 24)
+      comparePCM(okData, raw.subarray(okData.byteLength * 2), 24)
+    })
+
     it('encode using file and 24-bit input', async () => {
       const file = tmpFile.path
       const enc = new FileEncoder({