To do this, you can use a combination of switchMapand filter(provided that all actions, including start / end actions, come from the same source)
/ , , .
, .
const actions$ = new Rx.Subject();
const controllActions$ = actions$
.filter(action => action.type === "END" || action.type === "START");
const dataActions$ = actions$
.filter(action => action.type !== "END" && action.type !== "START");
const epic$ = controllActions$
.switchMap(action => {
if (action.type === "END") {
console.info("Pausing stream");
return Rx.Observable.never();
} else {
console.info("Starting/Resuming stream");
return dataActions$;
}
});
epic$.subscribe(console.log);
Rx.Observable.from([
"Some data, that will not be emitted...",
{type: "START"},
"Some data, that _will_ be emitted...",
"Some more data, that _will_ be emitted...",
{type: "END"},
"Some data, that will not be emitted...",
"Some data, that will not be emitted...",
{type: "START"},
"Some data, that _will_ be emitted...",
"Some more data, that _will_ be emitted..."
])
.concatMap(d => Rx.Observable.of(d).delay(400))
.subscribe(actions$);
<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>