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

Question

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

Случай использования состоит в том, что у меня есть 3 потока:

1. loginStream - пользователь вводит данные для входа в систему и нажимает кнопку "Отправить". Выдает кредиты для входа.
2. authorizeStream - система пытается авторизовать пользователя. Издает токен.
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, логин устанавливает состояние 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 говорит "прекратить излучать результаты из этого потока, когда элемент излучается этим другим потоком", что, я думаю, именно то, что вы хотите здесь.