Ignore/cancel/interrupt streams in Functional reactive programming?

Question

Is there a way to cancel/ignore/interrupt a stream based on the output of another stream?

The use case is that I have 3 streams:

1. loginStream - user enters login creds and clicks submit. Emits login creds.
2. authorizeStream- system tries to authorize user. Emits a token.
3. logoutStream - users presses the logout button

The common case is---the user logs in, the system tries to authorize, at some point in the future the user logs out. This is pretty standard FRP.

Common case
    loginStream: -------------------o---->
authorizeStream: ---------------a-------->
   logoutStream: ---------l-------------->

The edge case is---user logs in but immediately logs out before he's authorized.

Edge case
    loginStream: -------------------o---->
authorizeStream: --------a---------------> //ignore/cancel/interrupt this event
   logoutStream: -------------l---------->

To keep the system from going to a weird state, I need to ignore the authorize stream in this particular case. The only thing I can think of is to (1) merge the three streams, (2) track state via a scan, (3) logout sets the state to logout, login sets the state to login. Filter out an authorize event when not in login.

 merge([loginStream, authorizeStream, logoutStream])
   .scan((state, event) => event.type !== 'auth' ? {state: event.type, event} : {state, event}, {})
   .filter(event => !(event.state !== 'login' && event.event.type === 'auth'))

The above approach should work but I'm wondering if there's a cleaner, "more FRP" way of ignoring the authorizationStream.

Answer 1

This stream emits only after login, and only before logout:

loginStream.flatMapLatest(() => {
    return authorizeStream.take(1).takeUntilBy(logoutStream);
})

Depending on your case, you can omit .take(1) part to allow several authorizes for one login

Answer 2

takeUntil says "stop emitting results from this stream when an element is emitted by this other stream", which I think is what you want here.