I have a piece of code I'd like to execute periodically until all subscribers have unsubscribed.

// This function shall be called *once* per tick,
// no matter the quantity of subscriber.
function doSomething(val) {
    console.log("doing something");
    return val;
}

observable = Rx.Observable.timer(0, 1000).map(val => doSomething(val));

const first = observable.subscribe(val => console.log("first:", val));
const second = observable.subscribe(val => console.log("second:", val));

// After 1.5 seconds, stop first.
Rx.Observable.timer(1500).subscribe(_ => first.unsubscribe());
// After 2.5 seconds, stop second.
Rx.Observable.timer(2500).subscribe(_ => second.unsubscribe());

JSFiddle

My expected output would look like that:

doing something
first: 0
second: 0
doing something
first: 1
second: 1
doing something
second: 2
<nothing more>

However, the doSomething function is called twice when the two observable are called. Here is the actual output:

doing something
first: 0
doing something
second: 0
doing something
first: 1
doing something
second: 1
doing something
second: 2
<nothing more>

Am I doing a design mistake? Is there a way to do this?

I have a piece of code I'd like to execute periodically until all subscribers have unsubscribed.

// This function shall be called *once* per tick,
// no matter the quantity of subscriber.
function doSomething(val) {
    console.log("doing something");
    return val;
}

observable = Rx.Observable.timer(0, 1000).map(val => doSomething(val));

const first = observable.subscribe(val => console.log("first:", val));
const second = observable.subscribe(val => console.log("second:", val));

// After 1.5 seconds, stop first.
Rx.Observable.timer(1500).subscribe(_ => first.unsubscribe());
// After 2.5 seconds, stop second.
Rx.Observable.timer(2500).subscribe(_ => second.unsubscribe());

JSFiddle

My expected output would look like that:

doing something
first: 0
second: 0
doing something
first: 1
second: 1
doing something
second: 2
<nothing more>

However, the doSomething function is called twice when the two observable are called. Here is the actual output:

doing something
first: 0
doing something
second: 0
doing something
first: 1
doing something
second: 1
doing something
second: 2
<nothing more>

Am I doing a design mistake? Is there a way to do this?

• javascript
• rxjs
• observable

2 Answers 2

The behaviour you are seeing is correct. The observable returned by interval is cold. That is, no timer is created until an observer subscribes and, when one does, the timer that's created is specifically for that subscription.

The behaviour you were expecting can be effected using the share operator:

observable = Rx.Observable
  .timer(0, 1000)
  .map(val => doSomething(val))
  .share();

The share operator reference counts subscriptions and multicasts the source observable to multiple subscribers - so there will be only a single interval/timer, shared between the two subscribers.

For more information, you might find this article useful.

Live working example. Then you need to use Subjects. Plain Observables are unicast (it means each subscriber own an independent execution of the Observable). That is way each observer get called all the chain of the execution you have.

observable = Rx.Observable.timer(0, 1000)
  .map(val => doSomething(val));

map is getting called for each observer.

Subjects are a special type of Observables that allow values to be multicast, that means you share a single line of execution of your Observable. This is a rxjs6, if you get lost with pipeable operators, please take a look here.

First of all, get the imports,

import { Observable, Subject, timer } from 'rxjs';
import { map, share } from 'rxjs/operators';

Then you have,

const subject = new Subject();

const doSomething = val => {
  console.log("doing something");
  return val;
}

const observable = timer(0, 1000).pipe(
  map(val => doSomething(val)),
).pipe(share());

const first = observable.subscribe(val => console.log("first:", val));
const second = observable.subscribe(val => console.log("second:", val));
const tercer = observable.subscribe(val => console.log("tercer:", val));

// After 1.5 seconds, stop first.
timer(1500).subscribe(_ => first.unsubscribe());
// After 2.5 seconds, stop second.
timer(2500).subscribe(_ => second.unsubscribe());
// After 2.5 seconds, stop second.
timer(2500).subscribe(_ => tercer.unsubscribe());