Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
Fix looping behaviour to be identical as to before
+16
-12
+12
-8
src/operators.ts
+12
-8
src/operators.ts
···
256
256
257
257
export function mergeMap<In, Out>(map: (value: In) => Source<Out>): Operator<In, Out> {
258
258
return source => sink => {
259
-
const innerTalkbacks: TalkbackFn[] = [];
259
+
let innerTalkbacks: TalkbackFn[] = [];
260
260
let outerTalkback = talkbackPlaceholder;
261
261
let outerPulled = false;
262
262
let ended = false;
···
266
266
if (signal === SignalKind.End) {
267
267
if (innerTalkbacks.length) {
268
268
const index = innerTalkbacks.indexOf(talkback);
269
-
if (index > -1) innerTalkbacks.splice(index, 1);
269
+
if (index > -1) (innerTalkbacks = innerTalkbacks.slice()).splice(index, 1);
270
270
if (!innerTalkbacks.length) {
271
271
if (ended) {
272
272
sink(SignalKind.End);
···
309
309
ended = true;
310
310
outerTalkback(TalkbackKind.Close);
311
311
}
312
-
while (innerTalkbacks.length) innerTalkbacks.shift()!(TalkbackKind.Close);
312
+
for (let i = 0, a = innerTalkbacks, l = innerTalkbacks.length; i < l; i++)
313
+
a[i](TalkbackKind.Close);
314
+
innerTalkbacks.length = 0;
313
315
} else {
314
316
if (!ended && !outerPulled) {
315
317
outerPulled = true;
···
317
319
} else {
318
320
outerPulled = false;
319
321
}
320
-
for (let i = 0; i < innerTalkbacks.length; i++) innerTalkbacks[i](TalkbackKind.Pull);
322
+
for (let i = 0, a = innerTalkbacks, l = innerTalkbacks.length; i < l; i++)
323
+
a[i](TalkbackKind.Pull);
321
324
}
322
325
})
323
326
);
···
475
478
}
476
479
477
480
export function share<T>(source: Source<T>): Source<T> {
478
-
const sinks: Sink<T>[] = [];
481
+
let sinks: Sink<T>[] = [];
479
482
let talkback = talkbackPlaceholder;
480
483
let gotSignal = false;
481
484
return sink => {
···
483
486
if (sinks.length === 1) {
484
487
source(signal => {
485
488
if (signal === SignalKind.End) {
486
-
while (sinks.length) sinks.pop()!(SignalKind.End);
489
+
for (let i = 0, a = sinks, l = sinks.length; i < l; i++) a[i](SignalKind.End);
490
+
sinks.length = 0;
487
491
} else if (signal.tag === SignalKind.Start) {
488
492
talkback = signal[0];
489
493
} else {
490
494
gotSignal = false;
491
-
for (let i = 0; i < sinks.length; i++) sinks[i](signal);
495
+
for (let i = 0, a = sinks, l = sinks.length; i < l; i++) a[i](signal);
492
496
}
493
497
});
494
498
}
···
496
500
start(signal => {
497
501
if (signal === TalkbackKind.Close) {
498
502
const index = sinks.indexOf(sink);
499
-
if (index > -1) sinks.splice(index, 1);
503
+
if (index > -1) (sinks = sinks.slice()).splice(index, 1);
500
504
if (!sinks.length) talkback(TalkbackKind.Close);
501
505
} else if (!gotSignal) {
502
506
gotSignal = true;
+4
-4
src/sources.ts
+4
-4
src/sources.ts
···
73
73
}
74
74
75
75
export function makeSubject<T>(): Subject<T> {
76
-
const sinks: Sink<T>[] = [];
76
+
let sinks: Sink<T>[] = [];
77
77
let ended = false;
78
78
return {
79
79
source(sink: Sink<T>) {
···
82
82
start(signal => {
83
83
if (signal === TalkbackKind.Close) {
84
84
const index = sinks.indexOf(sink);
85
-
if (index > -1) sinks.splice(index, 1);
85
+
if (index > -1) (sinks = sinks.slice()).splice(index, 1);
86
86
}
87
87
})
88
88
);
···
90
90
next(value: T) {
91
91
if (!ended) {
92
92
const signal = push(value);
93
-
for (let i = 0; i < sinks.length; i++) sinks[i](signal);
93
+
for (let i = 0, a = sinks, l = sinks.length; i < l; i++) a[i](signal);
94
94
}
95
95
},
96
96
complete() {
97
97
if (!ended) {
98
98
ended = true;
99
-
for (let i = 0; i < sinks.length; i++) sinks[i](SignalKind.End);
99
+
for (let i = 0, a = sinks, l = sinks.length; i < l; i++) a[i](SignalKind.End);
100
100
}
101
101
},
102
102
};