fix(distinctUntilChanged): implement optional keySelector

spec/operators/distinctUntilChanged-spec.js

@@ -136,7 +136,7 @@ describe('Observable.prototype.distinctUntilChanged()', function () {
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
-  it('should emit once if comparer returns true always regardless of source emits', function () {
+  it('should emit once if comparator returns true always regardless of source emits', function () {
     var e1 =   hot('--a--b--c--d--e--f--|');
     var e1subs =   '^                   !';
     var expected = '--a-----------------|';
@@ -145,7 +145,7 @@ describe('Observable.prototype.distinctUntilChanged()', function () {
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
-  it('should emit all if comparer returns false always regardless of source emits', function () {
+  it('should emit all if comparator returns false always regardless of source emits', function () {
     var e1 =   hot('--a--a--a--a--a--a--|');
     var e1subs =   '^                   !';
     var expected = '--a--a--a--a--a--a--|';
@@ -154,30 +154,60 @@ describe('Observable.prototype.distinctUntilChanged()', function () {
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
-  it('should distinguish values by selector', function () {
+  it('should distinguish values by comparator', function () {
     var e1 =   hot('--a--b--c--d--e--f--|', {a: 1, b: 2, c: 3, d: 4, e: 5, f: 6});
     var e1subs =   '^                   !';
     var expected = '--a-----c-----e-----|';
-    var selector = function (x, y) {
+    var comparator = function (x, y) {
       return y % 2 === 0;
     };
 
-    expectObservable(e1.distinctUntilChanged(selector)).toBe(expected, {a: 1, c: 3, e: 5});
+    expectObservable(e1.distinctUntilChanged(comparator)).toBe(expected, {a: 1, c: 3, e: 5});
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
 
-  it('should raises error when comparer throws', function () {
+  it('should raises error when comparator throws', function () {
     var e1 =   hot('--a--b--c--d--e--f--|');
     var e1subs =   '^          !         ';
     var expected = '--a--b--c--#         ';
-    var selector = function (x, y) {
+    var comparator = function (x, y) {
       if (y === 'd') {
         throw 'error';
       }
       return x === y;
     };
 
-    expectObservable(e1.distinctUntilChanged(selector)).toBe(expected);
+    expectObservable(e1.distinctUntilChanged(comparator)).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
-});
+
+  it('should use the keySelector to pick comparator values', function () {
+    var e1 =   hot('--a--b--c--d--e--f--|', {a: 1, b: 2, c: 3, d: 4, e: 5, f: 6});
+    var e1subs =   '^                   !';
+    var expected = '--a--b-----d-----f--|';
+    var comparator = function (x, y) {
+      return y % 2 === 1;
+    };
+    var keySelector = function (x) {
+      return x % 2;
+    };
+
+    expectObservable(e1.distinctUntilChanged(comparator, keySelector)).toBe(expected, {a: 1, b: 2, d: 4, f: 6});
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+  });
+
+  it('should raises error when keySelector throws', function () {
+    var e1 =   hot('--a--b--c--d--e--f--|');
+    var e1subs =   '^          !         ';
+    var expected = '--a--b--c--#         ';
+    var keySelector = function (x) {
+      if (x === 'd') {
+        throw 'error';
+      }
+      return x;
+    };
+
+    expectObservable(e1.distinctUntilChanged(null, keySelector)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+  });
+});

src/operator/distinctUntilChanged.ts

@@ -3,49 +3,62 @@ import {Subscriber} from '../Subscriber';
 import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 
-export function distinctUntilChanged<T>(compare?: (x: T, y: T) => boolean) {
-  return this.lift(new DistinctUntilChangedOperator(compare));
+export function distinctUntilChanged<T>(compare?: (x: any, y: any) => boolean, keySelector?: (x: T) => any) {
+  return this.lift(new DistinctUntilChangedOperator(compare, keySelector));
 }
 
 class DistinctUntilChangedOperator<T, R> implements Operator<T, R> {
-  constructor(private compare: (x: T, y: T) => boolean) {
+  constructor(private compare: (x: any, y: any) => boolean,
+              private keySelector: (x: T) => any) {
   }
 
   call(subscriber: Subscriber<T>): Subscriber<T> {
-    return new DistinctUntilChangedSubscriber(subscriber, this.compare);
+    return new DistinctUntilChangedSubscriber(subscriber, this.compare, this.keySelector);
   }
 }
 
 class DistinctUntilChangedSubscriber<T> extends Subscriber<T> {
-  private value: T;
-  private hasValue: boolean = false;
+  private key: any;
+  private hasKey: boolean = false;
 
-  constructor(destination: Subscriber<T>, compare: (x: T, y: T) => boolean) {
+  constructor(destination: Subscriber<T>,
+              compare: (x: any, y: any) => boolean,
+              private keySelector: (x: T) => any) {
     super(destination);
     if (typeof compare === 'function') {
       this.compare = compare;
     }
   }
 
-  private compare(x: T, y: T): boolean {
+  private compare(x: any, y: any): boolean {
     return x === y;
   }
 
   _next(value: T): void {
+
+    const keySelector = this.keySelector;
+    let key: any = value;
+
+    if (keySelector) {
+      key = tryCatch(this.keySelector)(value);
+      if (key === errorObject) {
+        return this.destination.error(errorObject.e);
+      }
+    }
+
     let result: any = false;
 
-    if (this.hasValue) {
-      result = tryCatch(this.compare)(this.value, value);
+    if (this.hasKey) {
+      result = tryCatch(this.compare)(this.key, key);
       if (result === errorObject) {
-        this.destination.error(errorObject.e);
-        return;
+        return this.destination.error(errorObject.e);
       }
     } else {
-      this.hasValue = true;
+      this.hasKey = true;
     }
 
     if (Boolean(result) === false) {
-      this.value = value;
+      this.key = key;
       this.destination.next(value);
     }
   }