Как я могу обеспечить извлечение в глубину, используя расширение rxjs?

Question

Я работаю над проектом Angular 7 / Typescript / RxJS 6.3.3, использующим различные наблюдаемые RxJS и связанные с ними операторы для обработки иерархических коллекций (особенно в виде дерева) объектов, извлеченных из базы данных через http-сервер.

Я думал использовать оператор expand для создания поиска в глубину и использовал concatMap для поддержания порядка ... или я так надеялся. Не работает.

См. Дистиллированный пример здесь: https://stackblitz.com/edit/rxjs-vf4zem

Вывод консоли из этого примера:

dfs no delay: [1,8,9,22,23,24,2,4,7,3]
dfs with delay: [1,2,3,4,7,8,9,22,23,24]

(Вторая строка вывода может отличаться в зависимости от того, сколько добавляется задержка. Задержка предназначена для имитации выборки данных с http-сервера.)

Учитывая данные в примере, я хочу последовательно получать первую строку вывода: упорядочение по глубине. Ключевая функция из примера:

const dfs = (getter: Getter) => rootIds.pipe(
  concatMap(ids => from(ids)),
  expand(id =>
    getter(id).pipe(
      concatMap(children => from(children))
    )
  ),
  toArray()
);

Есть ли способ обеспечить обработку в глубину? Может ли expand не гарантировать этого, или это просто плохое средство для получения иерархических данных в уплощенный массив в глубину?

Answer 1

Я думаю, это хороший вопрос, и я согласен, что для параллельной выборки вам понадобится дополнительная структура данных для составления результатов после выборки.

Тем не менее, было интересно реализовать рекурсивную реконструкцию через expand, так что вот моя последовательная попытка:

sequentialDFS(getChildren: Getter, ids: number[]): Observable<number[]> {
  return of(ids).pipe(
    expand(([head, ...rest]) =>
      // here we have a sequence of ids
      // that we'll explore in left-to-right order,
      // e.g. [1, 17, 20] will...
      getChildren(head).pipe(
        switchMap(subChildren => {
          // ...will turn here into [2, 6, 17, 20]
          const acc = [...subChildren, ...rest ];
          return acc.length
            ? of(acc)
            : EMPTY;
        })
      )
    ),

    // collect the heads {{{
    map(([head])=>head),
    toArray()
    // }}}
  );
}

* Я немного изменил метод getChildren, чтобы он возвращал Observable<number[]> вместо Observable<number>

https://stackblitz.com/edit/rxjs-rf8d1j

Это ни в коем случае не ответ на параллельную выборку. Просто делюсь этим, потому что это было весело.

Answer 2

Итак, я немного поиграл, выполнил некоторую ручную рекурсию (вместо того, чтобы полагаться на expand) и придумал следующее (также обновленное по ссылке на stackblitz выше):

class Test {
  // This doesn't maintain depth-first order; it can vary depending on timing.
  brokenDFS(getChildren: Getter, ids: number[]): Observable<number[]> {
    return of(ids).pipe(
      concatMap(ids => from(ids)),
      expand(id => getChildren(id)),
      toArray()
    );
  }

  workingDFS(getChildren: Getter, ids: number[]): Observable<number[]> {
    return from(ids).pipe(
      concatMap(id => this.parentAndChildren(getChildren, id)),
      toArray()
    );
  }

  private parentAndChildren(getChildren: Getter, id: number): Observable<number> {
    return of(id).pipe(
      concat(
        getChildren(id).pipe(
          map(child => this.parentAndChildren(getChildren, child)),
          concatAll()
        )
      ),
    );
  }

}

const getter = getChildrenWithDelay;
const rootIds = [1, 17, 20];
const test = new Test();
test.brokenDFS(getter, rootIds).subscribe(data => console.log(`Broken: ${data}`));
test.workingDFS(getter, rootIds).subscribe(data => console.log(`Working: ${data}`));

Вывод (с учетом того, что «сломанный» вывод может изменяться из-за синхронизации):

Broken: 1,17,20,21,2,6,18,7,13,14,3,4,19,15,16,5,8,9,10,11,12
Working: 1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21

(Также обратите внимание, что я изменил значения дерева из исходного поста, так что правильный DFS - это числовой массив от 1 до 21).

Так что workingDFS работает, но намного медленнее, чем brokenDFS, так как каждый запрос к http-серверу должен ждать, пока все завершится, в то время как версия brokenDFS будет запускать несколько запросов одновременно (хотя это не правильно порядок).

Я не знаю, есть ли у меня лучшие варианты rxjs здесь. Возможно, мне придется пересмотреть свои методы, чтобы передать не только интересующие объекты, но также некоторую информацию о структурировании / сортировке, выполнить все / многие запросы одновременно, а затем объединить все в правильном порядке.