How to reset a RXJS scan operator based on another Observable

I think you could achieve what you want just by restructuring your chain (I'm omitting tap calls that trigger loading for simplicity):

search$.pipe(
  switchMap(searchterm =>
    concat(
      userService.getUsers(0, searchterm),
      offset$.pipe(concatMap(offset => userService.getUsers(offset, searchterm)))),
    ).pipe(
      map(({ data }) => data.map((user) => ({
        label: user.name,
        value: user.id
      }))),
      scan((acc, curr) => [...acc, ...curr], []),
    ),
  ),
);

Every emission from search$ will create a new inner Observable with its own scan that will start with an empty accumulator.


To manipulate the state of a scan you can write higher order functions that get the old state and the new update. Combine then with the merge operator. This way you stick to a clean stream-oriented solution without any side-effects.

const { Subject, merge } = rxjs;
const { scan, map } = rxjs.operators;

add$ = new Subject();
clear$ = new Subject();

add = (value) => (state) => [...state, value];
clear = () => (state) => [];

const result$ = merge(
  add$.pipe(map(add)),
  clear$.pipe(map(clear))
).pipe(
  scan((state, innerFn) => innerFn(state), [])
)

result$.subscribe(result => console.log(...result))

add$.next(1)
add$.next(2)
clear$.next()
add$.next(3)
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.3/rxjs.umd.min.js"></script>

This method can easily be extended and/or adapted for other state usecases in rxjs.

Example (remove last item)

removeLast$ = new Subject()

removeLast = () => (state) => state.slice(0, -1);

merge(
  ..
  removeLast$.pipe(map(removeLast)),
  ..
)