如何让 rxjs 暂停/恢复？

Posted 2023-02-19

【中文标题】如何让 rxjs 暂停/恢复？

【英文标题】：How to make rxjs pause / resume?

【发布时间】：2020-06-04 19:32:32

【问题描述】：

现在有一个数组，数组值就是图片链接，例如：

const imageList = [
  'https://cdn.pixabay.com/photo/2020/02/16/20/29/new-york-4854718_960_720.jpg',
  'https://cdn.pixabay.com/photo/2020/02/14/16/04/mallorca-4848741_960_720.jpg',
  'https://cdn.pixabay.com/photo/2020/02/14/04/20/old-city-4847469_960_720.jpg',
  // more...
];

我想用rxjs依次下载（我是Electron应用，所以可以下载）

当第一张图片下载完成后，再下载第二张图片。在下载第三张图片时，用户点击暂停按钮，等待第三张图片下载完成。然后不再下载。当用户点击继续按钮时，从第四张图片开始下载。

我参考这篇文章：https://medium.com/@kddsky/pauseable-observables-in-rxjs-58ce2b8c7dfd中的Buffering (lossless)部分。本文代码为：

merge(
  source$.pipe( bufferToggle(off$, ()=>on$)  ),
  source$.pipe( windowToggle(on$, ()=>off$) )
).pipe(
  // then flatten buffer arrays and window Observables
  flatMap(x => x)
)

演示网址为：https://thinkrx.io/gist/cef1572743cbf3f46105ec2ba56228cd

但是这段代码有两个问题，不符合我的需求。不知道怎么修改。

我使用redux-observable，所以我想用两个动作来触发它们：(this.props.start() / this.props.pause()) 恢复后，数据仍然是一个一个传输，不是一次性释放

当前代码如下所示：

