Reset nested Observable

Question

I'm trying to make an Observable that, on a button click, emits an array of PAGE_SIZE (the first click should emit [0, 1, 2, 3, 4]), with the caveat that after the button click an interval will begin emitting more numbers that are concatenated to the original (i.e. after the user clicks the interval should emit [0, 1, 2, 3, 4, 5, 6, 7, 8, 9], then [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14], etc.).

The following almost does what I want, but I need the whole process to start over when the user clicks the button again.

Any ideas?

const PAGE_SIZE = 5;
let currentPage = 0;
const buttonEl = document.getElementsByTagName('button')[0];
const refreshSource$ = new rxjs.Subject().pipe(rxjs.operators.concatMap(() => rxjs.interval(5000)));
const clickSource$ = new rxjs.fromEvent(buttonEl, 'click').pipe(rxjs.operators.tap(() => {
  refreshSource$.next();
  refreshSource$.complete();
}));
const clips$ = rxjs.merge(clickSource$, refreshSource$).pipe(rxjs.operators.mergeMap(value => {
  if (value.type === undefined)
    return makeObservable(value + 1);
  else
    return makeObservable(0);
}), rxjs.operators.scan((acc, value) => acc.concat(value)), rxjs.operators.startWith([]));
clips$.subscribe(value => {
  console.log('clips$', value);
});
function makeObservable(page) {
  return rxjs.of(d3.range(page * PAGE_SIZE, (page + 1) * PAGE_SIZE));
}

Can you elaborate on how clickSource$ and refreshSource$ are intended to interact? Also doesn't look like currentPage is used at present. — backtick
Specifically, it looks like the call to complete() in the tap operator might be canceling your interval before it starts. — backtick
yea, sure. I had copied this from jsfiddle.net/L3dy7zec/8 , loosely clickSource$ should emit each time the button is pressed, and refreshSource$ should emit only after clickSource$ has emitted.. the results should be concatenated together until clickSource$ emits again (i.e. the user clicks the button and restarts the interval) — lwiseman
about the complete().. if I console.log the result of the mergeMap it, seemingly, correctly shows a clickevent and then subsequent intervals, starting from 0 — lwiseman
the currentPage was before I realized I could just use the interval value itself.. it should be 0 on button click and increase by 1 on each interval tick — lwiseman

Mark van Straten Mark van Straten · Accepted Answer · 2019-09-24T10:25:20

What you need is to switchmap on every click to a new observable re-starting the generation of your array:

import { interval, fromEvent } from 'rxjs';
import { switchMap, take, map } from 'rxjs/operators';

const buttonEl = document.getElementById('button');
const pagesize = 5;

fromEvent(buttonEl, 'click')
  .pipe(
    // restart counter on every click
    switchMap(() => 
      interval(500).pipe(take(5)) 
    ),
    map(i => {
      const max = (i+1) * pagesize;
      return Array.from(Array(max).keys());
    })
  ) 
  .subscribe(console.log);

(stackblitz: https://stackblitz.com/edit/rxjs-v85ctu?file=index.html)

For full reference of the switchMap operator see: https://www.learnrxjs.io/operators/transformation/switchmap.html

Reset nested Observable

1 Answers