subscribe — RxJS Observable method example + marble diagram

Calling subscribe tells inner emitter function (producer) that the observer (consumer) is ready to receive emissions:

Also take a look at rxjs/constructor to better understand the producer-consumer connection

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
const { rxObserver } = require('api/v0.3');
const { interval } = require('rxjs');


// source stream -- an interval
// emitting at 50ms, 100ms, 150ms, 200ms, 250ms, etc
const source$ = interval(50);

// observer to draw marbles on the diagram
const marbleObserver = rxObserver();

// unsubscribe from source$ in 200ms
const subscription = source$
  .subscribe(v => {
    console.log(v);
    marbleObserver.next(v); // draw a marble
  });

// unsubscribe from source$ in 220ms
setTimeout(() => {
  subscription.unsubscribe();
  marbleObserver.complete(); // complete diagram
}, 220);