123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255 |
- import { ArrayObservable } from '../observable/ArrayObservable';
- import { isArray } from '../util/isArray';
- import { Subscriber } from '../Subscriber';
- import { OuterSubscriber } from '../OuterSubscriber';
- import { subscribeToResult } from '../util/subscribeToResult';
- import { iterator as Symbol_iterator } from '../symbol/iterator';
- /* tslint:enable:max-line-length */
- /**
- * @param observables
- * @return {Observable<R>}
- * @method zip
- * @owner Observable
- */
- export function zip(...observables) {
- return function zipOperatorFunction(source) {
- return source.lift.call(zipStatic(source, ...observables));
- };
- }
- /* tslint:enable:max-line-length */
- /**
- * Combines multiple Observables to create an Observable whose values are calculated from the values, in order, of each
- * of its input Observables.
- *
- * If the latest parameter is a function, this function is used to compute the created value from the input values.
- * Otherwise, an array of the input values is returned.
- *
- * @example <caption>Combine age and name from different sources</caption>
- *
- * let age$ = Observable.of<number>(27, 25, 29);
- * let name$ = Observable.of<string>('Foo', 'Bar', 'Beer');
- * let isDev$ = Observable.of<boolean>(true, true, false);
- *
- * Observable
- * .zip(age$,
- * name$,
- * isDev$,
- * (age: number, name: string, isDev: boolean) => ({ age, name, isDev }))
- * .subscribe(x => console.log(x));
- *
- * // outputs
- * // { age: 27, name: 'Foo', isDev: true }
- * // { age: 25, name: 'Bar', isDev: true }
- * // { age: 29, name: 'Beer', isDev: false }
- *
- * @param observables
- * @return {Observable<R>}
- * @static true
- * @name zip
- * @owner Observable
- */
- export function zipStatic(...observables) {
- const project = observables[observables.length - 1];
- if (typeof project === 'function') {
- observables.pop();
- }
- return new ArrayObservable(observables).lift(new ZipOperator(project));
- }
- export class ZipOperator {
- constructor(project) {
- this.project = project;
- }
- call(subscriber, source) {
- return source.subscribe(new ZipSubscriber(subscriber, this.project));
- }
- }
- /**
- * We need this JSDoc comment for affecting ESDoc.
- * @ignore
- * @extends {Ignored}
- */
- export class ZipSubscriber extends Subscriber {
- constructor(destination, project, values = Object.create(null)) {
- super(destination);
- this.iterators = [];
- this.active = 0;
- this.project = (typeof project === 'function') ? project : null;
- this.values = values;
- }
- _next(value) {
- const iterators = this.iterators;
- if (isArray(value)) {
- iterators.push(new StaticArrayIterator(value));
- }
- else if (typeof value[Symbol_iterator] === 'function') {
- iterators.push(new StaticIterator(value[Symbol_iterator]()));
- }
- else {
- iterators.push(new ZipBufferIterator(this.destination, this, value));
- }
- }
- _complete() {
- const iterators = this.iterators;
- const len = iterators.length;
- if (len === 0) {
- this.destination.complete();
- return;
- }
- this.active = len;
- for (let i = 0; i < len; i++) {
- let iterator = iterators[i];
- if (iterator.stillUnsubscribed) {
- this.add(iterator.subscribe(iterator, i));
- }
- else {
- this.active--; // not an observable
- }
- }
- }
- notifyInactive() {
- this.active--;
- if (this.active === 0) {
- this.destination.complete();
- }
- }
- checkIterators() {
- const iterators = this.iterators;
- const len = iterators.length;
- const destination = this.destination;
- // abort if not all of them have values
- for (let i = 0; i < len; i++) {
- let iterator = iterators[i];
- if (typeof iterator.hasValue === 'function' && !iterator.hasValue()) {
- return;
- }
- }
- let shouldComplete = false;
- const args = [];
- for (let i = 0; i < len; i++) {
- let iterator = iterators[i];
- let result = iterator.next();
- // check to see if it's completed now that you've gotten
- // the next value.
- if (iterator.hasCompleted()) {
- shouldComplete = true;
- }
- if (result.done) {
- destination.complete();
- return;
- }
- args.push(result.value);
- }
- if (this.project) {
- this._tryProject(args);
- }
- else {
- destination.next(args);
- }
- if (shouldComplete) {
- destination.complete();
- }
- }
- _tryProject(args) {
- let result;
- try {
- result = this.project.apply(this, args);
- }
- catch (err) {
- this.destination.error(err);
- return;
- }
- this.destination.next(result);
- }
- }
- class StaticIterator {
- constructor(iterator) {
- this.iterator = iterator;
- this.nextResult = iterator.next();
- }
- hasValue() {
- return true;
- }
- next() {
- const result = this.nextResult;
- this.nextResult = this.iterator.next();
- return result;
- }
- hasCompleted() {
- const nextResult = this.nextResult;
- return nextResult && nextResult.done;
- }
- }
- class StaticArrayIterator {
- constructor(array) {
- this.array = array;
- this.index = 0;
- this.length = 0;
- this.length = array.length;
- }
- [Symbol_iterator]() {
- return this;
- }
- next(value) {
- const i = this.index++;
- const array = this.array;
- return i < this.length ? { value: array[i], done: false } : { value: null, done: true };
- }
- hasValue() {
- return this.array.length > this.index;
- }
- hasCompleted() {
- return this.array.length === this.index;
- }
- }
- /**
- * We need this JSDoc comment for affecting ESDoc.
- * @ignore
- * @extends {Ignored}
- */
- class ZipBufferIterator extends OuterSubscriber {
- constructor(destination, parent, observable) {
- super(destination);
- this.parent = parent;
- this.observable = observable;
- this.stillUnsubscribed = true;
- this.buffer = [];
- this.isComplete = false;
- }
- [Symbol_iterator]() {
- return this;
- }
- // NOTE: there is actually a name collision here with Subscriber.next and Iterator.next
- // this is legit because `next()` will never be called by a subscription in this case.
- next() {
- const buffer = this.buffer;
- if (buffer.length === 0 && this.isComplete) {
- return { value: null, done: true };
- }
- else {
- return { value: buffer.shift(), done: false };
- }
- }
- hasValue() {
- return this.buffer.length > 0;
- }
- hasCompleted() {
- return this.buffer.length === 0 && this.isComplete;
- }
- notifyComplete() {
- if (this.buffer.length > 0) {
- this.isComplete = true;
- this.parent.notifyInactive();
- }
- else {
- this.destination.complete();
- }
- }
- notifyNext(outerValue, innerValue, outerIndex, innerIndex, innerSub) {
- this.buffer.push(innerValue);
- this.parent.checkIterators();
- }
- subscribe(value, index) {
- return subscribeToResult(this, this.observable, this, index);
- }
- }
- //# sourceMappingURL=zip.js.map
|