takeUntil with a filter

user_1337347 0
Viewed 496

I have some code in NestJS with a mergeMap operator where I need to stop the observables stream based on some values from the outer and inner observables. Consider this example:

type SomeEvent1 {
  value: string;
}

type SomeEvent2 {
  value: string;
}

@Injectable()
export class TradeSagas {
  private readonly logger = new MyLogger(TradeSagas.name);
  @Saga()
  someSaga = (events$: Observable<IEvent>): Observable<ICommand> => {
    return events$.pipe(
      ofType(SomeEvent1),
      mergeMap((event: SomeEvent1) => {
        events$.pipe(
          ofType(SomeEvent2),
          map((event) => new SomeEvent3())
        );
      })
    );
  };

I need to be able to unsubscribe from the stream which listens to SomeEvent1 or the stream which listens to SomeEvent2(not sure yet what is needed for my particular use case) when for example SomeEvent1.value === SomeEvent2.value.

I know I could use takeUntil operator but it accepts another observable so I am confused how I can make it work.

The reason why I want to unsubscribe is because mergeMap keeps track of all previous event values and sometimes it is messing the flow. I know I could use switchMap but that is not the solution in my particular problem since I don't need to cancel the outer observable just because the new event arrived - I have a bit more complicated logic to build the stream of events. If there are other ideas on how to handle this problem, please, let me know.

Thanks for any help.

rxjs nestjs typescript
1 Answers

For your case, takeWhile would be a better choice, since it takes a function rather than an observable:

  someSaga = (events$: Observable<IEvent>): Observable<ICommand> => {
    return events$.pipe(
      ofType(SomeEvent1),
      mergeMap(event1 => events$.pipe(
          ofType(SomeEvent2),
          takeWhile(event2 => event2.value === event1.value),
          map(event2 => new SomeEvent3())
        );
      })
    );
  };

Edit:

Sometimes it's easier to follow if you break the sources up into separate variables. So I think this might be what you are looking for:

const event1$ = events$.pipe(ofType(SomeEvent1));
const event2$ = events$.pipe(ofType(SomeEvent2));
const event3$ = events$.pipe(ofType(SomeEvent3));

const someSaga2 = events1$.pipe(
  mergeMap(event1 => events2$.pipe(
    takeUntil(events3$.pipe(
        filter(event3 => event3.value === event1.value)
    ))
  ))
);

Description of behavior: When Event1 emits, emit Event2 until an Event3 is emitted that has the same value as Event1.

user_18583570
Related