{"version":3,"sources":["stream.ts"],"sourcesContent":["import { ReadableStream } from \"../runtime/runtime.ts\";\nimport { Pointer } from \"../runtime/pointers.ts\";\nimport type { datex_scope } from \"../utils/global_types.ts\";\nimport type { StreamConsumer } from \"./abstract_types.ts\";\nimport { Logger } from \"../utils/logger.ts\";\n\nconst logger = new Logger(\"Stream\")\nlogger.log_to_streams = false;\n\n// <Stream> is stream sink and readable stream at the same time\nexport class Stream<T = ArrayBuffer> implements StreamConsumer<T> {\n\n\n    controller?: ReadableStreamDefaultController\n\n    readable_stream: ReadableStream<T> \n    #writable_stream?: WritableStream<T>\n\n    get writable_stream() {\n        if (!this.#writable_stream) {\n            this.#writable_stream = new WritableStream<T>({\n                write: (chunk) => {\n                    this.write(chunk);\n                }\n            });\n        }\n        return this.#writable_stream;\n    }\n\n    constructor(readable_stream?:ReadableStream<T>) {\n        this.readable_stream = readable_stream ?? new ReadableStream<T>({\n            start: controller => {this.controller = controller}\n        });\n        // immediately start stream out if readable_stream is given\n        if (readable_stream) this.#startStreamOut()\n    }\n\n    started_ptr_stream = false\n\n    #startStreamOut() {\n        const ptr = Pointer.createOrGet(this);\n        if (ptr instanceof Pointer) {\n            logger.debug(\"Start stream out for \" + ptr.idString());\n            setTimeout(() => {\n                ptr.startStreamOut(); // stream to all subscribers or origin, workaround: timeout to prevent stream init too early (TODO: fix)\n            }, 100)\n        }\n        else {\n            throw new Error(\"Could not bind stream to pointer.\")\n        }\n        this.started_ptr_stream = true;\n    }\n\n    write(chunk: T, scope?: datex_scope) {\n\n        // convert buffers\n        // if (chunk instanceof TypedArray) chunk = (<any>chunk).buffer;\n\n        if (!this.started_ptr_stream && !scope) {  // no scope -> assume called from JS, not DATEX\n            this.#startStreamOut()\n        }\n\n        try {\n            this.controller?.enqueue(chunk);\n        }\n        catch (e) {\n            console.error(\"stream write error\", e);\n        }\n    }\n\n    async pipe(in_stream:Stream<T>|ReadableStream<T>, scope?: datex_scope) {\n        const reader = in_stream.getReader();\n        let next:ReadableStreamReadResult<T>;\n        while (true) {\n            next = await reader.read()\n            if (next.done) break;\n            this.write(next.value, scope);\n        }\n    }\n\n    async pipeTo(out_stream:WritableStream<T>) {\n        const reader = this.getReader();\n        const writer = out_stream.getWriter();\n        let next:ReadableStreamReadResult<T>;\n        while (true) {\n            next = await reader.read()\n            if (next.done) break;\n            writer.write(next.value);\n        }\n    }\n\n    close() {\n        this.controller?.close()\n        this.controller = undefined;\n    }\n\n    getReader() {\n        // split in two readable streams\n        const streams = this.readable_stream.tee()\n        this.readable_stream = streams[1];\n        return streams[0].getReader()\n    }\n\n    values() {\n        return this.readable_stream.values()\n    }\n\n    get [Symbol.asyncIterator]() {\n        return this.readable_stream.values.bind(this.readable_stream)\n    }\n}\n"],"names":["ReadableStream","Pointer","Logger","logger","log_to_streams","Symbol","asyncIterator","Stream","controller","readable_stream","writable_stream","WritableStream","write","chunk","constructor","start","started_ptr_stream","ptr","createOrGet","debug","idString","setTimeout","startStreamOut","Error","scope","enqueue","e","console","error","pipe","in_stream","next","reader","getReader","read","done","value","pipeTo","out_stream","writer","getWriter","close","undefined","streams","tee","values","bind"],"mappings":";AAAA,SAASA,cAAc,QAAQ,wBAAwB;AACvD,SAASC,OAAO,QAAQ,yBAAyB;AAGjD,SAASC,MAAM,QAAQ,qBAAqB;AAE5C,IAAMC,SAAS,IAAID,OAAO;AAC1BC,OAAOC,cAAc,GAAG,CAAA,kBAoGfC,OAAOC,aAAa;AAjG7B,OAAO,MAAMC;IAGTC,WAA4C;IAE5CC,gBAAkC;IAClC,CAAA,eAAgB,CAAoB;IAEpC,IAAIC,kBAAkB;QAQlB,OAPK,IAAI,CAAC,CAAA,eAAgB,IACtB,CAAA,IAAI,CAAC,CAAA,eAAgB,GAAG,IAAIC,eAAkB;YAC1CC,OAAO,CAACC;gBACJ,IAAI,CAACD,KAAK,CAACC;YACf;QACJ,EAAC,GAEE,IAAI,CAAC,CAAA,eAAgB;IAChC;IAEAC,YAAYL,eAAkC,CAAE;QAC5C,IAAI,CAACA,eAAe,GAAGA,mBAAmB,IAAIT,eAAkB;YAC5De,OAAOP,CAAAA;gBAAe,IAAI,CAACA,UAAU,GAAGA;YAAU;QACtD,IAEIC,mBAAiB,IAAI,CAAC,CAAA,cAAe;IAC7C;IAEAO,qBAAqB,CAAA,EAAK;IAE1B,CAAA,cAAe;QACX,IAAMC,MAAMhB,QAAQiB,WAAW,CAAC,IAAI;QACpC,IAAID,eAAehB,SACfE,OAAOgB,KAAK,CAAC,0BAA0BF,IAAIG,QAAQ,KACnDC,WAAW;YACPJ,IAAIK,cAAc;QACtB,GAAG;aAGH,MAAM,AAAIC,MAAM;QAEpB,IAAI,CAACP,kBAAkB,GAAG,CAAA;IAC9B;IAEAJ,MAAMC,KAAQ,EAAEW,KAAmB,EAAE;QAK5B,IAAI,CAACR,kBAAkB,IAAKQ,SAC7B,IAAI,CAAC,CAAA,cAAe;QAGxB,IAAI;YACA,IAAI,CAAChB,UAAU,EAAEiB,QAAQZ;QAC7B,EACA,OAAOa,GAAG;YACNC,QAAQC,KAAK,CAAC,sBAAsBF;QACxC;IACJ;IAEA,MAAMG,KAAKC,SAAqC,EAAEN,KAAmB,EAAE;YAE/DO;QADJ,IAAMC,SAASF,UAAUG,SAAS;QAElC,OAEQF,AADJA,CAAAA,OAAO,MAAMC,OAAOE,IAAI,EAAC,EAChBC,IAAI,EACb,IAAI,CAACvB,KAAK,CAACmB,KAAKK,KAAK,EAAEZ;IAE/B;IAEA,MAAMa,OAAOC,UAA4B,EAAE;YAGnCP;QAFJ,IAAMC,SAAS,IAAI,CAACC,SAAS,IACvBM,SAASD,WAAWE,SAAS;QAEnC,OAEQT,AADJA,CAAAA,OAAO,MAAMC,OAAOE,IAAI,EAAC,EAChBC,IAAI,EACbI,OAAO3B,KAAK,CAACmB,KAAKK,KAAK;IAE/B;IAEAK,QAAQ;QACJ,IAAI,CAACjC,UAAU,EAAEiC,SACjB,IAAI,CAACjC,UAAU,GAAGkC,KAAAA;IACtB;IAEAT,YAAY;QAER,IAAMU,UAAU,IAAI,CAAClC,eAAe,CAACmC,GAAG;QAExC,OADA,IAAI,CAACnC,eAAe,GAAGkC,OAAO,CAAC,EAAE,EAC1BA,OAAO,CAAC,EAAE,CAACV,SAAS;IAC/B;IAEAY,SAAS;QACL,OAAO,IAAI,CAACpC,eAAe,CAACoC,MAAM;IACtC;IAEA,qBAA6B;QACzB,OAAO,IAAI,CAACpC,eAAe,CAACoC,MAAM,CAACC,IAAI,CAAC,IAAI,CAACrC,eAAe;IAChE;AACJ","sourceRoot":"/lib/datex-core-js-legacy-0.3.15/types/","file":"stream.ts"}