What is the difference between the RxJS operators delay, timer and interval?

Question

The behavior of delay, timer, interval operators seems to be the same in RxJS. Can anyone let me know what is the difference between those operators?

user_6513921 · Answer

`delay`

It's an operator.
Delays emitted notifications from a source observable by a given time.
Eg. of('e').pipe(delay(2000)).subscribe(...) - emit e after 2 seconds.

Demo

var { of } = rxjs;
var { tap, delay } = rxjs.operators;

var displayBox = document.getElementById('display');

of('e').pipe(
  tap(_ => displayBox.innerHTML = 'Source observable emitted.'),
  delay(2000)
).subscribe(
  _ => displayBox.innerHTML += '
Received emission in subscription.',
  null,
  () => displayBox.innerHTML += '
Observable complete.'
);

`interval`

It's a function.
Emits every given time period.
Important: The first emission is after the time period.
Eg. interval(2000).subscribe(...) - starting after 2 seconds, emit every 2 seconds.

Demo

var { interval } = rxjs;

var displayBox = document.getElementById('display');

interval(2000).subscribe(  // won't complete
  value => displayBox.innerHTML += '
Received emission in subscription: ' + value
);

`timer`

It's a function.
1^st variant timer(initialDelay) - emit once after the delay and complete.
- Eg. timer(2000).subscribe(...) - emit after 2 seconds and complete.
2^nd variant timer(initialDelay, period) - starting after initial delay, emit every given time period.
- Eg. timer(0, 2000).subscribe(...) - starting immediately, emit every 2 seconds.
- Eg. timer(2000, 2000).subscribe(...) - starting after 2 seconds, emit every 2 seconds (similar to interval(2000)).

Demo

var { timer } = rxjs;

var timerNoDelayBox = document.getElementById('timerNoDelay');
var timerDelayBox = document.getElementById('timerDelay');

// w/o initial delay
timer(0, 2000).subscribe(  // won't complete
  value => timerNoDelayBox.innerHTML += '
Received emission in subscription: ' + value
);

// w/ initial delay
timer(2000, 2000).subscribe(  // won't complete
  value => timerDelayBox.innerHTML += '
Received emission in subscription: ' + value
);

table {
  font-family: arial, sans-serif;
  border-collapse: collapse;
  width: 100%;
}

td, th {
  border: 1px solid #dddddd;
  text-align: left;
  padding: 8px;
  width: 50%;
}




  
    Timer w/o initial delay
    Timer w/ initial delay

user_14151275 · Answer

The following images are taken from https://rxmarbles.com, a site I really recommend when working with RxJS:

Delay

Delays the emission of items from the source Observable (top line).

Timer

Creates an Observable that starts emitting after a given time and emits increasing numbers after each period of time thereafter. There is no source/input Observable here.

Interval

Creates an Observable that emits sequential numbers every specified interval of time. There is no source/input Observable here as well.

user_9632621 · Answer

timer

The function signature:

export function timer(
  dueTime: number | Date = 0,
  periodOrScheduler?: number | SchedulerLike,
  scheduler?: SchedulerLike
)

dueTime - it can either be an integer or a Date object and represents the ms that have to pass before emitting the first value(0)
periodOrScheduler - the period of time between emissions or a scheduler
scheduler - defaults to asyncScheduler

The implementation is as follows:

return new Observable((subscriber) => {

  // scheduler!.now() - usually defaults to `Date.now()`
  const due = Math.max(0, isValidDate(dueTime) ? +dueTime - scheduler!.now() : dueTime);

  return scheduler!.schedule(dispatch as any, due, {
    counter: 0,
    period,
    subscriber,
  });
});

The scheduler is in charge of scheduling actions, where each action has its own task. An action can reschedule itself if needed.

In this case

return scheduler!.schedule(dispatch as any, due, {
  counter: 0,
  period,
  subscriber,
});

The action's task is the dispatch function and will take place in due milliseconds. The third object can be thought of as the state of the action, which is useful when the action is rescheduling itself multiple times(as it happens when the period is defined).

So, after due milliseconds, this will happen:

// this is the action's task
function dispatch(this: SchedulerAction, state: TimerState) {
  const { period, subscriber } = state;
  const counter = state.counter++;
  
  // sending the data to the subscriber
  subscriber.next(counter);

  // might be closed due `completion`
  if (!subscriber.closed) {
    if (period < 0) {
      // Periods scheduled with a negative number will just complete.
      
      // if used like `timer(X)`, the `period` arg defaults to -1
      // so, the action won't reschedule itself
      return subscriber.complete();
    }

    // here it's used like `timer(X, Y)` (X, Y - numbers)
    // the action reschedules itself after `period` ms, **NOT** `due` ms
    // we can also see why having a state is useful - we can know which values to emit
    this.schedule(state, period);
  }
}

interval

This function is very similar to timer, with the exception that it only has the period feature. Because of that, you can't pass in a date object, as you'd do with timer: timer(new Date(...)).

The function signature:

export function interval(
  period = 0,
  scheduler: SchedulerLike = async
): Observable { /* ... */ }

The scheduler always plays the same role: it keeps track of actions, where each action has something to.

In this case, the action's task is:

function dispatch(this: SchedulerAction, state: IntervalState) {
  const { subscriber, counter, period } = state;
  subscriber.next(counter);
  this.schedule({ subscriber, counter: counter + 1, period }, period);
}

