The Delay operator modifies its source Observable by pausing for a particular increment of time (that you specify) before emitting each of the source Observable’s items. This has the effect of shifting the entire sequence of items emitted by the Observable forward in time by that specified increment.
delay
delayWithSelector
TBD
delay delaySubscription
RxGroovy implements this operator as variants of delay and delaySubscription.
delaySubscription
The first variant of delay accepts parameters that define a duration of time (a quantity of time, and a TimeUnit that this quantity is denominated in). Each time the source Observable emits an item, delay starts a timer, and when that timer reaches the given duration, the Observable returned from delay emits the same item.
TimeUnit
Note that delay will not time-shift an onError notification in this fashion but it will forward such a notification immediately to its subscribers while dropping any pending onNext notifications. It will however time shift an onCompleted notification.
onError
onNext
onCompleted
By default this variant of delay operates on the computation Scheduler, but you can choose a different Scheduler by passing it in as an optional third parameter to delay
computation
delay(long,TimeUnit)
delay()
Another variant of delay does not use a constant delay duration, but sets its delay duration on a per-item basis by passing each item from the source Observable into a function that returns an Observable and then monitoring those Observables. When any such Observable emits a item or completes, the Observable returned by delay emits the associated item.
This variant of delay does not by default run on any particular Scheduler.
delay(Func1)
The variant of delay that uses a per-item Observable to set the delay has a variant that allows you to pass in a function that returns an Observable that acts as a delay timer for the subscription to the source Observable (in the absence of this, delay subscribes to the source Observable as soon as an observer subscribes to the Observable returned by delay).
delay(Func0,Func1)
There is also an operator with which you can delay the subscription to the source Observable: delaySubscription. It accepts parameters that define the amount of time to delay (a quantity of time, and a TimeUnit that this quantity is denominated in).
This variant of delay by default runs on the computation Scheduler, but you can choose a different Scheduler by passing it in as an optional third parameter to delaySubscription.
delaySubscription(long,TimeUnit)
delaySubscription(long,TimeUnit,Scheduler)
And there is a variant of delaySubscription that uses an Observable (returned by a function you supply) rather than a fixed duration in order to set the subscription delay.
This variant of delaySubscription does not by default run on any particular Scheduler.
delaySubscription(Func0)
RxJava implements this operator as variants of delay and delaySubscription.
Another variant of delay does not use a constant delay duration, but sets its delay duration on a per-item basis by passing each item from the source Observable into a function that returns an Observable and then monitoring those Observables. When any such Observable emits an item or completes, the Observable returned by delay emits the associated item.
delay delaySubscription delayWithSelector
In RxJS you can set the per-item delay in two ways: by passing a number of milliseconds into the delay operator (which will delay each emission by that amount of time), or by passing in a Date object (which will delay the beginning of the sequence of emissions until that absolute point in time).
Date
This operator operates by default on the timeout Scheduler, but you can override this by passing in another Scheduler as an optional second parameter.
timeout
var source = Rx.Observable.range(0, 3) .delay(new Date(Date.now() + 1000)); var subscription = source.subscribe( function (x) { console.log('Next: ' + x.toString()); }, function (err) { console.log('Error: ' + err); }, function () { console.log('Completed'); });
Next: 0 Next: 1 Next: 2 Completed
var source = Rx.Observable.range(0, 3) .delay(1000); var subscription = source.subscribe( function (x) { console.log('Next: ' + x.toString()); }, function (err) { console.log('Error: ' + err); }, function () { console.log('Completed'); });
delaySubscription is similar to delay but rather than timeshifting the emissions from the source Observable, it timeshifts the moment of subscription to that Observable. You pass to this operator a time value (either a Number, in which case this sets the number of milliseconds of delay, or a Date, in which case this sets an absolute future time at which delaySubscription will trigger the subscription). You may optionally pass a Scheduler as a second parameter, which delaySubscription will use to govern the delay period or trigger time.
Number
var start = Date.now(); var source = Rx.Observable.range(0, 3).delaySubscription(5000); var subscription = source.subscribe( function (x) { console.log('Next: %s, %s', x, Date.now() - start); }, function (err) { console.log('Error: ' + err); }, function () { console.log('Completed'); });
Next: 0, 5001 Next: 1, 5002 Next: 2, 5003 Completed
delayWithSelector is like delay but does not use a constant delay duration (or absolute time), but sets its delay duration on a per-item basis by passing each item from the source Observable into a function that returns an Observable and then monitoring those Observables. When any such Observable completes, the Observable returned by delay emits the associated item.
var source = Rx.Observable .range(0, 3) .delayWithSelector( function (x) { return Rx.Observable.timer(x * 400); }) .timeInterval() .map(function (x) { return x.value + ':' + x.interval; }); var subscription = source.subscribe( function (x) { console.log('Next: ' + x); }, function (err) { console.log('Error: ' + err); }, function () { console.log('Completed'); });
Next: 0:0 Next: 1:400 Next: 2:400 Completed
There is also a version of delayWithSelector that takes an additional (first) argument: an Observable that sets a delay before delayWithSelector subscribes to the source Observable after it itself is subscribed to.
var source = Rx.Observable .range(0, 3) .delayWithSelector( Rx.Observable.timer(300), function (x) { return Rx.Observable.timer(x * 400); } ) .timeInterval() .map(function (x) { return x.value + ':' + x.interval; }); var subscription = source.subscribe( function (x) { console.log('Next: ' + x); }, function (err) { console.log('Error: ' + err); }, function () { console.log('Completed'); });
Next: 0:300 Next: 1:400 Next: 2:400 Completed
delay, delaySubscription, and delayWithSelector require rx.lite.js or rx.lite.compat.js and are found in each of the following distributions:
rx.lite.js
rx.lite.compat.js
rx.all.js
rx.all.compat.js
rx.time.js
rx.js
rx.compat.js
Delay DelaySubscription
RxPHP implements this operator as delay.
Time shifts the observable sequence by dueTime. The relative time intervals between the values are preserved.
//from https://github.com/ReactiveX/RxPHP/blob/master/demo/delay/delay.php $loop = new \React\EventLoop\StreamSelectLoop(); $scheduler = new \Rx\Scheduler\EventLoopScheduler($loop); \Rx\Observable::interval(1000, $scheduler) ->doOnNext(function ($x) { echo "Side effect: " . $x . "\n"; }) ->delay(1000) ->take(5) ->subscribe($createStdoutObserver(), $scheduler); $loop->run();
Side effect: 0 Next value: 0 Side effect: 1 Next value: 1 Side effect: 2 Next value: 2 Side effect: 3 Next value: 3 Side effect: 4 Next value: 4 Complete!
delay delay_subscription delay_with_selector