import { BufferEncoding } from "@toruslabs/openlogin-utils" ; import eos from "end-of-stream" ; import once from "once" ; import pump from "pump" ; import { Duplex } from "readable-stream" ; import type { Readable , Writable } from "stream" ; import Substream from "./substream" ; export const IGNORE_SUBSTREAM = Symbol ( "IGNORE_SUBSTREAM" ) ; export type Stream = Readable | Writable ; interface Chunk { name : string ; data : unknown ; } export class ObjectMultiplex extends Duplex { public _substreams : Record < string , Substream | typeof IGNORE_SUBSTREAM > ; getStream : ( name : string ) => Substream | symbol ; constructor ( opts = { } ) { super ( { ... opts , objectMode : true , } ) ; this . _substreams = { } ; } createStream ( name ) { if ( ! name ) { throw new Error ( "ObjectMultiplex - name must not be empty" ) ; } if ( this . _substreams [ name ] ) { throw new Error ( `ObjectMultiplex - Substream for name "${ name }" already exists` ) ; } const substream = new Substream ( { parent : this , name } ) ; this . _substreams [ name ] = substream ; anyStreamEnd ( this , ( _error ? ) => substream . destroy ( _error || undefined ) ) ; return substream ; } ignoreStream ( name ) { if ( ! name ) { throw new Error ( "ObjectMultiplex - name must not be empty" ) ; } if ( this . _substreams [ name ] ) { throw new Error ( `ObjectMultiplex - Substream for name "${ name }" already exists` ) ; } this . _substreams [ name ] = IGNORE_SUBSTREAM ; } _read ( ) { return undefined ; } _write ( chunk , _encoding , callback ) { const { name , data } = chunk ; if ( ! name ) { window . console . warn ( `ObjectMultiplex - malformed chunk without name "${ chunk }"` ) ; return callback ( ) ; } const substream = this . _substreams [ name ] ; if ( ! substream ) { window . console . warn ( `ObjectMultiplex - orphaned data for stream "${ name }"` ) ; return callback ( ) ; } if ( substream !== IGNORE_SUBSTREAM ) { substream . push ( data ) ; } return callback ( ) ; } } function anyStreamEnd ( stream , _cb ) { const cb = once ( _cb ) ; eos ( stream as unknown as Stream , { readable : false } , cb ) ; eos ( stream as unknown as Stream , { writable : false } , cb ) ; } export function setupMultiplex ( stream ) { const mux = new ObjectMultiplex ( ) ; mux . getStream = function streamHelper ( name ) { if ( this . _substreams [ name ] ) { return this . _substreams [ name ] ; } return this . createStream ( name ) ; } ; pump ( stream as unknown as Stream , mux as unknown as Stream , stream as unknown as Stream , ( err ) => { if ( err ) window . console . error ( err ) ; } ) ; return mux ; }