As you can see, it will keep on emitting values, which means that the consumer(with the help of other operators) is the only one which can decide when to stop receiving values.

Actions scheduled with the asyncScheduler can be thought of as functions that will run with the help of setInterval.

delay

This is an operator, which is a function that returns a function whose arguments is an observable(the source) and whose return type is another observable.

The signature:

export function delay(delay: number | Date, scheduler: SchedulerLike = async): MonoTypeOperatorFunction { /* ... */ }

As a side note, MonoTypeOperatorFunction means that the type of the source observable will be the same as the one of the returned observable.

Before receiving any values, the DelaySubscriber is considered to be inactive. It achieves its functionality by keeping a queue of delayed events. It also makes use of the scheduler(asyncScheduler by default) to schedule actions.

So, when it receives the first next notification, it becomes active:

protected _next(value: T) {
  const scheduler = this.scheduler;
  
  // the first arg is the time(in ms) when the delay expires
  const message = new DelayMessage(scheduler.now() + this.delay, value);
  this.queue.push(message);
  if (this.active === false) {
    this._schedule(scheduler);
  }
}

What this._schedule(scheduler) does is to schedule the action whose job is to go through the queue items and determine the ones which have their delay expired:

private _schedule(scheduler: SchedulerLike): void {
  this.active = true;
  const { destination } = this;

  // `this.delay` - the arguments passed to `delay()` operator
  destination.add(
    scheduler.schedule>(DelaySubscriber.dispatch as any, this.delay, {
      source: this,
      destination,
      scheduler,
    } as DelayState)
  );
}

And the action's job is represented by DelaySubscriber.dispatch:

private static dispatch(this: SchedulerAction>, state: DelayState): void {
  const source = state.source;
  const queue = source.queue;
  const scheduler = state.scheduler;
  const destination = state.destination;

  // iterating over the queue and determining which events should not be in the queue anymore
  while (queue.length > 0 && queue[0].time - scheduler.now() <= 0) {
    destination.next(queue.shift()!.value);
  }

  if (queue.length > 0) {
    const delay = Math.max(0, queue[0].time - scheduler.now());
    this.schedule(state, delay);
  } else if (source.isStopped) {
    source.destination.complete();
    source.active = false;
  } else {
    this.unsubscribe();
    source.active = false;
  }
}

queue[0].time - scheduler.now() <= 0 is the interesting part here. It would be the same as reading: while the event's time has expired Note that the queue item was added with certain time values, which was that current time(scheduler.now() + this.delay). Now, the value of scheduler.now() will be different, since the action has been scheduled to do its task at a later point in time.

So, let's say the first emission arrives at moment X and let's assume the delay value is 400 ms. Then, an action will be scheduled at X + delay, which is X + 400. But 200 ms after, which is X + 200, another value arrives. Since the DelaySubscriber is active, the only thing it can do is to add a new event to the queue, with the time value of X + 200 + 400(once again, 400 is the value of delay).

Then, at X + 400 the only one scheduled actions performs its task. The queue would look like this:

[
  { time: `X + 400`, value: /* ... */ },
  { time: `X + 400 + 200`, value: /* ... */ },
]

But since the action is now executing at X + 400, it means that scheduler.now() is X + 400. This means that for the first item, the queue[0].time - scheduler.now() <= 0 is true, which means that its value will be passed along to the next subscriber and the item will be removed from the queue.

Next, we have queue[0].time - scheduler.now() <= 0 for the second item, which is X + 400 + 200 - (X + 400) <= 0, which is false, so the loop will stop there.

Let's see what happens next:

// this is the current case
if (queue.length > 0) {
  // determining the amount of time left from the current item
  // since the queue's first item is the one because of whom the loop started
  // this is the way to determine the time left
  const delay = Math.max(0, queue[0].time - scheduler.now());
  this.schedule(state, delay);
} else if (source.isStopped) {
  // if the subscriber completed
  source.destination.complete();
  source.active = false;
} else {
  // becomes inactive
  // NOTE: `this.unsubscribe()` does NOT refer to the source, but to the scheduler
  // basically, the scheduler will forget about this action
  this.unsubscribe();
  source.active = false;
}

What happens if the DelaySubscriber receives an error or a complete notification?

On error

protected _error(err: any) {
  this.queue.length = 0;
  this.destination.error(err);
  this.unsubscribe();
}

we can see that the queue will be emptied and the error notification will be passed along to the next subscriber, which might mean the end of the entire stream.

On complete

protected _complete() {
  if (this.queue.length === 0) {
    this.destination.complete();
  }
  this.unsubscribe();
}

it will first check if the queue is empty. If that's the case, then it will immediately pass along the complete notification. Otherwise, it will unsubscribe, so that when the queue finally becomes empty, the

else if (source.isStopped) {
  source.destination.complete();
  source.active = false;
}

block will be reached and the complete notification will be finally sent to the next subscriber in the chain.

Here would be an example:

const src$ = new Observable(s => {
  s.next(1);

  setTimeout(() => {
    console.log('sending 2');
    s.next(2);
    s.complete();
  }, 1000);
});

src$.pipe(
  delay(2000)
).subscribe(console.log, null, () => console.warn('completed!'))

StackBlitz.

What is the difference between the RxJS operators delay, timer and interval?

3 Answers

delay

interval

timer

Delay

Timer

Interval

timer

interval

delay

`delay`

`interval`

`timer`