RxJS中如何在mergeMap所有observable完成后执行副作用且保留原流

你之前的写法核心问题是：定义的第二个pipe链没有被订阅，RxJS所有流都是冷流，只有被订阅时才会执行内部逻辑，所以tap回调永远不会触发。

方案1：上游流最终会完成的场景 #

如果你处理的action$是有限流（最终会触发complete回调），可以直接通过合并流的方式实现，完全不影响原流的正常输出：

import { merge, of } from 'rxjs'
import { ignoreElements, endWith, tap, share, filter, mergeMap } from 'rxjs/operators'

const main$ = action$.pipe(
  filter(/* 你的过滤逻辑 */),
  mergeMap(action => nextAction(action)),
  share() // 避免多个订阅重复执行mergeMap逻辑
)

const sideEffect$ = main$.pipe(
  ignoreElements(), // 忽略所有正常输出，只传递error和complete信号
  endWith(undefined),
  tap(() => {
    // 你的副作用逻辑
    console.log('所有mergeMap产出的observable已全部完成')
  })
)

// 合并两个流返回，sideEffect$不会输出任何值，完全不干扰原流的正常输出
return merge(main$, sideEffect$)

方案2：上游流是永不完成的无限流（比如redux-observable的action$） #

这种场景下全局的complete信号永远不会触发，你需要明确「所有mergeMap输出的observable完成」的边界：
如果是要监听每一个mergeMap内部的observable完成事件，直接在inner observable上加finalize即可：

import { finalize, filter, mergeMap } from 'rxjs/operators'

return action$.pipe(
  filter(/* 你的过滤逻辑 */),
  mergeMap(action => 
    nextAction(action).pipe(
      finalize(() => {
        // 当前inner observable完成时触发副作用
      })
    )
  )
)

如果是要监听某一批次所有inner observable全部完成，可以通过外部变量计数实现：

import { finalize, tap, filter, mergeMap } from 'rxjs/operators'

let pendingCount = 0
return action$.pipe(
  filter(/* 你的过滤逻辑 */),
  tap(() => pendingCount++),
  mergeMap(action => 
    nextAction(action).pipe(
      finalize(() => {
        pendingCount--
        if (pendingCount === 0) {
          // 所有pending的inner observable全部完成，触发副作用
        }
      })
    )
  )
)

内容的提问来源于stack exchange，提问作者Dennis