added methods taking key selectors

Author: Joachim Hofer

rxjava-core/src/main/java/rx/operators/OperationDistinctUntilChanged.java

@@ -22,9 +22,6 @@
 import static rx.Observable.empty;
 import static rx.Observable.from;
 
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.InOrder;
@@ -36,35 +33,49 @@
 import rx.Subscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
+import rx.util.functions.Func1;
+import rx.util.functions.Functions;
 
 /**
  * Returns an Observable that emits the first item emitted by the source
  * Observable, or a default value if the source emits nothing.
  */
 public final class OperationDistinctUntilChanged {
 
+    /**
+     * Returns an Observable that emits all sequentially distinct items emitted by the source
+     * @param source
+     *            The source Observable to emit the sequentially distinct items for.
+     * @return A subscription function for creating the target Observable.
+     */
+    public static <T, U> OnSubscribeFunc<T> distinctUntilChanged(Observable<? extends T> source, Func1<? super T, ? extends U> keySelector) {
+        return new DistinctUntilChanged<T, U>(source, keySelector);
+    }
+    
     /**
      * Returns an Observable that emits all sequentially distinct items emitted by the source
      * @param source
      *            The source Observable to emit the sequentially distinct items for.
      * @return A subscription function for creating the target Observable.
      */
     public static <T> OnSubscribeFunc<T> distinctUntilChanged(Observable<? extends T> source) {
-        return new DistinctUntilChanged<T>(source);
+        return new DistinctUntilChanged<T, T>(source, Functions.<T>identity());
     }
 
-    private static class DistinctUntilChanged<T> implements OnSubscribeFunc<T> {
+    private static class DistinctUntilChanged<T, U> implements OnSubscribeFunc<T> {
         private final Observable<? extends T> source;
-
-        private DistinctUntilChanged(Observable<? extends T> source) {
+        private final Func1<? super T, ? extends U> keySelector;
+        
+        private DistinctUntilChanged(Observable<? extends T> source, Func1<? super T, ? extends U> keySelector) {
             this.source = source;
+            this.keySelector = keySelector;
         }
 
         @Override
         public Subscription onSubscribe(final Observer<? super T> observer) {
             final Subscription sourceSub = source.subscribe(new Observer<T>() {
-                private final AtomicReference<T> lastEmittedValue = new AtomicReference<T>(null);
-                private final AtomicBoolean hasEmitted = new AtomicBoolean();
+                private U lastEmittedKey;
+                private boolean hasEmitted;
 
                 @Override
                 public void onCompleted() {
@@ -78,16 +89,27 @@ public void onError(Throwable e) {
 
                 @Override
                 public void onNext(T next) {
-                    boolean hasAlreadyEmitted = hasEmitted.getAndSet(true);
-                    T lastEmitted = lastEmittedValue.getAndSet(next);
-                    if (!hasAlreadyEmitted) {
-                        observer.onNext(next);
-                    } else if (lastEmitted == null) {
-                        if (next != null) {
+                    U lastKey = lastEmittedKey;
+                    try {
+                        U nextKey = keySelector.call(next);
+                        lastEmittedKey = nextKey;
+                        if (!hasEmitted) {
+                            hasEmitted = true;
                             observer.onNext(next);
+                        } else {
+                            if (lastKey == null) {
+                                if (nextKey != null) {
+                                    observer.onNext(next);
+                                }
+                            } else {
+                                if (!lastKey.equals(nextKey)) {
+                                    observer.onNext(next);
+                                }
+                            }
                         }
-                    } else if (!lastEmitted.equals(next)) {
-                        observer.onNext(next);
+                    } catch (Throwable t) {
+                        // keySelector is a user function, may throw something
+                        observer.onError(t);
                     }
                 }
             });
@@ -105,6 +127,14 @@ public static class UnitTest {
         @Mock
         Observer<? super String> w;
 
+        // nulls lead to exceptions
+        final Func1<String, String> TO_UPPER_WITH_EXCEPTION = new Func1<String, String>() {
+            @Override
+            public String call(String s) {
+                return s.toUpperCase();
+            }
+        };
+        
         @Before
         public void before() {
             initMocks(this);
@@ -120,6 +150,16 @@ public void testDistinctUntilChangedOfNone() {
             verify(w, times(1)).onCompleted();
         }
 
+        @Test
+        public void testDistinctUntilChangedOfNoneWithKeySelector() {
+            Observable<String> src = empty();
+            create(distinctUntilChanged(src, TO_UPPER_WITH_EXCEPTION)).subscribe(w);
+
+            verify(w, never()).onNext(anyString());
+            verify(w, never()).onError(any(Throwable.class));
+            verify(w, times(1)).onCompleted();
+        }
+
         @Test
         public void testDistinctUntilChangedOfNormalSource() {
             Observable<String> src = from("a", "b", "c", "c", "c", "b", "b", "a", "e");
@@ -137,6 +177,23 @@ public void testDistinctUntilChangedOfNormalSource() {
             verify(w, never()).onError(any(Throwable.class));
         }
 
+        @Test
+        public void testDistinctUntilChangedOfNormalSourceWithKeySelector() {
+            Observable<String> src = from("a", "b", "c", "C", "c", "B", "b", "a", "e");
+            create(distinctUntilChanged(src, TO_UPPER_WITH_EXCEPTION)).subscribe(w);
+
+            InOrder inOrder = inOrder(w); 
+            inOrder.verify(w, times(1)).onNext("a");
+            inOrder.verify(w, times(1)).onNext("b");
+            inOrder.verify(w, times(1)).onNext("c");
+            inOrder.verify(w, times(1)).onNext("B");
+            inOrder.verify(w, times(1)).onNext("a");
+            inOrder.verify(w, times(1)).onNext("e");
+            inOrder.verify(w, times(1)).onCompleted();
+            inOrder.verify(w, never()).onNext(anyString());
+            verify(w, never()).onError(any(Throwable.class));
+        }
+
         @Test
         public void testDistinctUntilChangedOfSourceWithNulls() {
             Observable<String> src = from(null, "a", "a", null, null, "b", null, null);
@@ -152,5 +209,18 @@ public void testDistinctUntilChangedOfSourceWithNulls() {
             inOrder.verify(w, never()).onNext(anyString());
             verify(w, never()).onError(any(Throwable.class));
         }
+
+        @Test
+        public void testDistinctUntilChangedOfSourceWithExceptionsFromKeySelector() {
+            Observable<String> src = from("a", "b", null, "c");
+            create(distinctUntilChanged(src, TO_UPPER_WITH_EXCEPTION)).subscribe(w);
+
+            InOrder inOrder = inOrder(w); 
+            inOrder.verify(w, times(1)).onNext("a");
+            inOrder.verify(w, times(1)).onNext("b");
+            verify(w, times(1)).onError(any(NullPointerException.class));
+            inOrder.verify(w, never()).onNext(anyString());
+            inOrder.verify(w, never()).onCompleted();
+        }
     }
 }