The Distinct operator filters an Observable by only allowing items through that have not already been emitted.
In some implementations there are variants that allow you to adjust the criteria by which two items are considered “distinct.” In some, there is a variant of the operator that only compares an item against its immediate predecessor for distinctness, thereby filtering only consecutive duplicate items from the sequence.
ReactiveXdistinctdistinctUntilChanged
RxGroovy implements this operator as distinct.
distinct()
There is also a variant of this operator that takes a function as a parameter. This function
operates on items emitted by the source Observable to generate a “key.” It is
these keys, then, and not the items themselves, that distinct will compare to
determine whether or not two items are distinct.
distinct(Func1)
RxGroovy also implements the distinctUntilChanged operator. It only compares
emitted items from the source Observable against their immediate predecessors in order to
determine whether or not they are distinct.
distinctUntilChanged()
As with distinct, there is also a version of distinctUntilChanged
that accepts a key selector function and that uses the resulting key to determine whether or
not two adjacently-emitted items are distinct.
distinctUntilChanged(Func1)
distinct and distinctUntilChanged do not by default operate on any
particular Scheduler.
RxJava implements this operator as distinct.
Observable.just(1, 2, 1, 1, 2, 3)
.distinct()
.subscribe(new Subscriber<Integer>() {
@Override
public void onNext(Integer item) {
System.out.println("Next: " + item);
}
@Override
public void onError(Throwable error) {
System.err.println("Error: " + error.getMessage());
}
@Override
public void onCompleted() {
System.out.println("Sequence complete.");
}
});Next: 1 Next: 2 Next: 3 Sequence complete.
distinct()
There is also a variant of this operator that takes a function as a parameter. This function
operates on items emitted by the source Observable to generate a “key.” It is
these keys, then, and not the items themselves, that distinct will compare to
determine whether or not two items are distinct.
distinct(Func1)
RxJava also implements the distinctUntilChanged operator. It only compares
emitted items from the source Observable against their immediate predecessors in order to
determine whether or not they are distinct.
distinctUntilChanged()
As with distinct, there is also a version of distinctUntilChanged
that accepts a key selector function and that uses the resulting key to determine whether or
not two adjacently-emitted items are distinct.
distinctUntilChanged(Func1)
distinct and distinctUntilChanged do not by default operate on any
particular Scheduler.
In RxJS, the distinct operator has two optional parameters:
false if they are distinct (an equality function is the default if you do not supply your own function here)
/* Without key selector */
var source = Rx.Observable.fromArray([
42, 24, 42, 24
])
.distinct();
var subscription = source.subscribe(
function (x) { console.log('Next: ' + x.toString()); },
function (err) { console.log('Error: ' + err); },
function () { console.log('Completed'); });Next: 42 Next: 24 Completed
/* With key selector */
var source = Rx.Observable.fromArray([
{value: 42}, {value: 24}, {value: 42}, {value: 24}
])
.distinct(function (x) { return x.value; });
var subscription = source.subscribe(
function (x) {
console.log('Next: ' + x.toString());
},
function (err) {
console.log('Error: ' + err);
},
function () {
console.log('Completed');
});
Next: { value: 42 }
Next: { value: 24 }
Completed
RxJS also has a distinctUntilChanged operator. It only compares emitted items
from the source Observable against their immediate predecessors in order to determine whether
or not they are distinct. It takes the same two optional parameters as the
distinct operator.
/* Without key selector */
var source = Rx.Observable.fromArray([
24, 42, 24, 24
])
.distinctUntilChanged();
var subscription = source.subscribe(
function (x) { console.log('Next: ' + x); },
function (err) { console.log('Error: ' + err); },
function () { console.log('Completed'); });Next: 24 Next: 42 Next: 24 Completed
/* With key selector */
var source = Rx.Observable.fromArray([
{value: 24}, {value: 42}, {value: 42}, {value: 24}
])
.distinctUntilChanged(function (x) { return x.value; });
var subscription = source.subscribe(
function (x) { console.log('Next: ' + x.toString()); },
function (err) { console.log('Error: ' + err); },
function () { console.log('Completed'); });
Next: { value: 24 }
Next: { value: 42 }
Next: { value: 24 }
Completed
distinct and distinctUntilChanged are found in each of the following
distributions:
rx.jsrx.all.jsrx.all.compat.jsrx.compat.jsrx.lite.jsrx.lite.compat.js
RxPHP implements this operator as distinct.
Returns an observable sequence that contains only distinct elements according to the keySelector and the comparer. Usage of this operator should be considered carefully due to the maintenance of an internal lookup structure which can grow large.
//from https://github.com/ReactiveX/RxPHP/blob/master/demo/distinct/distinct.php
$source = \Rx\Observable::fromArray([
42, 24, 42, 24
])->distinct();
$subscription = $source->subscribe($stdoutObserver);
Next value: 42
Next value: 24
Complete!
RxPHP also has an operator distinctKey.
Variant of distinct that takes a key selector
//from https://github.com/ReactiveX/RxPHP/blob/master/demo/distinct/distinctKey.php
$source = \Rx\Observable::fromArray([
['id' => '42'],
['id' => '24'],
['id' => '42'],
['id' => '24']
])
->distinctKey(function ($x) {
return $x['id'];
})
->map(function ($x) {
return $x['id'];
});
$subscription = $source->subscribe($stdoutObserver);
Next value: 42
Next value: 24
Complete!
RxPHP also has an operator distinctUntilChanged.
Returns an observable sequence that contains only distinct contiguous elements according to the keySelector and the comparer.
//from https://github.com/ReactiveX/RxPHP/blob/master/demo/distinct/distinctUntilChanged.php
$source = \Rx\Observable::fromArray([
24, 42, 24, 24
])->distinctUntilChanged();
$subscription = $source->subscribe($stdoutObserver);
Next value: 24
Next value: 42
Next value: 24
Complete!
RxPHP also has an operator distinctUntilKeyChanged.
Variant of distinctUntilChanged that takes a key selector and the comparer.
//from https://github.com/ReactiveX/RxPHP/blob/master/demo/distinct/distinctUntilKeyChanged.php
$source = \Rx\Observable::fromArray([
['id' => '24'],
['id' => '42'],
['id' => '24'],
['id' => '24']
])
->distinctUntilKeyChanged(function ($x) {
return $x['id'];
})
->map(function ($x) {
return $x['id'];
});
$subscription = $source->subscribe($stdoutObserver);
Next value: 24
Next value: 42
Next value: 24
Complete!