Skip to content

Commit

Permalink
feat(zip): add higher-order lettable version of zip
Browse files Browse the repository at this point in the history
  • Loading branch information
@benlesh
benlesh committed Jul 24, 2017
1 parent 866af37 commit 8a9b9b2
Show file tree
Hide file tree
Showing 5 changed files with 350 additions and 317 deletions.
4 changes: 2 additions & 2 deletions src/observable/zip.ts
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
import { zipStatic } from '../operator/zip';
import { zipStatic } from '../operators/zip';

export const zip = zipStatic;
export const zip = zipStatic;
316 changes: 2 additions & 314 deletions src/operator/zip.ts
View file
Original file line number Diff line number Diff line change
@@ -1,13 +1,5 @@
import { Observable, ObservableInput } from '../Observable';
import { ArrayObservable } from '../observable/ArrayObservable';
import { isArray } from '../util/isArray';
import { Operator } from '../Operator';
import { PartialObserver } from '../Observer';
import { Subscriber } from '../Subscriber';
import { OuterSubscriber } from '../OuterSubscriber';
import { InnerSubscriber } from '../InnerSubscriber';
import { subscribeToResult } from '../util/subscribeToResult';
import { iterator as Symbol_iterator } from '../symbol/iterator';
import { zip as higherOrder } from '../operators';

/* tslint:disable:max-line-length */
export function zipProto<T, R>(this: Observable<T>, project: (v1: T) => R): Observable<R>;
Expand All @@ -33,309 +25,5 @@ export function zipProto<T, TOther, R>(this: Observable<T>, array: Array<Observa
* @owner Observable
*/
export function zipProto<T, R>(this: Observable<T>, ...observables: Array<ObservableInput<any> | ((...values: Array<any>) => R)>): Observable<R> {
return this.lift.call(zipStatic<R>(this, ...observables));
}

/* tslint:disable:max-line-length */
export function zipStatic<T, T2>(v1: ObservableInput<T>, v2: ObservableInput<T2>): Observable<[T, T2]>;
export function zipStatic<T, T2, T3>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>): Observable<[T, T2, T3]>;
export function zipStatic<T, T2, T3, T4>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>): Observable<[T, T2, T3, T4]>;
export function zipStatic<T, T2, T3, T4, T5>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>): Observable<[T, T2, T3, T4, T5]>;
export function zipStatic<T, T2, T3, T4, T5, T6>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, v6: ObservableInput<T6>): Observable<[T, T2, T3, T4, T5, T6]>;

export function zipStatic<T, R>(v1: ObservableInput<T>, project: (v1: T) => R): Observable<R>;
export function zipStatic<T, T2, R>(v1: ObservableInput<T>, v2: ObservableInput<T2>, project: (v1: T, v2: T2) => R): Observable<R>;
export function zipStatic<T, T2, T3, R>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, project: (v1: T, v2: T2, v3: T3) => R): Observable<R>;
export function zipStatic<T, T2, T3, T4, R>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, project: (v1: T, v2: T2, v3: T3, v4: T4) => R): Observable<R>;
export function zipStatic<T, T2, T3, T4, T5, R>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, project: (v1: T, v2: T2, v3: T3, v4: T4, v5: T5) => R): Observable<R>;
export function zipStatic<T, T2, T3, T4, T5, T6, R>(v1: ObservableInput<T>, v2: ObservableInput<T2>, v3: ObservableInput<T3>, v4: ObservableInput<T4>, v5: ObservableInput<T5>, v6: ObservableInput<T6>, project: (v1: T, v2: T2, v3: T3, v4: T4, v5: T5, v6: T6) => R): Observable<R>;

export function zipStatic<T>(array: ObservableInput<T>[]): Observable<T[]>;
export function zipStatic<R>(array: ObservableInput<any>[]): Observable<R>;
export function zipStatic<T, R>(array: ObservableInput<T>[], project: (...values: Array<T>) => R): Observable<R>;
export function zipStatic<R>(array: ObservableInput<any>[], project: (...values: Array<any>) => R): Observable<R>;

