RxJs декартово произведение сгруппированного горячих наблюдаемых

Question

Мотивация:

Цель состоит в том, чтобы написать выскабливание библиотеки, которая позволяет извлечение реляционных/нормализованных данных.

Для этого есть:

• исходных потоки: испускают Cheerio документы: либо путем сканирования URL-адресов или обзорного родитель Cheerio документы (например, здор`овы документы -> несколько Cheerio документов, по одному для каждого лития в родитель)
• таблица потоков: испускать строки таблицы, подписаться на комбинацию исходных потоков и извлекать данные из приведенных документов Cheerio

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

Пример: Блог

на простой блог там может быть сообщений, авторы и категории; Основная структура данных, например,

Post: {id, author, title, text} 
Author: {id, name} 
Category: {id, name} 
author_post: {author_id, post_id} 
post_category: {post_id, category_id} 

Чтобы воссоздать структуру данных с Царапины HTML, мы создаем три источника потока:

1. пост: кормил пост URLs, возвращает Cheerio документы сообщений
2. post.author: ребенок сообщений , следует за гиперссылкой на автора сообщений и испускает cheerio документы авторов
3. post.category: child to posts, возвращает документ cheerio для каждой категории, указанной в сообщении (например, в '.categories li')

Чтобы воссоздать таблицу post_category, каждое сообщение должно быть объединено с каждой категорией, принадлежащей этому сообщению (== carthesian product).

Моя актуальная проблема еще более надуманна, так как потоки детей уже выпускают собственный документ cheerio, а также каждый из их родителей, т. Е. {Post: cheerio, author: cheerio}.

Проблема объединения потоков только возникает, например. братья и сестры.

Я также не могу обойти проблему группировки испуская все ребенок родителя внутри одного потока (например, {пост, автор, категория}), как более сложные структуры данных требуют группировок по дедушке

(I может служить примером для этого, если это необходимо, но это становится достаточно длинным, как есть).

Проблема:

Это не представляется возможным использовать GroupBy и почтовый индекс, чтобы объединить группы горячих наблюдаемыми.

GroupObservables, излучаемые groupBy, генерируют свои значения, как только они создаются, и zip, ожидая, что все zipped наблюдаемые, чтобы испускать GroupObservables, пропускают любые значения, испущенные до запуска zip-функции.

Вопрос:

Как сделать I группы и пронестись горячие наблюдаемыми без потери значения? Информация о времени (например, все дети, излучающие до исхода следующего родителя) не может полагаться, поскольку дети, возможно, разрешены асинхронно (обход).

Дополнительная информация:

Лучшее, что я могу представить себе это так:

Child1 и Child2 отображаются версии родителя, C1C2 является результатом группировки ребенка1 и ребенка2 по родителю и вычисления декартово продукт внутри этих групп.

Parent: -1------2-------3-------- 

Child1: --a------b--c------------ 

Child2: ---1-2----3---4---------- 

C1C2: --a1-a2---b3-c3-b4-c4---- 

самаформируяил декартово произведение не проблема, так как есть простая реализация (из выпуска № 807 на RxJs, не может оставлять больше ссылок)

function xprod (o1, o2) { 
    return o1.concatMap(x => o2, (x, y) => [x, y]); 
}; 

Проблема не хватает какие-либо значений ,

Edit: jsbin показывает простую ситуацию: 1 родителя, 2 детей.

Ссылки:

• question on RxJs Github // удалены дополнительные ссылки, как я могу только создавать 2

Answer 1

Решение, которое я придумал наивная реализация оператора группировки возвращая ReplaySubjects.В то время как решение относится только к моим требованиям, общая суть может быть полезным:

Observable.prototype.splitBy = function(keySelector) { 
    const parentObservable = this; 
    let group, lastKey; 
    return Observable.create(observable => { 
    return parentObservable.subscribe(
     value => { 
     const currentKey = keySelector(value); 
     if(currentKey === lastKey) { 
      group.next(value); 
     } else { 
      if(group) group.complete(); 
      group = new ReplaySubject(); 
      observable.next(group); 
      group.key = currentKey; 
      group.next(value); 
     } 
     lastKey = currentKey; 
     }, 
     error => observable.error(error), 
     completed => { 
     group.complete(); 
     observable.complete(); 
     }); 
    }); 
}; 

Answer 2

добавить еще один ответ здесь на то, что мое новое понимание вашего вопроса. Я удалю его позже, если я неправильно понял.

Я понимаю,:

• есть родитель наблюдаемых значений, которые генерируют
• этих значений используются для генерации ребенка1 других наблюдаемым, которая представляет собой последовательность значений из асинхронной операции ($ .get, или что-то еще)
• такой же для ребенка2
• одно значение родителя, следовательно, генерирует два наблюдаемых, и вы хотите декартово произведение этих двух наблюдаемых.

Этот код должен делать:

var parent = Rx.Observable.interval(500) 
    .take(3) 
    .publish(); 
var Obs3Val = function (x, delay) { 
    return Rx.Observable.interval(delay) 
    .map(function(y){return "P"+x+":"+"C" + y;}) 
    .take(3); 
}; 
function emits(who){ 
    return function (x) {console.log(who + " emits " + x);}; 
} 

var child1 = parent.map(x => Obs3Val(x, 200)); 
var child2 = parent.map(x => Obs3Val(x, 100)); 

Rx.Observable.zip(child1, child2) 
      .concatMap(function(zipped){ 
    var o1 = zipped[0], o2 = zipped[1]; 
o1.subscribe(emits("child1")); 
o2.subscribe(emits("child2")); 
    return o1.concatMap(x => o2, (x, y) => [x, y]); 
}) 
    .subscribe(function(v) { 
       console.log('cartesian product : ' +v); 
       }); 

parent.subscribe(); 

parent.connect(); 

jsbin: https://jsbin.com/revurohoce/1/edit?js,console, https://jsbin.com/wegayacede/edit?js,console

Вход:

"child2 emits P0:C0" 
"child1 emits P0:C0" 
"child2 emits P0:C1" 
"cartesian product : P0:C0,P0:C0" 
"child2 emits P0:C2" 
"child1 emits P0:C1" 
"cartesian product : P0:C0,P0:C1" 
"cartesian product : P0:C0,P0:C2" 
"child2 emits P1:C0" 
"child1 emits P0:C2" 
"cartesian product : P0:C1,P0:C0" 
"child1 emits P1:C0" 
"child2 emits P1:C1" 
"cartesian product : P0:C1,P0:C1" 
"child2 emits P1:C2" 
"cartesian product : P0:C1,P0:C2" 
"child1 emits P1:C1" 
"cartesian product : P0:C2,P0:C0" 
"cartesian product : P0:C2,P0:C1" 
"child1 emits P1:C2" 
"child2 emits P2:C0" 
"cartesian product : P0:C2,P0:C2" 
"child1 emits P2:C0" 
"child2 emits P2:C1" 
"child2 emits P2:C2" 
"child1 emits P2:C1" 
"cartesian product : P1:C0,P1:C0" 
"cartesian product : P1:C0,P1:C1" 
"child1 emits P2:C2" 
"cartesian product : P1:C0,P1:C2" 
"cartesian product : P1:C1,P1:C0" 
"cartesian product : P1:C1,P1:C1" 
"cartesian product : P1:C1,P1:C2" 
"cartesian product : P1:C2,P1:C0" 
"cartesian product : P1:C2,P1:C1" 
"cartesian product : P1:C2,P1:C2" 
"cartesian product : P2:C0,P2:C0" 
"cartesian product : P2:C0,P2:C1" 
"cartesian product : P2:C0,P2:C2" 
"cartesian product : P2:C1,P2:C0" 
"cartesian product : P2:C1,P2:C1" 
"cartesian product : P2:C1,P2:C2" 
"cartesian product : P2:C2,P2:C0" 
"cartesian product : P2:C2,P2:C1" 
"cartesian product : P2:C2,P2:C2"