Реактивные расширения: Фильтруйте синхронизированные наблюдаемые

Question

У меня есть несколько потоков (наблюдаемые RxCpp), которые основаны на одном входном потоке. Каждый поток создается путем применения операций фильтрации и сопоставления к этому входному потоку.

Мраморная диаграмма примера:

Input: --I0--I1--I2--I3--|->
AS:    --A0------A1------|->
BS:    ------B0--B1------|->

Первый поток является входным потоком. I0 сопоставлен с A0 но отфильтрованы в потоке BS, I1 сопоставлен с B0 но отфильтрованы в потоке AS, I2 сопоставлен с A1 а также B1, I3 отфильтровывается в обоих потоках.

Я хотел бы строить потоки AS' а также BS':

• AS' должен излучать элементы AS если элемент не испускается BS в то же время. Это случай для элемента A0 в примере. Если элемент испускается на AS а также BS в то же время функция предиката должна решить, должен ли элемент быть выдан. Это случай для элемента A1 в примере, где B1 испускается в то же время.

• Короче: BS' похож на AS' но должен выделять элементы BS, Более подробный: BS' должен излучать элементы BS если элемент не испускается AS в то же время. Это случай для элемента B0 в примере. Если элемент испускается на AS а также BS в то же время функция предиката должна решить, должен ли элемент быть выдан. Это случай для элемента B1 в примере, где A1 испускается в то же время.

Мраморная диаграмма примера с конкретными значениями может выглядеть так, если AS' использует функцию предиката (a, b) := a <= b а также BS' использует функцию предиката (b, a) := b < a:

AS:  --0-----2--5--20--13--|->
BS:  -----7--3--4------13--|->
AS': --0-----2-----20--13--|->
BS': -----7-----4----------|->

Как я могу строить потоки AS' а также BS'?

Примечание. В конце концов мне нужно решение RxCpp (C++), но я предполагаю, что решения на других языках, которые имеют реактивные расширения, могут мне тоже помочь.