export function zipStatic<T>(...observables: Array<ObservableInput<T>>): Observable<T[]>;
export function zipStatic<T, R>(...observables: Array<ObservableInput<T> | ((...values: Array<T>) => R)>): Observable<R>;
export function zipStatic<R>(...observables: Array<ObservableInput<any> | ((...values: Array<any>) => R)>): Observable<R>;
/* 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<T, R>(...observables: Array<ObservableInput<any> | ((...values: Array<any>) => R)>): Observable<R> {
const project = <((...ys: Array<any>) => R)> observables[observables.length - 1];
if (typeof project === 'function') {
observables.pop();
}
return new ArrayObservable(observables).lift(new ZipOperator(project));
}

export class ZipOperator<T, R> implements Operator<T, R> {

project: (...values: Array<any>) => R;

constructor(project?: (...values: Array<any>) => R) {
this.project = project;
}

call(subscriber: Subscriber<R>, source: any): any {
return source.subscribe(new ZipSubscriber(subscriber, this.project));
}
}

/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
export class ZipSubscriber<T, R> extends Subscriber<T> {
private values: any;
private project: (...values: Array<any>) => R;
private iterators: LookAheadIterator<any>[] = [];
private active = 0;

constructor(destination: Subscriber<R>,
project?: (...values: Array<any>) => R,
values: any = Object.create(null)) {
super(destination);
this.project = (typeof project === 'function') ? project : null;
this.values = values;
}

protected _next(value: any) {
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));
}
}

protected _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: ZipBufferIterator<any, any> = <any>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: any[] = [];
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();
}
}

protected _tryProject(args: any[]) {
let result: any;
try {
result = this.project.apply(this, args);
} catch (err) {
this.destination.error(err);
return;
}
this.destination.next(result);
}
}

interface LookAheadIterator<T> extends Iterator<T> {
hasValue(): boolean;
hasCompleted(): boolean;
}

class StaticIterator<T> implements LookAheadIterator<T> {
private nextResult: IteratorResult<T>;

constructor(private iterator: Iterator<T>) {
this.nextResult = iterator.next();
}

hasValue() {
return true;
}

next(): IteratorResult<T> {
const result = this.nextResult;
this.nextResult = this.iterator.next();
return result;
}

hasCompleted() {
const nextResult = this.nextResult;
return nextResult && nextResult.done;
}
}

class StaticArrayIterator<T> implements LookAheadIterator<T> {
private index = 0;
private length = 0;

constructor(private array: T[]) {
this.length = array.length;
}

[Symbol_iterator]() {
return this;
}

next(value?: any): IteratorResult<T> {
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<T, R> extends OuterSubscriber<T, R> implements LookAheadIterator<T> {
stillUnsubscribed = true;
buffer: T[] = [];
isComplete = false;

constructor(destination: PartialObserver<T>,
private parent: ZipSubscriber<T, R>,
private observable: Observable<T>) {
super(destination);
}

[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(): IteratorResult<T> {
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: T, innerValue: any,
outerIndex: number, innerIndex: number,
innerSub: InnerSubscriber<T, R>): void {
this.buffer.push(innerValue);
this.parent.checkIterators();
}

subscribe(value: any, index: number) {
return subscribeToResult<any, any>(this, this.observable, this, index);
}
return higherOrder(...observables)(this);
}
2 changes: 1 addition & 1 deletion src/operator/zipAll.ts
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { ZipOperator } from './zip';
import { ZipOperator } from '../operators/zip';
import { Observable } from '../Observable';

/**
Expand Down
1 change: 1 addition & 0 deletions src/operators/index.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,3 +38,4 @@ export { windowCount } from './windowCount';
export { windowTime } from './windowTime';
export { windowToggle } from './windowToggle';
export { windowWhen } from './windowWhen';
export { zip } from './zip';
Loading

0 comments on commit 8a9b9b2

Please sign in to comment.