export const epicDownloadResources = (
  action$: ActionsObservable<AnyAction>,
  store$: StateObservable<RootState>,
) => 
  return action$.pipe(
    ofType(appActions.other.start()),
    of([
      'https://cdn.pixabay.com/photo/2020/02/16/20/29/new-york-4854718_960_720.jpg',
      'https://cdn.pixabay.com/photo/2020/02/14/16/04/mallorca-4848741_960_720.jpg',
      'https://cdn.pixabay.com/photo/2020/02/14/04/20/old-city-4847469_960_720.jpg',
    ]),
    mergeMap(() => 
      // code
    ),
    mergeMap((url: string) => 
      // downloading
    )

在真实产品中，会下载公司内部图片资源，而不是其他无版权的图片。

【参考方案1】：

我采取了完全不同的方法。

如果我理解正确，您希望在用户恢复后按顺序继续。这实际上意味着进行窗口化或缓冲是没有意义的。

在我看来，concatMap 嵌套的一些简单使用就足够了。

const pause$ = fromEvent(pauseButton, "click").pipe(
 mapTo(false),
);

const resume$ = fromEvent(resumeButton, "click").pipe(
 mapTo(true),
);

const pauseResume$ = merge(pause$,resume$).pipe(
 startWith(true),
 shareReplay(1),
)

const source = of(...imageList).pipe(
 concatMap((url, i) =>
   pauseResume$.pipe(
     tap(v => console.log(`should resume $v`)),
     filter(v => v), // Only resume if true
     take(1),
     concatMap(() =>
       from(fetch(url)).pipe(
         delay(1000), // Simulate slow request
         mapTo(i) // just for logging which request we just completed
       )
     )
   )
 )
);
source.subscribe(x => console.log(x));

这将暂停开始一个新请求，直到 resume$ 发出一个新值。我相信根据您的情况，这就是您想要的。

我不确定您是否希望在用户暂停的情况下在您的场景中完成第三个请求。我假设你这样做了，但如果没有，你可以在请求之后使用另一个 concatMap 来 pauseResume$ 和过滤器。

stackblitz

【讨论】：

太棒了！这个答案正是我需要的。非常感谢！

【参考方案2】：

这是我的尝试：

const urlArr = Array.from( length: 10 , (_, idx) => 'url/' + idx);
let idx = 0;

const urlEmitter = new Subject();
const url$ = urlEmitter.asObservable();
const stopEmitter = new Subject();
const stopValues$ = stopEmitter.asObservable();

const start$ = fromEvent(start, 'click');
start$.pipe(take(1)).subscribe(() => (stopEmitter.next(false), urlEmitter.next(urlArr[idx++]))); // Start emitting valeus

const stopSubscription = fromEvent(stop, 'click').pipe(mapTo(true)).subscribe(stopEmitter);

const shouldContinue$ = stopValues$.pipe(map(shouldStop => !shouldStop));

const subsequentStartClicks$ = start$.pipe(
  skip(1), // Skip the first time the `start` button is clicked
  observeOn(asyncScheduler), // Make sure it emits after the buffer has been initialized
  tap(() => stopEmitter.next(false)), // shouldContinue$ will emit `true`
);

const downloadedUrls$ = url$.pipe(
  mergeMap(url => of(url).pipe(delay(idx * 500))), // Simulate a file downloading
  combineLatest(shouldContinue$), // Make sure it acts according to `shouldContinue$`
  filter(([_, shouldContinue]) => shouldContinue),
  map(([v]) => v),
  tap((v) => console.warn(v)), // Debugging purposes...

  // Because of `combineLatest`
  // If you click `start` and wait some time, then you click `stop`
  // then you click again `start`, you might get the last value added to the array
  // this is because `shouldContinue$` emitted a new value
  // So you want to make sure you won't get the same value multiple times
  distinctUntilChanged(), 

  tap(() => urlEmitter.next(urlArr[idx++])),

  bufferToggle(
    start$,
    () => stopValues$.pipe(filter(v => !!v)),
  )
);

merge(
  subsequentStartClicks$.pipe(mapTo(false)), // Might not be interested in click events 
  downloadedUrls$
)
  .pipe(filter(v => !!v))
  .subscribe(console.log);

我受到bufferToggle's 图表的启发。

我的想法是遵循相同的方法，除了应该只在start$ 流发出时发出值，并且应该在stop$ 发出时停止。

----X--X----------------------------------> urls$

-Y----------------------------------------> start$

-----------Z------------------------------> end$


-----------[X, X]-------------------------------> urls$

每次按下stop 按钮时，都会将true 值推送到stopValues$ 流中。 shouldContinue$ 确定 url$ 流是否应继续推送值，具体取决于 stopValues$。

StackBlitz.

【讨论】：

非常感谢您的回答，我尝试了您的示例。发现一些小问题，有时会丢失数据

我想我知道问题出在哪里。几个小时后我会去看看。

@Black-Hole 作为快速修复，您可以在按下停止按钮时减少“idx”

感谢您的回复。我试了一下，应该是我能力的问题，最后没有修改成功

@Black-Hole，我已经更新了我的答案。现在应该可以正常工作了！

【参考方案3】：

Pausing and resuming an observable stream, please suggest better options

delayWhen 是一个非常强大的运算符。我的解决方案使用mergeMap 和delayWhen。

功能：重试、节流、暂停、恢复

创建和订阅 Observable

const concurrentLimit = 5
const retryLimit = 10
const source$ = from(new Array(100).fill(0).map((_, i) => i))
// remove <boolean> if not typescript
const pause$ = new BehaviorSubject<boolean>(false);
const pass$ = pause$.pipe(filter((v) => !v));

const throttledTask$ = source$.pipe(
  mergeMap((item) => 
    return of(item).pipe(
      delayWhen(() => pass$),
      mergeMap(async (item) => 
         // you can also throw some errors
         return await new Promise((resolve)=>
             setTimeout(resolve(item), Math.random()*1000))
      ),
      retryWhen((errors$) => errors$.pipe(delay(1000), take(retryLimit)))
    );
  , concurrentLimit)

const subscription = throttledTask$.subscribe(x => console.log(x))

添加暂停/恢复事件处理程序

const pause = () =>  pause$.next(true) 
const resume = () =>  pause$.next(false) 

解释：

delayWhen 将暂停流并等待pass$ 信号发出。 BehaviorSubject 用于组合pass$ 信号，它会在订阅时发出最后一个值。 mergeMap 可以处理异步任务，并且有并发线程数限制参数。当delayWhen 暂停流时， 流将保留在mergeMap 内并占用并发 '线程'。 retryWhen 将重新订阅，直到 errors$.pipe(delay(1000), take(retryLimit)) 发出完成或错误。