CODEWITHSUNDEEP
rxjsrxjs6
customcommander
customcommander

Reputation: 18921

In RxJS, how to prevent operators from emitting values if the source observable hasn't emitted any?

I have three observables foo$, bar$ and baz$ that I merge together to form another observable. This is working as expected:

const foo$ = of('foo');
const bar$ = of('bar');
const baz$ = of('baz');

merge(foo$, bar$, baz$).pipe(startWith('>>>'), endWith('<<<')).subscribe(str => {
  console.log(str);
});
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.min.js"></script>
<script>const {merge, of} = rxjs; const {startWith, endWith} = rxjs.operators;</script>

Now if none of the three observables above emit a value, I do not want to output neither >>> nor <<<. So startWith and endWith can only "run" if merge(foo$, bar$, baz$) actually emits a value.

In order to simulate that I'm rejecting all values emitted by foo$, bar$ and baz$ with a filtering function.

const foo$ = of('foo').pipe(filter(() => false));
const bar$ = of('bar').pipe(filter(() => false));
const baz$ = of('baz').pipe(filter(() => false));

merge(foo$, bar$, baz$).pipe(startWith('>>>'), endWith('<<<')).subscribe(str => {
  console.log(str);
});
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.min.js"></script>
<script>const {merge, of} = rxjs; const {startWith, endWith, filter} = rxjs.operators</script>

However as you can see in the output, both startWith and endWith have emitted their value even though the merge() hasn't produced any.

Question: How can I prevent startWith and endWith from executing if the observable did not emit a single value?

Upvotes: 4

Views: 5691

Answers (5)

Lloren&#231; Pujol Ferriol
Lloren&#231; Pujol Ferriol

Reputation: 2638

I see that all answers are done combining operators, but doing this solution with only operators is kind of tricky, why not just create your own observable? I think this solution is the easiest one to understand. No hidden cleverness, no complex operator combinations, no subscription repetitions...

Solution:

const foo$ = of('foo')//.pipe(filter(() => false));
const bar$ = of('bar')//.pipe(filter(() => false));
const baz$ = of('baz')//.pipe(filter(() => false));

function wrapIfOnEmit$(...obs) {
  return new Observable(observer => {
    let hasEmitted;
    const subscription = merge(...obs).subscribe((data) => {
      if (!hasEmitted) {
        observer.next('>>>');
        hasEmitted = true;
      }
      observer.next(data);
    },
    (error) => observer.error(error),
    () => {
      if (hasEmitted) observer.next('<<<');
      observer.complete();
    })
    return () => subscription.unsubscribe();
  })
}

wrapIfOnEmit$(foo$, bar$, baz$).subscribe(console.log);
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.min.js"></script>
<script>const {merge, of, Observable} = rxjs; const {filter} = rxjs.operators</script>

Hope this helps!

Upvotes: 2

frido
frido

Reputation: 14129

You could assign the merged observable to a variable, take the first element from the stream and then map to the observable you want to execute when atleast one value has been emitted.

const foo$ = of('foo').pipe(filter(() => false))
const bar$ = of('bar').pipe(filter(() => false))
const baz$ = of('baz').pipe(filter(() => false))

const merged$ = merge(foo$, bar$, baz$);
merged$.pipe(
  take(1),
  switchMap(() => merged$.pipe(
    startWith('>>>'),
    endWith('<<<')
  ))
).subscribe(console.log);

https://stackblitz.com/edit/rxjs-phspsc

Upvotes: 3

martin
martin

Reputation: 96949

The first condition is simple. You can just prepend the first emission with concatMap:

mergeMap((v, index) => index === 0 ? of('>>>', v) : of(v))

The second condition is more tricky. You want basically the right opposite to defaultIfEmpty. I can't think of any simple solution so I'd probably use endWith anyway and just ignore the emission if it's the first and only emission (which means the source just completed without emitting anything):

endWith('<<<'),
filter((v, index) => index !== 0 || v !== '<<<'),

Complete example:

const foo$ = of('foo');//.pipe(filter(() => false));
const bar$ = of('bar');//).pipe(filter(() => false));
const baz$ = of('baz');//.pipe(filter(() => false));

merge(foo$, bar$, baz$).pipe(
  mergeMap((v, index) => index === 0 ? of('>>>', v) : of(v)),
  endWith('<<<'),
  filter((v, index) => index !== 0 || v !== '<<<'),
).subscribe(console.log);

Live demo: https://stackblitz.com/edit/rxjs-n2alkc?file=index.ts

Upvotes: 3

kos
kos

Reputation: 5364

One way to achieve that is to use a materialize-dematerialize operators pair:

source$.pipe(
  // turn all events on stream into Notifications
  materialize(),

  // wrap elements only if they are present
  switchMap((event, index) => {
    // if its first event and its a value
    if (index === 0 && event.kind === 'N') {
      const startingNotification = new Notification('N', '>>>', undefined);
      return of(startingNotification, event);
    }

    // if its a completion event and it not a first event
    if (index > 0 && event.kind === 'C') {
      const endingNotification = new Notification('N', '<<<', undefined);
      return of(endingNotification, event);
    }

    return of(event);
  }),

  // turn Notifications back to events on stream
  dematerialize()
)

Wrap source emission only if its not empty

Play with this code in a playground:
https://thinkrx.io/gist/5a7a7f1338737e452ff6a1937b5fe05a
for convenience, I've added an empty and error source there as well

Hope this helps

Upvotes: 0

Code Spirit
Code Spirit

Reputation: 5061

From the documentation of startWith:

Returns an Observable that emits the items you specify as arguments before it begins to emit items emitted by the source Observable.

So startWith and endWith will always run.

I dont know what your expected result should be, but if you only want to concatenate the strings for each emitted value you could use the map or switchMap operators.

EDIT: Example with map to concat each value:

const foo$ = of('foo').pipe(filter(() => false));
const bar$ = of('bar').pipe(filter(() => false));
const baz$ = of('baz').pipe(filter(() => false));

merge(foo$, bar$, baz$).pipe(map(v => `>>>${v}<<<`)).subscribe(str => {
  console.log(str);
});
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.min.js"></script>
<script>const {merge, of} = rxjs; const {startWith, endWith, filter, map} = rxjs.operators</script>

Upvotes: 1

Related Questions