如何行走的树结构

Question

当我试图重写使用RxJS的软件包管理器避免`shareReplay`（PM被pnpm和PR是here）。

在重写，我用了很多.shareReplay(Infinity)，这是我一直在说是坏的（我在反应式编程初学者）

有人能提出一个替代方案如何改写如下代码，W/o使用.shareReplay(Infinity)：

'use strict' 
const Rx = require('@reactivex/rxjs') 

const nodes = [ 
    {id: 'a', children$: Rx.Observable.empty()}, 
    {id: 'b', children$: Rx.Observable.empty()}, 
    {id: 'c', children$: Rx.Observable.from(['a', 'b', 'd'])}, 
    {id: 'd', children$: Rx.Observable.empty()}, 
    {id: 'e', children$: Rx.Observable.empty()}, 
] 

// I want this stream to be executed once, that is why the .shareReplay 
const node$ = Rx.Observable.from(nodes).shareReplay(Infinity) 

const children$ = node$.mergeMap(node => node.children$.mergeMap(childId => node$.single(node => node.id === childId))) 

children$.subscribe(v => console.log(v)) 

Answer 1

的 groupBy经营者应在这里工作。纵观PR这可能是一个毛过于简单化，但这里有云：

'use strict' 
const Rx = require('@reactivex/rxjs') 

const nodes = [ 
    {id: 'a', children$: Rx.Observable.empty()}, 
    {id: 'b', children$: Rx.Observable.empty()}, 
    {id: 'c', children$: Rx.Observable.from(['a', 'b', 'd'])}, 
    {id: 'd', children$: Rx.Observable.empty()}, 
    {id: 'e', children$: Rx.Observable.empty()}, 
] 

Rx.Observable.from(nodes) 
// Group each of the nodes by its id 
.groupBy(node => node.id) 
// Flatten out each of the children by only forwarding children with the same id 
.flatMap(group$ => group$.single(childId => group$.key === childId)) 
.subscribe(v => console.log(v)); 

编辑：比我想象

好比较困难，所以我的第二个通读我发现这需要比我最初想象的更多的工作，所以它不能简化得如此简单。基本上，你将不得不在内存复杂度和时间复杂度之间进行选择，因为没有一个灵丹妙药。

从优化的角度来看，如果初始源只是一个数组，那么你可以删除shareReplay，它将以完全相同的方式工作，因为当订阅ArrayObservable时，唯一的开销是迭代通过Array ，重新运行源代码没有任何额外的成本。

基本上对于这个我觉得你可以想到两个维度，节点数量m和子女平均人数n。在速度优化的版本，你最终将不得不通过m运行两次，你将需要通过“N”的节点进行迭代。既然你有m * n个孩子，最糟糕的情况就是他们都是独一无二的。这意味着你需要做的(m + m*n)操作，如果我没有记错，这简化到O(m*n)。这种方法的缺点是，您需要同时具有（nodeId - > Node）的Map和用于删除重复依赖关系的Map。

'use strict' 
const Rx = require('@reactivex/rxjs') 

const nodes = [ 
    {id: 'a', children$: Rx.Observable.empty()}, 
    {id: 'b', children$: Rx.Observable.empty()}, 
    {id: 'c', children$: Rx.Observable.from(['a', 'b', 'd'])}, 
    {id: 'd', children$: Rx.Observable.empty()}, 
    {id: 'e', children$: Rx.Observable.empty()}, 
] 

const node$ = Rx.Observable.from(nodes); 

// Convert Nodes into a Map for faster lookup later 
// Note this will increase your memory pressure. 
const nodeMap$ = node$ 
    .reduce((map, node) => { 
    map[node.id] = node; 
    return map; 
    }); 


node$ 
    // Flatten the children 
    .flatMap(node => node.children$) 
    // Emit only distinct children (you can remove this to relieve memory pressure 
    // But you will still need to perform de-duping at some point. 
    .distinct() 
    // For each child find the associated node 
    .withLatestFrom(nodeMap$, (childId, nodeMap) => nodeMap[childId]) 
    // Remove empty nodes, this could also be a throw if that is an error 
    .filter(node => !!node) 
    .subscribe(v => console.log(v)); 

另一种方法是使用类似于其重点存储器减压以性能为代价你的方法。注意，就像我说的，如果你的源是一个数组，那么你基本上可以删除shareReplay，因为当它重新评估时，它正在做的是重新迭代数组。这消除了额外的Map的开销。虽然我认为你仍然需要独特的删除重复。这样做的最坏情况下的运行复杂性将是O(m^2*n)或者干脆O(m^2)如果n很小，因为你需要通过所有的儿童和为每个孩子还需要通过m迭代再次找到匹配的节点进行迭代。

const node$ = Rx.Observable.from(nodes); 
node$ 
    // Flatten the children 
    .flatMap(node => node.children$) 
    // You may still need a distinct to do the de-duping 
    .flatMap(childId => node$.single(n => n.id === childId))); 

我会说第一个选项在几乎所有情况下都是可取的，但我把它留给你来确定你的用例。这可能是因为你在某些情况下建立了一种启发式算法，可以选择另一种算法。

旁注：对不起，这是不容易的，但爱PNPM所以保持良好的工作！