Круговые зависимости с наблюдаемыми

Question

У меня есть несколько потоков, которые имеют циклические зависимости.

import input from './input'

const streamA = input.thru( someUsageOf_streamC );
const streamB = input.thru( someOtherUsageOf_streamC );
const streamC = most.merge(
                    streamA.constant('A'),
                    streamB.constant('B'),
                );

Функции переданы .thru(...) взять поток и применить некоторую комбинацию операторов, таких как .filter(), .map(), .until() но, в частности (см. ниже, почему), .delay(),

Это то, что я уже пробовал:

Способ 1

const someUsageOf_StreamC = streamX =>
    streamX.sample( (x,c) => [x,c], streamX, streamC )
           .filter( ([x,c]) => c === 'something' )
           .map( ([x,c]) => x );

Проблема здесь заключается в том, что вы должны передать потоки для выборки заранее, но streamC не может быть объявлено раньше streamA а также streamB,

Способ 2 (или 1.1)

import { subject } from 'most-subject'

const streamC = subject();

const streamA = ...;
const streamB = ...;
const streamC_proxy = most.merge( /* ...as above... */ );

streamC_proxy.subscribe( streamC );

Этот метод основан на методе 1, но просто передает события из streamC_proxy в streamC который может быть объявлен заранее с использованием наиболее предмета.

Я думаю, что это на самом деле работает, но это не тестируется с использованием большинства тестов (это требует использования своего собственного планировщика, но .subscribe()/.observe() неявно использовать defaultScheduler). Как я уже сказал, я использую .delay() и ждать секунд / минут для запуска тестов - с использованием планировщика по умолчанию - не представляется возможным.

Способ 3

import hold from '@most/hold'

const someUsageOf_StreamC = streamX =>
    streamX.flatMap( x => streamC.take(1) // @most/hold
                                 .filter( c => c === 'something' )
                                 .constant(x) )
           .multicast();

const streamA = ...;
const streamB = ...;
const streamC = most.merge( ... ).thru( hold );

Помимо того, что этот код далек от идеала, он должен работать. Но это не так:(

Вопрос

Можно ли это сделать тестируемым способом?