CODEWITHSUNDEEP
javascriptangulartypescriptrxjs
Jacob
Jacob

Reputation: 155

How do I use values emitted by 2 observables as parameters to an http request, with 1 observable resetting the other to a default value?

I am subscribing to a mapped observable that is based on 2 other observables. Whenever either of the 2 emits a value, I want the mapped observable to emit.

Whenever the outer observable emits, the inner observable value should reset to the default value (0 in the example). The problem is that when I reset the inner value, an extra value is emitted.

(For context - the 2 observables provide parameters for an http request, which is the mapped observable. A new outer value should reset the inner value to 0)

const outer = new Subject();
const inner = new BehaviorSubject(0);
const http = of("");
let mapped = outer.pipe(
  tap(() => inner.next(0)),
  switchMap(outerVal => {
    return inner.pipe(map(innerVal => ({ innerVal, outerVal })));
  }),
  switchMap(({ innerVal, outerVal }) => {
    return http.pipe(tap(() => console.log("http: ", outerVal, innerVal)));
  })
);

mapped.subscribe(result => {
  console.log(result);
});

outer.next("first");
inner.next(1);
inner.next(2);
outer.next("second");
inner.next(3);


// http: first 0
// http: first 1
// http: first 2
// http: first 0 <--- extra
// http: second 0
// http: second 3

I know why it's happening but I'm not sure how to deal with it. I have a solution that gets the correct results at the bottom of this stackblitz, but I think it's causing a memory leak.

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

Upvotes: 0

Views: 268

Answers (2)

Owen Kelvin
Owen Kelvin

Reputation: 15083

Consider below approach where I combine the two Observable stream and filter to only emit 1 value. We can then use the new stream to merge with the http request


const http = of("");
const triggerSubject$ = new BehaviorSubject({ inner: 0, outer: null });
const trigger = triggerSubject$.pipe(
  filter(({ inner, outer }) => inner && outer)
);
const triggerInner = val => {
  console.log("Inner", val);
  return triggerSubject$.next({
    inner: val,
    outer: triggerSubject$.value.outer
  });
};
const triggerOuter = val => {
  console.log("Outer", val);
  return triggerSubject$.next({ inner: 0, outer: val });
};

const mapped = trigger.pipe(
  switchMap(({ inner: innerVal, outer: outerVal }) =>
    http.pipe(tap(() => console.log("http: ", outerVal, innerVal)))
  )
);

mapped.subscribe(result => {
  console.log(result);
});

triggerOuter("first");
triggerInner(1);
triggerInner(2);
triggerOuter("second");
triggerInner(3);

See sample stackblitz demo

Upvotes: 0

Mrk Sef
Mrk Sef

Reputation: 8052

The following has the output that you want. I'm not sure it meets your criteria. Was there a reason your inner observable has to be a behavior subject? If it's just to give it an initial value, then the startWith operator does a much cleaner job of that.

const outer = new Subject();
const inner = new Subject();
const http = of("");

const mapped = outer.pipe(
  switchMap(outerVal =>
    inner.pipe(
      startWith(0),
      map(innerVal => ({ innerVal, outerVal }))
    )
  ),
  switchMap(({ innerVal, outerVal }) => 
    http.pipe(
      tap(() => console.log("http: ", outerVal, innerVal))
    )
  )
);

mapped.subscribe(console.log); // ""

Your stackblitz memory leak

You're creating long-lived multicasted streams and never completing them. You could try:

tap(() => {
  inner2.complete();
  inner2 = new BehaviorSubject(0);
}),

SwitchMap will unsubscribe, but multicasted streams don't complete at unsubscribe (depending on how its refcount works).

Upvotes: 1

Related Questions