вторая объединенная ветвь игнорируется в XStreams

Проблема возникает из-за того, что storage.session.getItem отправляет как только подписывается, а merge(a$, b$) подписывается на a$ раньше, чем подписывается на b$, поэтому a$ получает событие, но к тому времени, когда b$ подписывается, оно приходит слишком поздно. Эта проблема хорошо известна и называется сбоем (в реактивном программировании) и обычно возникает при наличии потокового графа в форме ромба, как раз в вашем случае.

У меня есть два сообщения в блоге о сбоях, которые могут дать вам больше контекста: Primer on RxJS планировщики и сбои Rx на самом деле не проблема. В нем упоминается RxJS, но xstream довольно близок к RxJS с точки зрения реализации. Разница между xstream и RxJS заключается в том, что потоки xstream всегда многоадресные («общие»), а RxJS имеет много типов планировщика, а xstream — только один. Планировщик по умолчанию в RxJS ведет себя так же, как и в xstream.

Решение состоит в том, чтобы применить .remember() до достижения алмаза. Это связано с тем, что испускаемое значение необходимо кэшировать для других потребителей этого потока. .remember() просто преобразует Stream в MemoryStream. Я думаю, что исходный поток изначально был MemoryStream, и сопоставление MemoryStream создает другие MemoryStream, но filter — это оператор, который нарушает это. filter всегда возвращает поток, и причина этого в том, что MemoryStreams всегда должно иметь текущее значение, но filter может удалять значения, поэтому возможно, что отфильтрованный поток не иметь любую текущую стоимость.

Как автор Cycle.js, я считаю, что способ разработки cyclejs/storage не самый лучший, и я думаю, что мы найдем способы разработки этих API, чтобы свести к минимуму путаницу с MemoryStream и Stream. Но на данный момент важно понимать разницу между ними и планировать свое приложение так, чтобы избежать ромбов (и сбоев) или использовать .remember() в правильных местах.