分支和(重新)在RXJS中连接值对

Question

我想创建一个

的流

1. 在单独的流上分配值并处理每个部分
2. 每个流将转换数据，我无法控制应用转换
3. (重新 - (加入部分值及其相应的计数器部分

我要这样做的原因是要确保值的完整性。或至少在其中的某些部分。

因为每个流可以在加入流时都不会出现一些异步操作。使用某种concat()也无效，因为它会阻止所有传入值。处理应并行进行。

说明：

                            o
                            |
                            | [{a1,b1}, {a2,b2}, ...]
                            |
                            +
                           / 
                   {a<x>} /    {b<x>}
                         /     
                        |       |
                        |       + async(b<x>) -> b'<x>
                        |       |
                               /
                              /
                             /
                            /
                            + join(a<x>, b'<x>)
                            |
                            | [{a1,b'1}, {a2,b'2}, ...]
                            |
                       (subscribe)

我知道这可以通过result selector函数完成。例如

input$.mergeMap(
  ({a, b}) => Rx.Observable.of(b).let(async), 
  ({a}, processedB) => ({a, b:processedB})
);

但是，(a(这将导致async始终为每个值设置/拆卸。我希望部分流只能初始化一次。另外，(b(这仅适用于一个异步流。

我也尝试使用window*，但无法弄清楚如何再次重新加入值。还尝试使用goupBy没有运气。

---

编辑：

这是我目前的尝试。它具有上述问题(a(。Init...和Completed...只能记录一次。

const doSomethignAsync = data$ => {
  console.log('Init...') // Should happen once.
  return data$
    .mergeMap(val => Rx.Observable.of(val.data).delay(val.delay))
    .finally(() => console.log('Completed...')); // Should never happen
};
const input$ = new Rx.Subject();
const out$ = input$
  .mergeMap(
    ({ a, b }) => Rx.Observable.of(b).let(doSomethignAsync),
    ({ a }, asyncResult ) => ({ a, b:asyncResult })
  )
  
  .subscribe(({a, b}) => {
    if (a === b) { 
      console.log(`Re-joined [${a}, ${b}] correctly.`);
    } else {
      console.log(`Joined [${a}, ${b}]...`); // Should never happen
    }
  });
input$.next({ a: 1, b: { data: 1, delay: 2000 } });
input$.next({ a: 2, b: { data: 2, delay: 1000 } });
input$.next({ a: 3, b: { data: 3, delay: 3000 } });
input$.next({ a: 4, b: { data: 4, delay: 0 } });

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

Answer 1

这是一个相当复杂的问题，确切的工作方式将取决于未包括的用例非常具体的细节。

也就是说，这是一种可能的假设的一种可能的方法。它是有点通用的，例如与let一起使用的自定义操作员。

(旁注：我将其命名为"整理"，但这是一个不好且高度误导的名称，但是没有时间来命名事物...(

const collate = (...segments) => source$ =>
  source$
    .mergeMap((obj, index) => {
      return segments.map(({ key, work }) => {
        const input = obj[key];
        const output$ = work(input);
        return Observable.from(output$).map(output => ({
          index,
          result: { [key]: output }
        }))
      })
    })
    .mergeAll()
    .groupBy(
      obj => obj.index,
      obj => obj.result,
      group$ => group$.skip(segments.length - 1)
    )
    .mergeMap(group$ =>
      group$.reduce(
        (obj, result) => Object.assign(obj, result),
        {}
      )
    );

这是一个用法示例：

const result$ = input$.let(
  collate({
    key: 'a',
    work: a => {
      // do stuff with "a"
      return Observable.of(a).map(d => d + '-processed-A');
    }
  }, {
    key: 'b',
    work: b => {
      // do stuff with "b"
      return Observable.of(b).map(d => d + '-processed-B');
    }
  })
);

给定的输入{ a: '1', b: '1 }它将输出{ a: '1-processed-A', b: '1-processed-B' }等，依此类推，在尽可能多地进行同时进行的同时进行分组 - 唯一的缓冲是将所有段匹配在一起的特定输入。

。

这是一个正在运行的演示https://jsbin.com/yuruvar/edit?js ,, output

---

故障

可能会有更清晰/更简单的方法来执行此操作，尤其是如果您可以进行硬编码而不是使它们成为通用的情况。但是让我们分解我做了什么。

const collate = (...segments) => source$ =>
  source$
    // for every input obj we use the index as an ID
    // (which is provided by Rx as autoincrementing)
    .mergeMap((obj, index) => {
      // segments is the configuration of how we should
      // chunk our data into concurrent processing channels.
      // So we're returning an array, which mergeMap will consume
      // as if it were an Observable, or we could have used
      // Observable.from(arr) to be even more clear
      return segments.map(({ key, work }) => {
        const input = obj[key];
        // the `work` function is expected to return
        // something Observable-like
        const output$ = work(input);
        return Observable.from(output$).map(output => ({
          // Placing the index we closed over lets us later
          // stitch each segment back to together
          index,
          result: { [key]: output }
        }))
      })
    })
    // I had returned Array<Observable> in mergeMap
    // so we need to flatten one more level. This is
    // rather confusing...prolly clearer ways but #YOLO
    .mergeAll()
    // now we have a stream of all results for each segment
    // in no guaranteed order so we need to group them together
    .groupBy(
      obj => obj.index,
      obj => obj.result,
      // this is tough to explain. this is used as a notifier
      // to say when to complete() the group$, we want complete() it
      // after we've received every segment for that group, so in the
      // notifier we skip all except the last one we expect
      // but remember this doesn't skip the elements downstream!
      // only as part of the durationSelector notifier
      group$ => group$.skip(segments.length - 1)
    )
    .mergeMap(group$ =>
      // merge every segment object that comes back into one object
      // so it has the same shape as it came in, but which the results
      group$.reduce(
        (obj, result) => Object.assign(obj, result),
        {}
      )
    );

---

我不考虑或不必担心错误处理/传播可能会如何工作，因为这高度取决于您的用例。如果您无法控制每个细分市场的处理，则建议包括某种超时和.take(1)，否则您可能会泄漏订阅。