Skip to content

Commit

Permalink
Merge pull request #2806 from jasonaden/lettable_operators
Browse files Browse the repository at this point in the history 
Lettable operators
  • Loading branch information
@benlesh
benlesh committed Aug 29, 2017
2 parents 3136403 + e911aef commit 5b8a7c1
Show file tree
Hide file tree
Showing 20 changed files with 879 additions and 480 deletions.
2 changes: 1 addition & 1 deletion src/observable/onErrorResumeNext.ts
View file
@@ -1,3 +1,3 @@
import { onErrorResumeNextStatic } from '../operator/onErrorResumeNext';
import { onErrorResumeNextStatic } from '../operators/onErrorResumeNext';

export const onErrorResumeNext = onErrorResumeNextStatic;
95 changes: 2 additions & 93 deletions src/operator/last.ts
View file
@@ -1,7 +1,5 @@
import { Observable } from '../Observable';
import { Operator } from '../Operator';
import { Subscriber } from '../Subscriber';
import { EmptyError } from '../util/EmptyError';
import { last as higherOrder } from '../operators';

/* tslint:disable:max-line-length */
export function last<T, S extends T>(this: Observable<T>,
Expand Down Expand Up @@ -45,94 +43,5 @@ export function last<T>(this: Observable<T>,
export function last<T, R>(this: Observable<T>, predicate?: (value: T, index: number, source: Observable<T>) => boolean,
resultSelector?: ((value: T, index: number) => R) | void,
defaultValue?: R): Observable<T | R> {
return this.lift(new LastOperator(predicate, resultSelector, defaultValue, this));
}

class LastOperator<T, R> implements Operator<T, R> {
constructor(private predicate?: (value: T, index: number, source: Observable<T>) => boolean,
private resultSelector?: ((value: T, index: number) => R) | void,
private defaultValue?: any,
private source?: Observable<T>) {
}

call(observer: Subscriber<R>, source: any): any {
return source.subscribe(new LastSubscriber(observer, this.predicate, this.resultSelector, this.defaultValue, this.source));
}
}

/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class LastSubscriber<T, R> extends Subscriber<T> {
private lastValue: T | R;
private hasValue: boolean = false;
private index: number = 0;

constructor(destination: Subscriber<R>,
private predicate?: (value: T, index: number, source: Observable<T>) => boolean,
private resultSelector?: ((value: T, index: number) => R) | void,
private defaultValue?: any,
private source?: Observable<T>) {
super(destination);
if (typeof defaultValue !== 'undefined') {
this.lastValue = defaultValue;
this.hasValue = true;
}
}

protected _next(value: T): void {
const index = this.index++;
if (this.predicate) {
this._tryPredicate(value, index);
} else {
if (this.resultSelector) {
this._tryResultSelector(value, index);
return;
}
this.lastValue = value;
this.hasValue = true;
}
}

private _tryPredicate(value: T, index: number) {
let result: any;
try {
result = this.predicate(value, index, this.source);
} catch (err) {
this.destination.error(err);
return;
}
if (result) {
if (this.resultSelector) {
this._tryResultSelector(value, index);
return;
}
this.lastValue = value;
this.hasValue = true;
}
}

private _tryResultSelector(value: T, index: number) {
let result: any;
try {
result = (<any>this).resultSelector(value, index);
} catch (err) {
this.destination.error(err);
return;
}
this.lastValue = result;
this.hasValue = true;
}

protected _complete(): void {
const destination = this.destination;
if (this.hasValue) {
destination.next(this.lastValue);
destination.complete();
} else {
destination.error(new EmptyError);
}
}
return higherOrder(predicate, resultSelector as any, defaultValue)(this);
}
37 changes: 2 additions & 35 deletions src/operator/mapTo.ts
View file
@@ -1,6 +1,5 @@
import { Operator } from '../Operator';
import { Subscriber } from '../Subscriber';
import { Observable } from '../Observable';
import { mapTo as higherOrder } from '../operators';

/**
* Emits the given constant value on the output Observable every time the source
Expand Down Expand Up @@ -29,37 +28,5 @@ import { Observable } from '../Observable';
* @owner Observable
*/
export function mapTo<T, R>(this: Observable<T>, value: R): Observable<R> {
return this.lift(new MapToOperator(value));
}

class MapToOperator<T, R> implements Operator<T, R> {

value: R;

constructor(value: R) {
this.value = value;
}

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

/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class MapToSubscriber<T, R> extends Subscriber<T> {

value: R;

constructor(destination: Subscriber<R>, value: R) {
super(destination);
this.value = value;
}

protected _next(x: T) {
this.destination.next(this.value);
}
return higherOrder(value)(this);
}
115 changes: 2 additions & 113 deletions src/operator/mergeMapTo.ts
View file
@@ -1,11 +1,5 @@
import { Observable, ObservableInput } from '../Observable';
import { Operator } from '../Operator';
import { PartialObserver } from '../Observer';
import { Subscriber } from '../Subscriber';
import { Subscription } from '../Subscription';
import { OuterSubscriber } from '../OuterSubscriber';
import { InnerSubscriber } from '../InnerSubscriber';
import { subscribeToResult } from '../util/subscribeToResult';
import { mergeMapTo as higherOrder } from '../operators/mergeMapTo';

/* tslint:disable:max-line-length */
export function mergeMapTo<T, R>(this: Observable<T>, observable: ObservableInput<R>, concurrent?: number): Observable<R>;
Expand Down Expand Up @@ -58,110 +52,5 @@ export function mergeMapTo<T, I, R>(this: Observable<T>, observable: ObservableI
export function mergeMapTo<T, I, R>(this: Observable<T>, innerObservable: Observable<I>,
resultSelector?: ((outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R) | number,
concurrent: number = Number.POSITIVE_INFINITY): Observable<R> {
if (typeof resultSelector === 'number') {
concurrent = <number>resultSelector;
resultSelector = null;
}
return this.lift(new MergeMapToOperator(innerObservable, <any>resultSelector, concurrent));
}

// TODO: Figure out correct signature here: an Operator<Observable<T>, R>
// needs to implement call(observer: Subscriber<R>): Subscriber<Observable<T>>
export class MergeMapToOperator<T, I, R> implements Operator<Observable<T>, R> {
constructor(private ish: ObservableInput<I>,
private resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R,
private concurrent: number = Number.POSITIVE_INFINITY) {
}

call(observer: Subscriber<R>, source: any): any {
return source.subscribe(new MergeMapToSubscriber(observer, this.ish, this.resultSelector, this.concurrent));
}
}

/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
export class MergeMapToSubscriber<T, I, R> extends OuterSubscriber<T, I> {
private hasCompleted: boolean = false;
private buffer: T[] = [];
private active: number = 0;
protected index: number = 0;

constructor(destination: Subscriber<R>,
private ish: ObservableInput<I>,
private resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R,
private concurrent: number = Number.POSITIVE_INFINITY) {
super(destination);
}

protected _next(value: T): void {
if (this.active < this.concurrent) {
const resultSelector = this.resultSelector;
const index = this.index++;
const ish = this.ish;
const destination = this.destination;

this.active++;
this._innerSub(ish, destination, resultSelector, value, index);
} else {
this.buffer.push(value);
}
}

private _innerSub(ish: ObservableInput<I>,
destination: PartialObserver<I>,
resultSelector: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R,
value: T,
index: number): void {
this.add(subscribeToResult<T, I>(this, ish, value, index));
}

protected _complete(): void {
this.hasCompleted = true;
if (this.active === 0 && this.buffer.length === 0) {
this.destination.complete();
}
}

notifyNext(outerValue: T, innerValue: I,
outerIndex: number, innerIndex: number,
innerSub: InnerSubscriber<T, I>): void {
const { resultSelector, destination } = this;
if (resultSelector) {
this.trySelectResult(outerValue, innerValue, outerIndex, innerIndex);
} else {
destination.next(innerValue);
}
}

private trySelectResult(outerValue: T, innerValue: I,
outerIndex: number, innerIndex: number): void {
const { resultSelector, destination } = this;
let result: R;
try {
result = resultSelector(outerValue, innerValue, outerIndex, innerIndex);
} catch (err) {
destination.error(err);
return;
}

destination.next(result);
}

notifyError(err: any): void {
this.destination.error(err);
}

notifyComplete(innerSub: Subscription): void {
const buffer = this.buffer;
this.remove(innerSub);
this.active--;
if (buffer.length > 0) {
this._next(buffer.shift());
} else if (this.active === 0 && this.hasCompleted) {
this.destination.complete();
}
}
return higherOrder(innerObservable, resultSelector as any, concurrent)(this);
}
98 changes: 3 additions & 95 deletions src/operator/mergeScan.ts
View file
@@ -1,12 +1,6 @@
import { Operator } from '../Operator';

import { Observable } from '../Observable';
import { Subscriber } from '../Subscriber';
import { Subscription } from '../Subscription';
import { tryCatch } from '../util/tryCatch';
import { errorObject } from '../util/errorObject';
import { subscribeToResult } from '../util/subscribeToResult';
import { OuterSubscriber } from '../OuterSubscriber';
import { InnerSubscriber } from '../InnerSubscriber';
import { mergeScan as higherOrder } from '../operators/mergeScan';

/**
* Applies an accumulator function over the source Observable where the
Expand Down Expand Up @@ -43,91 +37,5 @@ export function mergeScan<T, R>(this: Observable<T>,
accumulator: (acc: R, value: T) => Observable<R>,
seed: R,
concurrent: number = Number.POSITIVE_INFINITY): Observable<R> {
return this.lift(new MergeScanOperator(accumulator, seed, concurrent));
}

export class MergeScanOperator<T, R> implements Operator<T, R> {
constructor(private accumulator: (acc: R, value: T) => Observable<R>,
private seed: R,
private concurrent: number) {
}

call(subscriber: Subscriber<R>, source: any): any {
return source.subscribe(new MergeScanSubscriber(
subscriber, this.accumulator, this.seed, this.concurrent
));
}
}

/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
export class MergeScanSubscriber<T, R> extends OuterSubscriber<T, R> {
private hasValue: boolean = false;
private hasCompleted: boolean = false;
private buffer: Observable<any>[] = [];
private active: number = 0;
protected index: number = 0;

constructor(destination: Subscriber<R>,
private accumulator: (acc: R, value: T) => Observable<R>,
private acc: R,
private concurrent: number) {
super(destination);
}

protected _next(value: any): void {
if (this.active < this.concurrent) {
const index = this.index++;
const ish = tryCatch(this.accumulator)(this.acc, value);
const destination = this.destination;
if (ish === errorObject) {
destination.error(errorObject.e);
} else {
this.active++;
this._innerSub(ish, value, index);
}
} else {
this.buffer.push(value);
}
}

private _innerSub(ish: any, value: T, index: number): void {
this.add(subscribeToResult<T, R>(this, ish, value, index));
}

protected _complete(): void {
this.hasCompleted = true;
if (this.active === 0 && this.buffer.length === 0) {
if (this.hasValue === false) {
this.destination.next(this.acc);
}
this.destination.complete();
}
}

notifyNext(outerValue: T, innerValue: R,
outerIndex: number, innerIndex: number,
innerSub: InnerSubscriber<T, R>): void {
const { destination } = this;
this.acc = innerValue;
this.hasValue = true;
destination.next(innerValue);
}

notifyComplete(innerSub: Subscription): void {
const buffer = this.buffer;
this.remove(innerSub);
this.active--;
if (buffer.length > 0) {
this._next(buffer.shift());
} else if (this.active === 0 && this.hasCompleted) {
if (this.hasValue === false) {
this.destination.next(this.acc);
}
this.destination.complete();
}
}
return higherOrder(accumulator, seed, concurrent)(this);
}

0 comments on commit 5b8a7c1

Please sign in to comment.