Skip to content

1.x: added distinctUntilChanged(comparator) #4001

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 3 commits into from
Closed

1.x: added distinctUntilChanged(comparator) #4001

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ceb6ebc
added distinctUntilChanged(comparator)
Nexen23 Jun 9, 2016
9e90bb9
moved distintUntilChanged(comparator) implementation to internal package
Nexen23 Jun 9, 2016
2f64e23
added tests
Nexen23 Jun 9, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions src/main/java/rx/Observable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4761,6 +4761,27 @@ public final Observable<T> distinctUntilChanged() {
public final <U> Observable<T> distinctUntilChanged(Func1<? super T, ? extends U> keySelector) {
return lift(new OperatorDistinctUntilChanged<T, U>(keySelector));
}

/**
* Returns an Observable that emits all items emitted by the source Observable that are distinct from their
* immediate predecessors, according to a comparator function.
* <p>
* <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/distinctUntilChanged.key.png" alt="">
* <dl>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code distinctUntilChanged} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
*
* @param comparator
* a function that compares two successive items and decide whether an item
* is distinct from another one or not
* @return an Observable that emits those items from the source Observable whose keys are distinct from
* those of their immediate predecessors
* @see <a href="http://reactivex.io/documentation/operators/distinct.html">ReactiveX operators documentation: Distinct</a>
*/
public final Observable<T> distinctUntilChanged(Func2<? super T, ? super T, Boolean> comparator) {
return lift(new OperatorDistinctUntilChanged<T, Void>(comparator));
}

/**
* Modifies the source Observable so that it invokes an action when it calls {@code onCompleted}.
Expand Down
128 changes: 90 additions & 38 deletions src/main/java/rx/internal/operators/OperatorDistinctUntilChanged.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import rx.Subscriber;
import rx.exceptions.Exceptions;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.internal.util.UtilityFunctions;

/**
Expand All @@ -27,7 +28,7 @@
* @param <U> the key type
*/
public final class OperatorDistinctUntilChanged<T, U> implements Operator<T, T> {
final Func1<? super T, ? extends U> keySelector;
final Func1<Subscriber<? super T>, Subscriber<T>> subscriberProvider;

private static class Holder {
static final OperatorDistinctUntilChanged<?,?> INSTANCE = new OperatorDistinctUntilChanged<Object,Object>(UtilityFunctions.identity());
Expand All @@ -46,50 +47,101 @@ public static <T> OperatorDistinctUntilChanged<T, T> instance() {
return (OperatorDistinctUntilChanged<T, T>) Holder.INSTANCE;
}

public OperatorDistinctUntilChanged(Func1<? super T, ? extends U> keySelector) {
this.keySelector = keySelector;
}

@Override
public Subscriber<? super T> call(final Subscriber<? super T> child) {
return new Subscriber<T>(child) {
U previousKey;
boolean hasPrevious;
public OperatorDistinctUntilChanged(final Func1<? super T, ? extends U> keySelector) {
subscriberProvider = new Func1<Subscriber<? super T>, Subscriber<T>>() {
@Override
public void onNext(T t) {
U currentKey = previousKey;
final U key;
try {
key = keySelector.call(t);
} catch (Throwable e) {
Exceptions.throwOrReport(e, child, t);
return;
}
previousKey = key;

if (hasPrevious) {
if (!(currentKey == key || (key != null && key.equals(currentKey)))) {
child.onNext(t);
} else {
request(1);
public Subscriber<T> call(final Subscriber<? super T> child) {
return new Subscriber<T>(child) {
U previousKey;
boolean hasPrevious;

@Override
public void onNext(T t) {
U currentKey = previousKey;
final U key;
try {
key = keySelector.call(t);
} catch (Throwable e) {
Exceptions.throwOrReport(e, child, t);
return;
}
previousKey = key;

if (hasPrevious) {
if (!(currentKey == key || (key != null && key.equals(currentKey)))) {
child.onNext(t);
} else {
request(1);
}
} else {
hasPrevious = true;
child.onNext(t);
}
}
} else {
hasPrevious = true;
child.onNext(t);
}
}

@Override
public void onError(Throwable e) {
child.onError(e);
@Override
public void onError(Throwable e) {
child.onError(e);
}

@Override
public void onCompleted() {
child.onCompleted();
}

};
}
};
}

public OperatorDistinctUntilChanged(final Func2<? super T, ? super T, Boolean> comparator) {
subscriberProvider = new Func1<Subscriber<? super T>, Subscriber<T>>() {
@Override
public void onCompleted() {
child.onCompleted();
public Subscriber<T> call(final Subscriber<? super T> child) {
return new Subscriber<T>(child) {
T previousValue;
boolean hasPrevious;

@Override
public void onNext(T t) {
if (hasPrevious) {
boolean valuesAreEqual;
try {
valuesAreEqual = comparator.call(previousValue, t);
} catch (Throwable e) {
Exceptions.throwOrReport(e, child, t);
return;
}
if (valuesAreEqual) {
request(1);
} else {
previousValue = t;
child.onNext(t);
}
} else {
hasPrevious = true;
previousValue = t;
child.onNext(t);
}
}

@Override
public void onError(Throwable e) {
child.onError(e);
}

@Override
public void onCompleted() {
child.onCompleted();
}

};
}

};
}


@Override
public Subscriber<? super T> call(final Subscriber<? super T> child) {
return subscriberProvider.call(child);
}
}
107 changes: 90 additions & 17 deletions src/test/java/rx/internal/operators/OperatorDistinctUntilChangedTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,26 +15,27 @@
*/
package rx.internal.operators;

import static org.junit.Assert.assertFalse;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.MockitoAnnotations.initMocks;

import java.util.concurrent.atomic.AtomicBoolean;

import org.junit.Before;
import org.junit.Test;
import org.mockito.InOrder;
import org.mockito.Mock;

import java.util.concurrent.atomic.AtomicBoolean;

import rx.Observable;
import rx.Observer;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.functions.Func2;

import static org.junit.Assert.assertFalse;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.MockitoAnnotations.initMocks;

public class OperatorDistinctUntilChangedTest {

Expand All @@ -44,7 +45,7 @@ public class OperatorDistinctUntilChangedTest {
private Observer<String> w2;

// nulls lead to exceptions
private final static Func1<String, String> TO_UPPER_WITH_EXCEPTION = new Func1<String, String>() {
private final static Func1<String, String> KEYSELECTOR_TO_UPPER_WITH_EXCEPTION = new Func1<String, String>() {
@Override
public String call(String s) {
if (s.equals("x")) {
Expand All @@ -54,13 +55,29 @@ public String call(String s) {
}
};

private final static Func1<String, String> THROWS_NON_FATAL = new Func1<String, String>() {
private final static Func1<String, String> KEYSELECTOR_THROWS_NON_FATAL = new Func1<String, String>() {
@Override
public String call(String s) {
throw new RuntimeException();
}
};

private final static Func2<String, String, Boolean> COMPARATOR_TO_UPPER_WITH_EXCEPTION = new Func2<String, String, Boolean>() {
@Override
public Boolean call(String s1, String s2) {
s1 = KEYSELECTOR_TO_UPPER_WITH_EXCEPTION.call(s1);
s2 = KEYSELECTOR_TO_UPPER_WITH_EXCEPTION.call(s2);
return s1.equals(s2);
}
};

private final static Func2<String, String, Boolean> COMPARATOR_THROWS_NON_FATAL = new Func2<String, String, Boolean>() {
@Override
public Boolean call(String s1, String s2) {
throw new RuntimeException();
}
};

@Before
public void before() {
initMocks(this);
Expand All @@ -79,7 +96,17 @@ public void testDistinctUntilChangedOfNone() {
@Test
public void testDistinctUntilChangedOfNoneWithKeySelector() {
Observable<String> src = Observable.empty();
src.distinctUntilChanged(TO_UPPER_WITH_EXCEPTION).subscribe(w);
src.distinctUntilChanged(KEYSELECTOR_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 testDistinctUntilChangedOfNoneWithComparator() {
Observable<String> src = Observable.empty();
src.distinctUntilChanged(COMPARATOR_TO_UPPER_WITH_EXCEPTION).subscribe(w);

verify(w, never()).onNext(anyString());
verify(w, never()).onError(any(Throwable.class));
Expand All @@ -106,7 +133,24 @@ public void testDistinctUntilChangedOfNormalSource() {
@Test
public void testDistinctUntilChangedOfNormalSourceWithKeySelector() {
Observable<String> src = Observable.just("a", "b", "c", "C", "c", "B", "b", "a", "e");
src.distinctUntilChanged(TO_UPPER_WITH_EXCEPTION).subscribe(w);
src.distinctUntilChanged(KEYSELECTOR_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 testDistinctUntilChangedOfNormalSourceWithComparator() {
Observable<String> src = Observable.just("a", "b", "c", "C", "c", "B", "b", "a", "e");
src.distinctUntilChanged(COMPARATOR_TO_UPPER_WITH_EXCEPTION).subscribe(w);

InOrder inOrder = inOrder(w);
inOrder.verify(w, times(1)).onNext("a");
Expand Down Expand Up @@ -139,7 +183,20 @@ public void testDistinctUntilChangedOfSourceWithNulls() {
@Test
public void testDistinctUntilChangedOfSourceWithExceptionsFromKeySelector() {
Observable<String> src = Observable.just("a", "b", null, "c");
src.distinctUntilChanged(TO_UPPER_WITH_EXCEPTION).subscribe(w);
src.distinctUntilChanged(KEYSELECTOR_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();
}

@Test
public void testDistinctUntilChangedOfSourceWithExceptionsFromComparator() {
Observable<String> src = Observable.just("a", "b", null, "c");
src.distinctUntilChanged(COMPARATOR_TO_UPPER_WITH_EXCEPTION).subscribe(w);

InOrder inOrder = inOrder(w);
inOrder.verify(w, times(1)).onNext("a");
Expand All @@ -160,8 +217,24 @@ public void call(Throwable t) {
errorOccurred.set(true);
}
})
.distinctUntilChanged(THROWS_NON_FATAL)
.distinctUntilChanged(KEYSELECTOR_THROWS_NON_FATAL)
.subscribe(w);
assertFalse(errorOccurred.get());
}

@Test
public void testDistinctUntilChangedWhenNonFatalExceptionThrownByComparatorIsNotReportedByUpstream() {
Observable<String> src = Observable.just("a", "b", null, "c");
final AtomicBoolean errorOccurred = new AtomicBoolean(false);
src
.doOnError(new Action1<Throwable>() {
@Override
public void call(Throwable t) {
errorOccurred.set(true);
}
})
.distinctUntilChanged(COMPARATOR_THROWS_NON_FATAL)
.subscribe(w);
assertFalse(errorOccurred.get());
}
}