CODEWITHSUNDEEP
javascriptrxjsobservable
zaggi
zaggi

Reputation: 940

RxJs operator with unconditional execution in the end of each iteration

Is there a Rx operator or composition to guarantee some logic execution last per each observable emission?

Let's assume the following context:

  1. endless or continuous sequence
  2. conditional logic like filter() to skip some emissions
  3. some logic in the end of each iteration in a doAlways()-like operator

Please refer to numbered comments in the code-sample below

Notes:

Code snippet to illustrate: Step (3) should execute always, per-iteration, last, i.e. we want always start and finish log in a doAlways()-like operator instead of tap()

import { of, interval } from 'rxjs'; 
import { tap, filter, switchMap } from 'rxjs/operators';


const dataService = { update: (z) => of(z /*posts data to back-end*/) };

const sub = interval(1000).pipe(            // <-- (1)
  tap(a => console.log('starting', a)),
  filter(b => b % 100 === 0),               // <-- (2)
  switchMap(c => dataService.update(c)),
  tap(d => console.log('finishing', d))     // <-- (3) should execute always even (2)
)
.subscribe(x => console.log(x));

Upvotes: 1

Views: 421

Answers (4)

bvdb
bvdb

Reputation: 24760

A possible work-around is to eliminate the .filter() of course.

Given:

observable.pipe(
  filter(() => condition),
  switchMap(p => service.otherObservable(p)),
  tap(d => console.log('not triggered when condition is false'))
)

Can be rewritten as:

observable.pipe(
  switchMap(p =>
   (!condition) ? 
     of(null) : 
     service.otherObservable(p)),
  tap(d => console.log('always triggered'))
)

Upvotes: 0

Maksim Romanenko
Maksim Romanenko

Reputation: 365

I would split it for 2 streams with partition, and them would merge them back.

https://stackblitz.com/edit/rxjs-8z9jit?file=index.ts

import { of, interval, merge } from 'rxjs';
import { tap, switchMap, partition, share } from 'rxjs/operators';

const dataService = { update: z => of(z /*posts data to back-end*/) };

const [sub, rest] = interval(1000).pipe(
  tap(a => console.log('starting', a)),
  share(),
  partition(b => b % 2 === 0)
);

merge(
    sub.pipe(
      switchMap(c => dataService.update(c)),
      tap(() => console.log('Ok'))
    ), 
    rest
)
.pipe(tap(d => console.log('finishing', d)))
.subscribe(x => console.log(x));

Upvotes: 1

ggradnig
ggradnig

Reputation: 14189

Nope, what you ask is not possible due to the fact that filtered notifications won't ever be passed on. You would need a totally new type of notification that is not consumed by any other operator than the one you describe.

Now, here is an idea that is probably not recommended. You can misuse the error notification to skip some operators, but that will interfere with any other error handling so that's not something you should do... 

const sub = interval(1000).pipe(            
  tap(a => console.log('starting', a)),
  mergeMap(b => b % 100 === 0 ? of(b) : throwError(b)),
  switchMap(c => dataService.update(c)),
  catchError(b => of(b)),
  tap(d => console.log('finishing', d))
)

Note that we don't use filter but map to either a next or an error notification depending on the condition. Errors will naturally be ignored by most operators and consumed by tap, subscribe or catchError. This is a way to put a tag on the item so the workers know they shouldn't touch it (in the analogy described by Ingo)

Upvotes: 1

Ingo B&#252;rk
Ingo B&#252;rk

Reputation: 20043

No such operator exists, and that's because it can't exist. Once you filtered out a value, the resulting observable just doesn't emit it anymore. Any operator "downstream" just simply doesnt know about its existence.

To illustrate, you included a switchMap to some service, which depends on the emitted value. For obvious reasons that operator that cannot logically be applied if there is no value to switch on.

You would have to "tag" each value instead of filtering it and defer the filter to after the tap call, but even then scenarios like switching to another observable would require more detailed requirements.

Think of the observable as a conveyor belt on which you place items. Each operator is a room through which the belt leads. Inside each room a worker can decide what to do with each item: modify it, take it away, put new items in instead etc. However, each worker only sees what the conveyor belt brings along — they don't know what other rooms came before them or what has been done there.

To achieve what you want, the worker in the last room would have to know about an item that he never received, which would require additional knowledge they don't have.

Upvotes: 1

Related Questions