diff options
Diffstat (limited to 'node_modules/rxjs/dist/cjs/internal/operators/mergeInternals.js')
| -rw-r--r-- | node_modules/rxjs/dist/cjs/internal/operators/mergeInternals.js | 65 |
1 files changed, 65 insertions, 0 deletions
diff --git a/node_modules/rxjs/dist/cjs/internal/operators/mergeInternals.js b/node_modules/rxjs/dist/cjs/internal/operators/mergeInternals.js new file mode 100644 index 0000000..17a8a02 --- /dev/null +++ b/node_modules/rxjs/dist/cjs/internal/operators/mergeInternals.js @@ -0,0 +1,65 @@ +"use strict"; +Object.defineProperty(exports, "__esModule", { value: true }); +exports.mergeInternals = void 0; +var innerFrom_1 = require("../observable/innerFrom"); +var executeSchedule_1 = require("../util/executeSchedule"); +var OperatorSubscriber_1 = require("./OperatorSubscriber"); +function mergeInternals(source, subscriber, project, concurrent, onBeforeNext, expand, innerSubScheduler, additionalFinalizer) { + var buffer = []; + var active = 0; + var index = 0; + var isComplete = false; + var checkComplete = function () { + if (isComplete && !buffer.length && !active) { + subscriber.complete(); + } + }; + var outerNext = function (value) { return (active < concurrent ? doInnerSub(value) : buffer.push(value)); }; + var doInnerSub = function (value) { + expand && subscriber.next(value); + active++; + var innerComplete = false; + innerFrom_1.innerFrom(project(value, index++)).subscribe(OperatorSubscriber_1.createOperatorSubscriber(subscriber, function (innerValue) { + onBeforeNext === null || onBeforeNext === void 0 ? void 0 : onBeforeNext(innerValue); + if (expand) { + outerNext(innerValue); + } + else { + subscriber.next(innerValue); + } + }, function () { + innerComplete = true; + }, undefined, function () { + if (innerComplete) { + try { + active--; + var _loop_1 = function () { + var bufferedValue = buffer.shift(); + if (innerSubScheduler) { + executeSchedule_1.executeSchedule(subscriber, innerSubScheduler, function () { return doInnerSub(bufferedValue); }); + } + else { + doInnerSub(bufferedValue); + } + }; + while (buffer.length && active < concurrent) { + _loop_1(); + } + checkComplete(); + } + catch (err) { + subscriber.error(err); + } + } + })); + }; + source.subscribe(OperatorSubscriber_1.createOperatorSubscriber(subscriber, outerNext, function () { + isComplete = true; + checkComplete(); + })); + return function () { + additionalFinalizer === null || additionalFinalizer === void 0 ? void 0 : additionalFinalizer(); + }; +} +exports.mergeInternals = mergeInternals; +//# sourceMappingURL=mergeInternals.js.map
\ No newline at end of file |
