Игнорировать/отменить/прерывать потоки в функциональном реактивном программировании?

Question

Есть ли способ отменить/игнорировать/прерывать поток на основе вывода другого потока?

Прецедент является то, что у меня есть 3 потока:

1. loginStream - пользователь вводит войти кредитки и нажимает кнопку Отправить. Испускает логин.
2. authorizeStream-system пытается авторизировать пользователя. Выдает токен.
3. logoutStream - пользователи нажимает кнопки выхода

Общий случай --- пользователь входит в систему, система пытается разрешить, в какой-то момент в будущем пользователь выходит из системы. Это довольно стандартный FRP.

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

Кромка края --- пользователь регистрируется, но сразу же выходит из системы, прежде чем он авторизован.

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

Чтобы система не перешла в странное состояние, мне нужно игнорировать поток авторизации в данном конкретном случае. Единственное, о чем я могу думать, это (1) объединить три потока, (2) состояние трека через сканирование, (3) logout устанавливает состояние в logout, логин устанавливает состояние в login. Отфильтруйте событие авторизации, если не в 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')) 

выше подход должен работать, но мне интересно, если есть пылесос, «более FRP» путь игнорирования authorizationStream.

Answer 1

Этот поток излучает только после login, и только перед logout:

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

В зависимости от вашего дела, вы можете опустить .take(1) часть, чтобы позволить несколько авторизует для одного входа

Answer 2

takeUntil говорит: «Прекратите излучать результаты из этого потока, когда элемент испускается этим другим потоком», который, как я думаю, вам нужен.