feat: Allow overriding default stream constructor options

src/ObjectMultiplex.ts

@@ -1,4 +1,4 @@
-import { Duplex, finished } from 'readable-stream';
+import { Duplex, finished, type DuplexOptions } from 'readable-stream';
 import once from 'once';
 import { Substream } from './Substream';
 
@@ -12,15 +12,15 @@ interface Chunk {
 export class ObjectMultiplex extends Duplex {
   private _substreams: Record<string, Substream | typeof IGNORE_SUBSTREAM>;
 
-  constructor(opts: Record<string, unknown> = {}) {
+  constructor(opts: DuplexOptions = {}) {
     super({
-      ...opts,
       objectMode: true,
+      ...opts,
     });
     this._substreams = {};
   }
 
-  createStream(name: string): Substream {
+  createStream(name: string, opts: DuplexOptions = {}): Substream {
     // guard stream against destroyed already
     if (this.destroyed) {
       throw new Error(
@@ -47,7 +47,11 @@ export class ObjectMultiplex extends Duplex {
     }
 
     // create substream
-    const substream = new Substream({ parent: this, name });
+    const substream = new Substream({
+      name,
+      parent: this,
+      ...opts,
+    });
     this._substreams[name] = substream;
 
     // listen for parent stream to end

src/Substream.ts

@@ -1,7 +1,7 @@
-import { Duplex } from 'readable-stream';
+import { Duplex, type DuplexOptions } from 'readable-stream';
 import { ObjectMultiplex } from './ObjectMultiplex';
 
-export interface SubstreamOptions {
+export interface SubstreamOptions extends DuplexOptions {
   parent: ObjectMultiplex;
   name: string;
 }
@@ -11,8 +11,11 @@ export class Substream extends Duplex {
 
   private readonly _name: string;
 
-  constructor({ parent, name }: SubstreamOptions) {
-    super({ objectMode: true });
+  constructor({ parent, name, ...streamOptions }: SubstreamOptions) {
+    super({
+      objectMode: true,
+      ...streamOptions,
+    });
     this._parent = parent;
     this._name = name;
   }