refactor(last): smaller last

spec/operators/last-spec.ts

@@ -1,19 +1,18 @@
 
-import * as Rx from 'rxjs/Rx';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
+import { last, mergeMap } from 'rxjs/operators';
+import { EmptyError, of, from, Observable } from 'rxjs';
 
 declare function asDiagram(arg: string): Function;
 
-const Observable = Rx.Observable;
-
 /** @test {last} */
 describe('Observable.prototype.last', () => {
   asDiagram('last')('should take the last value of an observable', () => {
     const e1 =   hot('--a----b--c--|');
     const e1subs =   '^            !';
     const expected = '-------------(c|)';
 
-    expectObservable(e1.last()).toBe(expected);
+    expectObservable(e1.pipe(last())).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
@@ -22,7 +21,7 @@ describe('Observable.prototype.last', () => {
     const e1subs =      '^    !';
     const expected =    '-----#';
 
-    expectObservable(e1.last()).toBe(expected, null, new Rx.EmptyError());
+    expectObservable(e1.pipe(last())).toBe(expected, null, new EmptyError());
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
@@ -31,7 +30,7 @@ describe('Observable.prototype.last', () => {
     const e1subs =   '(^!)';
     const expected = '#';
 
-    expectObservable(e1.last()).toBe(expected, null, new Rx.EmptyError());
+    expectObservable(e1.pipe(last())).toBe(expected, null, new EmptyError());
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
@@ -40,7 +39,7 @@ describe('Observable.prototype.last', () => {
     const e1subs =   '^';
     const expected = '-';
 
-    expectObservable(e1.last()).toBe(expected);
+    expectObservable(e1.pipe(last())).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
@@ -49,11 +48,7 @@ describe('Observable.prototype.last', () => {
     const e1subs =    '^             !';
     const expected =  '--------------(b|)';
 
-    const predicate = function (value) {
-      return value === 'b';
-    };
-
-    expectObservable(e1.last(predicate)).toBe(expected);
+    expectObservable(e1.pipe(last(value => value === 'b'))).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
@@ -63,7 +58,7 @@ describe('Observable.prototype.last', () => {
     const e1subs =    '^      !       ';
     const expected =  '--------       ';
 
-    expectObservable(e1.last(), unsub).toBe(expected);
+    expectObservable(e1.pipe(last()), unsub).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
@@ -73,10 +68,11 @@ describe('Observable.prototype.last', () => {
     const expected =  '--------       ';
     const unsub =     '       !       ';
 
-    const result = e1
-      .mergeMap((x: string) => Rx.Observable.of(x))
-      .last()
-      .mergeMap((x: string) => Rx.Observable.of(x));
+    const result = e1.pipe(
+      mergeMap(x => of(x)),
+      last(),
+      mergeMap(x => of(x)),
+    );
 
     expectObservable(result, unsub).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
@@ -87,7 +83,7 @@ describe('Observable.prototype.last', () => {
     const e1subs =   '(^!)';
     const expected = '(a|)';
 
-    expectObservable(e1.last(null, 'a')).toBe(expected);
+    expectObservable(e1.pipe(last(null, 'a'))).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
@@ -96,7 +92,7 @@ describe('Observable.prototype.last', () => {
     const e1subs =       '^               !';
     const expected =     '----------------(d|)';
 
-    expectObservable(e1.last(null, 'x')).toBe(expected);
+    expectObservable(e1.pipe(last(null, 'x'))).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
@@ -105,15 +101,15 @@ describe('Observable.prototype.last', () => {
     const e1subs =      '^       !           ';
     const expected =    '--------#           ';
 
-    const predicate = function (x) {
+    const predicate = function (x: string) {
       if (x === 'c') {
         throw 'error';
       } else {
         return false;
       }
     };
 
-    expectObservable(e1.last(predicate)).toBe(expected);
+    expectObservable(e1.pipe(last(predicate))).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
@@ -130,48 +126,48 @@ describe('Observable.prototype.last', () => {
       const isBaz = (x: any): x is Baz => x && (<Baz>x).baz !== undefined;
 
       const foo: Foo = new Foo();
-      Observable.of(foo).last()
+      of(foo).pipe(last())
         .subscribe(x => x.baz); // x is Foo
-      Observable.of(foo).last(foo => foo.bar === 'name')
+      of(foo).last(foo => foo.bar === 'name')
         .subscribe(x => x.baz); // x is still Foo
-      Observable.of(foo).last(isBar)
+      of(foo).last(isBar)
         .subscribe(x => x.bar); // x is Bar!
 
       const foobar: Bar = new Foo(); // type is the interface, not the class
-      Observable.of(foobar).last()
+      of(foobar).pipe(last())
         .subscribe(x => x.bar); // x is Bar
-      Observable.of(foobar).last(foobar => foobar.bar === 'name')
+      of(foobar).pipe(last(foobar => foobar.bar === 'name'))
         .subscribe(x => x.bar); // x is still Bar
-      Observable.of(foobar).last(isBaz)
+      of(foobar).pipe(last(isBaz))
         .subscribe(x => x.baz); // x is Baz!
 
       const barish = { bar: 'quack', baz: 42 }; // type can quack like a Bar
-      Observable.of(barish).last()
+      of(barish).pipe(last())
         .subscribe(x => x.baz); // x is still { bar: string; baz: number; }
-      Observable.of(barish).last(x => x.bar === 'quack')
+      of(barish).pipe(last(x => x.bar === 'quack'))
         .subscribe(x => x.bar); // x is still { bar: string; baz: number; }
-      Observable.of(barish).last(isBar)
+      of(barish).pipe(last(isBar))
         .subscribe(x => x.bar); // x is Bar!
     }
 
     // type guards with primitive types
     {
-      const xs: Rx.Observable<string | number> = Observable.from([ 1, 'aaa', 3, 'bb' ]);
+      const xs: Observable<string | number> = from([ 1, 'aaa', 3, 'bb' ]);
 
       // This type guard will narrow a `string | number` to a string in the examples below
       const isString = (x: string | number): x is string => typeof x === 'string';
 
       // missing predicate preserves the type
-      xs.last().subscribe(x => x); // x is still string | number
+      xs.pipe(last()).subscribe(x => x); // x is still string | number
 
       // After the type guard `last` predicates, the type is narrowed to string
-      xs.last(isString)
+      xs.pipe(last(isString))
         .subscribe(s => s.length); // s is string
-      xs.last(isString, s => s.substr(0)) // s is string in predicate
+      xs.pipe(last(isString, s => s.substr(0))) // s is string in predicate)
         .subscribe(s => s.length); // s is string
 
       // boolean predicates preserve the type
-      xs.last(x => typeof x === 'string')
+      xs.pipe(last(x => typeof x === 'string'))
         .subscribe(x => x); // x is still string | number
     }
 

src/internal/operators/last.ts

@@ -3,6 +3,11 @@ import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
 import { EmptyError } from '../util/EmptyError';
 import { MonoTypeOperatorFunction } from '../../internal/types';
+import { filter } from './filter';
+import { takeLast } from './takeLast';
+import { throwIfEmpty } from './throwIfEmpty';
+import { defaultIfEmpty } from './defaultIfEmpty';
+import { identity } from '../util/identity';
 
 /**
  * Returns an Observable that emits only the last item emitted by the source Observable.
@@ -21,74 +26,14 @@ import { MonoTypeOperatorFunction } from '../../internal/types';
  * from the source, or an NoSuchElementException if no such items are emitted.
  * @throws - Throws if no items that match the predicate are emitted by the source Observable.
  */
-export function last<T>(predicate?: (value: T, index: number, source: Observable<T>) => boolean,
-                        defaultValue?: T): MonoTypeOperatorFunction<T> {
-  return (source: Observable<T>) => source.lift(new LastOperator(predicate, defaultValue, source));
-}
-
-class LastOperator<T> implements Operator<T, T> {
-  constructor(private predicate: (value: T, index: number, source: Observable<T>) => boolean,
-              private defaultValue: any,
-              private source: Observable<T>) {
-  }
-
-  call(observer: Subscriber<T>, source: any): any {
-    return source.subscribe(new LastSubscriber(observer, this.predicate, this.defaultValue, this.source));
-  }
-}
-
-/**
- * We need this JSDoc comment for affecting ESDoc.
- * @ignore
- * @extends {Ignored}
- */
-class LastSubscriber<T> extends Subscriber<T> {
-  private lastValue: T;
-  private hasValue = false;
-  private index = 0;
-
-  constructor(destination: Subscriber<T>,
-              private predicate: (value: T, index: number, source: Observable<T>) => boolean,
-              private defaultValue: T,
-              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 {
-      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) {
-      this.lastValue = value;
-      this.hasValue = true;
-    }
-  }
-
-  protected _complete(): void {
-    const destination = this.destination;
-    if (this.hasValue) {
-      destination.next(this.lastValue);
-      destination.complete();
-    } else {
-      destination.error(new EmptyError);
-    }
-  }
+export function last<T>(
+  predicate?: (value: T, index: number, source: Observable<T>) => boolean,
+  defaultValue?: T
+): MonoTypeOperatorFunction<T> {
+  const hasDefaultValue = arguments.length >= 2;
+  return (source: Observable<T>) => source.pipe(
+    predicate ? filter((v, i) => predicate(v, i, source)) : identity,
+    takeLast(1),
+    hasDefaultValue ? defaultIfEmpty(defaultValue) : throwIfEmpty(() => new EmptyError()),
+  );
 }