代码之家 › 专栏 › 技术社区 › Robby Cornelissen

防止RXJS中异步可管道运算符过早完成

rxjs-pipeable-operators rxjs asynchronous javascript

Robby Cornelissen · 技术社区 · 8 年前

我正在创造 pipeable operators 使用RXJS 6,但不清楚如何 complete() 当操作是 异步的 .

对于一个 同步的 操作,逻辑简单。在下面的示例中,源中的所有值 Observable 将传递给 observer.next() 在那之后 observer.complete() 被称为。

const syncOp = () => (source) =>
  new rxjs.Observable(observer => {
    return source.subscribe({
      next: (x) => observer.next(x),
      error: (e) => observer.error(err),
      complete: () => observer.complete()
    })
  });
  
rxjs.from([1, 2, 3]).pipe(syncOp()).subscribe(x => console.log(x));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.2.2/rxjs.umd.min.js">
</script>

对于一个 异步的 手术,不过,我有点不知所措。在下面的示例中,异步操作由对 setTimeout() . 显然, 观察者。完成() 将被称为之前任何值都传递给 观察者(下) .

const asyncOp = () => (source) =>
  new rxjs.Observable(observer => {
    return source.subscribe({
      next: (x) => setTimeout(() => observer.next(x), 100),
      error: (e) => observer.error(err),
      complete: () => observer.complete()
    })
  });
  
rxjs.from([1, 2, 3]).pipe(asyncOp()).subscribe(x => console.log(x));

<script src=“https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.2.2/rxjs.umd.min.js”>
& /脚本& GT;

所以问题是:什么是惯用的RXJS方法来实现它,以便调用 观察者。完成() 仅在所有值异步传递到 观察者(下) ?我应该手动跟踪挂起的呼叫还是有一个更“反应式”的解决方案?

(请注意,上面的示例简化了我的实际代码,并且调用 设置TimeTo() 表示“任何异步操作”。我正在寻找一种处理可管道操作员中异步操作的通用方法,而不是关于如何处理RXJS中的延迟或超时的建议。)

3 回复 | 直到 8 年前

Picci 8 年前

一个思路可能是重组 asyncOp 使用其他运算符,如 mergeMap .

这是使用此方法复制示例的代码

const asyncOp = () => source => source.pipe(mergeMap(x => of(x).pipe(delay(100))));
from([1, 2, 3]).pipe(asyncOp1()).subscribe(x => console.log(x));

这是否值得考虑取决于 异步电动机 做。如果它是异步的,因为它依赖于某种回调,比如在HTTPS调用或从文件系统读取的情况下,那么我认为这种方法可以工作,因为您可以将基于回调的函数转换为可观察的函数。

Robby Cornelissen 8 年前

仍然希望得到关于更被动/惯用实现的输入,但下面是我暂时决定采用的方法。

本质上,我只是在用一个计数器进行飞行操作( pending )使得只有当源可观测数据完成时,运算符才能完成。( completed )没有未决的操作( !pending )

const asyncOp = () => (source) =>
  new rxjs.Observable(observer => {
    let pending = 0; // the number of in-flight operations
    let completed = false; // whether or not the source observable completed
    
    return source.subscribe({
      next: (x) => {
        pending++;
        
        setTimeout(() => {              
          observer.next(x);
          
          if (!--pending && completed) { // no ops pending and source completed
            observer.complete();
          }
        }, 100);
      },
      error: (e) => observer.error(err),
      complete: () => {
        completed = true;
        
        if (!pending) { // no ops pending
          observer.complete();
        }
      }
    })
  });
  
rxjs.from([1, 2, 3]).pipe(asyncOp()).subscribe(x => console.log(x));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.2.2/rxjs.umd.min.js">
</script>

kctang 8 年前

我创建了这个runnable StackBlitz demo 以表明我认为应该做什么。

这里的想法是用 toArray() 将源中可观测到的所有值放入一个数组中。后码 托拉() 是单个值(数组)。

注意:解决问题的方法(操作符)有很多种,这只是一个基于我从这个问题中了解到的示例-这对于RXJS观测值来说既是好也是坏。希望这有帮助。-)

主要演示代码是:

// --- for each value, do the async service
of(...[1, 2, 3]).pipe(
  // let each value be processed by both async service...
  concatMap(no => myAsyncService$(no)),
  concatMap(no => myAsyncService2$(no)),

  // --- toArray() combines all the values (i.e. they completed)
  toArray(),

  // --- this will only be called once - with all completed values
  // --- testing: try commenting the toArray() to see the values as individual "next" value
  tap(val => {
    // see the combined values
    console.log(val)
  })
).subscribe();