Skip to content

Commit

Permalink
feat(skipUntil): add higher-order lettable version of skipUntil
Browse files Browse the repository at this point in the history
  • Loading branch information
jasonaden committed Sep 7, 2017
1 parent 6e1ff3c commit 6cc2cd6
Show file tree
Hide file tree
Showing 3 changed files with 80 additions and 60 deletions.
62 changes: 2 additions & 60 deletions src/operator/skipUntil.ts
Original file line number Diff line number Diff line change
@@ -1,10 +1,5 @@
import { Operator } from '../Operator';
import { Subscriber } from '../Subscriber';
import { Observable } from '../Observable';
import { TeardownLogic } from '../Subscription';
import { OuterSubscriber } from '../OuterSubscriber';
import { InnerSubscriber } from '../InnerSubscriber';
import { subscribeToResult } from '../util/subscribeToResult';
import { skipUntil as higherOrder } from '../operators/skipUntil';

/**
* Returns an Observable that skips items emitted by the source Observable until a second Observable emits an item.
Expand All @@ -19,58 +14,5 @@ import { subscribeToResult } from '../util/subscribeToResult';
* @owner Observable
*/
export function skipUntil<T>(this: Observable<T>, notifier: Observable<any>): Observable<T> {
return this.lift(new SkipUntilOperator(notifier));
}

class SkipUntilOperator<T> implements Operator<T, T> {
constructor(private notifier: Observable<any>) {
}

call(subscriber: Subscriber<T>, source: any): TeardownLogic {
return source.subscribe(new SkipUntilSubscriber(subscriber, this.notifier));
}
}

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

private hasValue: boolean = false;
private isInnerStopped: boolean = false;

constructor(destination: Subscriber<any>,
notifier: Observable<any>) {
super(destination);
this.add(subscribeToResult(this, notifier));
}

protected _next(value: T) {
if (this.hasValue) {
super._next(value);
}
}

protected _complete() {
if (this.isInnerStopped) {
super._complete();
} else {
this.unsubscribe();
}
}

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

notifyComplete(): void {
this.isInnerStopped = true;
if (this.isStopped) {
super._complete();
}
}
return higherOrder(notifier)(this);
}
1 change: 1 addition & 0 deletions src/operators/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ export { refCount } from './refCount';
export { scan } from './scan';
export { skip } from './skip';
export { skipLast } from './skipLast';
export { skipUntil } from './skipUntil';
export { subscribeOn } from './subscribeOn';
export { switchAll } from './switchAll';
export { switchMap } from './switchMap';
Expand Down
77 changes: 77 additions & 0 deletions src/operators/skipUntil.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
import { Operator } from '../Operator';
import { Subscriber } from '../Subscriber';
import { Observable } from '../Observable';
import { TeardownLogic } from '../Subscription';
import { OuterSubscriber } from '../OuterSubscriber';
import { InnerSubscriber } from '../InnerSubscriber';
import { subscribeToResult } from '../util/subscribeToResult';
import { MonoTypeOperatorFunction } from '../interfaces';

/**
* Returns an Observable that skips items emitted by the source Observable until a second Observable emits an item.
*
* <img src="./img/skipUntil.png" width="100%">
*
* @param {Observable} notifier - The second Observable that has to emit an item before the source Observable's elements begin to
* be mirrored by the resulting Observable.
* @return {Observable<T>} An Observable that skips items from the source Observable until the second Observable emits
* an item, then emits the remaining items.
* @method skipUntil
* @owner Observable
*/
export function skipUntil<T>(notifier: Observable<any>): MonoTypeOperatorFunction<T> {
return (source: Observable<T>) => source.lift(new SkipUntilOperator(notifier));
}

class SkipUntilOperator<T> implements Operator<T, T> {
constructor(private notifier: Observable<any>) {
}

call(subscriber: Subscriber<T>, source: any): TeardownLogic {
return source.subscribe(new SkipUntilSubscriber(subscriber, this.notifier));
}
}

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

private hasValue: boolean = false;
private isInnerStopped: boolean = false;

constructor(destination: Subscriber<any>,
notifier: Observable<any>) {
super(destination);
this.add(subscribeToResult(this, notifier));
}

protected _next(value: T) {
if (this.hasValue) {
super._next(value);
}
}

protected _complete() {
if (this.isInnerStopped) {
super._complete();
} else {
this.unsubscribe();
}
}

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

notifyComplete(): void {
this.isInnerStopped = true;
if (this.isStopped) {
super._complete();
}
}
}

0 comments on commit 6cc2cd6

Please sign in to comment.