ldiamond/strom
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Rename construct

Browse Source
This commit is contained in:
Jerry Kurian 2020-05-08 16:21:20 -04:00
parent 58a95a91d0
commit a42560edfc
1 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
src/functions/demux.ts
View File

@ -9,14 +9,14 @@ export interface DemuxOptions extends DuplexOptions {
}
export function demux(
construct: (
pipelineConstructor: (
destKey?: string,
chunk?: any,
) => DemuxStreams | Array<DemuxStreams>,
demuxBy: string | ((chunk: any) => string),
options?: DemuxOptions,
): Duplex {
return new Demux(construct, demuxBy, options);
return new Demux(pipelineConstructor, demuxBy, options);
}
class Demux extends Duplex {
@ -24,11 +24,14 @@ class Demux extends Duplex {
[key: string]: Array<DemuxStreams>;
};
private demuxer: (chunk: any) => string;
private construct: (destKey?: string, chunk?: any) => Array<DemuxStreams>;
private pipelineConstructor: (
destKey?: string,
chunk?: any,
) => Array<DemuxStreams>;
private remultiplex: boolean;
private transform: Transform;
constructor(
construct: (
pipelineConstructor: (
destKey?: string,
chunk?: any,
) => DemuxStreams | Array<DemuxStreams>,
@ -38,8 +41,8 @@ class Demux extends Duplex {
super(options);
this.demuxer =
typeof demuxBy === "string" ? chunk => chunk[demuxBy] : demuxBy;
this.construct = (destKey: string, chunk?: any) => {
const pipeline = construct(destKey, chunk);
this.pipelineConstructor = (destKey: string, chunk?: any) => {
const pipeline = pipelineConstructor(destKey, chunk);
return Array.isArray(pipeline) ? pipeline : [pipeline];
};
this.remultiplex =
@ -62,7 +65,7 @@ class Demux extends Duplex {
public async _write(chunk: any, encoding: any, cb: any) {
const destKey = this.demuxer(chunk);
if (this.streamsByKey[destKey] === undefined) {
const newPipelines = this.construct(destKey, chunk);
const newPipelines = this.pipelineConstructor(destKey, chunk);
this.streamsByKey[destKey] = newPipelines;
newPipelines.forEach(newPipeline => {