You can emit only the final n items emitted by an Observable and ignore those items that come before them, by modifying the Observable with the TakeLast operator.
ReactiveXtakeLastTBD
TBD
You can emit only the final n items emitted by an Observable and ignore those items that precede
them, by modifying the Observable with the takeLast(n) operator. Note that this will
delay the emission of any item from the source Observable until the source Observable completes.
numbers = Observable.from([1, 2, 3, 4, 5, 6, 7, 8, 9]);
numbers.takeLast(2).subscribe(
{ println(it); }, // onNext
{ println("Error: " + it.getMessage()); }, // onError
{ println("Sequence complete"); } // onCompleted
);8 9 Sequence complete
This variant of takeLast does not by default operate on any particular
Scheduler.
takeLast(int)
There is also a variant of takeLast that takes a temporal duration rather than a quantity of
items. It emits only those items that are emitted during that final duration of the source
Observable’s lifespan. You set this duration by passing in a length of time and the time units
this length is denominated in as parameters to takeLast.
Note that this will delay the emission of any item from the source Observable until the source Observable completes.
This variant of takeLast by default operates on the computation
Scheduler, but you may also pass in a Scheduler of your choosing as an
optional third parameter.
takeLast(long,TimeUnit)takeLast(long,TimeUnit,Scheduler)
There is also a variant that combines the two methods. It emits the minimum of the number of items emitted during a specified time window or a particular count of items.
This variant of takeLast by default operates on the computation
Scheduler, but you may also pass in a Scheduler of your choosing as an
optional fourth parameter.
takeLast(int,long,TimeUnit)takeLast(int,long,TimeUnit,Scheduler)
There is also an operator called takeLastBuffer. It exists in the same set of variants
as described above for takeLast, and only differs in behavior by emitting its items not
individually but collected into a single List of items that is emitted as a single item.
takeLastBuffer(int)takeLastBuffer(long,TimeUnit)takeLastBuffer(long,TimeUnit,Scheduler)takeLastBuffer(int,long,TimeUnit)takeLastBuffer(int,long,TimeUnit,Scheduler)
You can emit only the final n items emitted by an Observable and ignore those items that precede
them, by modifying the Observable with the takeLast(n) operator. Note that this will
delay the emission of any item from the source Observable until the source Observable completes.
This variant of takeLast does not by default operate on any particular
Scheduler.
takeLast(int)
There is also a variant of takeLast that takes a temporal duration rather than a quantity of
items. It emits only those items that are emitted during that final duration of the source
Observable’s lifespan. You set this duration by passing in a length of time and the time units
this length is denominated in as parameters to takeLast.
Note that this will delay the emission of any item from the source Observable until the source Observable completes.
This variant of takeLast by default operates on the computation
Scheduler, but you may also pass in a Scheduler of your choosing as an
optional third parameter.
takeLast(long,TimeUnit)takeLast(long,TimeUnit,Scheduler)
There is also a variant that combines the two methods. It emits the minimum of the number of items emitted during a specified time window or a particular count of items.
This variant of takeLast by default operates on the computation
Scheduler, but you may also pass in a Scheduler of your choosing as an
optional fourth parameter.
takeLast(int,long,TimeUnit)takeLast(int,long,TimeUnit,Scheduler)
There is also an operator called takeLastBuffer. It exists in the same set of variants
as described above for takeLast, and only differs in behavior by emitting its items not
individually but collected into a single List of items that is emitted as a single item.
takeLastBuffer(int)takeLastBuffer(long,TimeUnit)takeLastBuffer(long,TimeUnit,Scheduler)takeLastBuffer(int,long,TimeUnit)takeLastBuffer(int,long,TimeUnit,Scheduler)
You can emit only the final n items emitted by an Observable and ignore those items that precede
them, by modifying the Observable with the takeLast(n) operator. Note that this will
delay the emission of any item from the source Observable until that Observable completes.
var source = Rx.Observable.range(0, 5)
.takeLast(3);
var subscription = source.subscribe(
function (x) { console.log('Next: ' + x); },
function (err) { console.log('Error: ' + err); },
function () { console.log('Completed'); });Next: 2 Next: 3 Next: 4 Completed
takeLast is found in each of the following distributions:
rx.jsrx.alljsrx.all.compatjsrx.compat.jsrx.lite.jsrx.lite.compat.js
The takeLastWithTime operator takes a temporal duration rather than a quantity of items. It
emits only those items that are emitted during that final duration of the source Observable’s
lifespan. You set this duration by passing in a number of milliseconds as a parameter to
takeLastWithTime.
Note that the mechanism by which this is implemented will delay the emission of any item from the source Observable until that Observable completes.
takeLastWithTime by default operates the timer on the timeout
Scheduler and emits items on the currentThread Scheduler,
but you may also pass in Schedulers of your choosing to override these, as an optional second and third
parameters, respectively.
var source = Rx.Observable.timer(0, 1000)
.take(10)
.takeLastWithTime(5000);
var subscription = source.subscribe(
function (x) { console.log('Next: ' + x); },
function (err) { console.log('Error: ' + err); },
function () { console.log('Completed'); });Next: 5 Next: 6 Next: 7 Next: 8 Next: 9 Completed
takeLastWithTime is found in each of the following distributions:
rx.all.jsrx.all.compat.jsrx.time.js (requires rx.js or rx.compat.js)rx.lite.jsrx.lite.compat.js
There is also an operator called takeLastBuffer. It differs in behavior from
takeLast by emitting its items not individually but collected into a single array of items
that is emitted as a single item.
var source = Rx.Observable.range(0, 5)
.takeLastBuffer(3);
var subscription = source.subscribe(
function (x) { console.log('Next: ' + x); },
function (err) { console.log('Error: ' + err); },
function () { console.log('Completed'); });Next: 2,3,4 Completed
takeLastBuffer is found in each of the following distributions:
rx.jsrx.all.jsrx.all.compat.jsrx.compat.jsrx.lite.jsrx.lite.compat.js
takeLastBuffer also has its duration-based variant, takeLastBufferWithTime,
which is similar to takeLastWithTime except that it emits its items not individually but
collected into a single array of items that is emitted as a single item.
var source = Rx.Observable
.timer(0, 1000)
.take(10)
.takeLastBufferWithTime(5000);
var subscription = source.subscribe(
function (x) { console.log('Next: ' + x); },
function (err) { console.log('Error: ' + err); },
function () { console.log('Completed'); });Next: 5,6,7,8,9 Completed
takeLastBufferWithTime is found in each of the following distributions:
rx.jsrx.all.jsrx.all.compat.jsrx.compat.jsrx.time.js (requires rx.js or rx.compat.js)rx.lite.jsrx.lite.compat.jsTBD
RxPHP implements this operator as takeLast.
Returns a specified number of contiguous elements from the end of an observable sequence.
//from https://github.com/ReactiveX/RxPHP/blob/master/demo/take/takeLast.php
$source = \Rx\Observable::range(0, 5)
->takeLast(3);
$source->subscribe($stdoutObserver);
Next value: 2
Next value: 3
Next value: 4
Complete!
TBD