使用惯用的 redux observable 排队和取消事件

Question

以下载管理器为例。可以有任意数量的活动下载。

可以调度操作以启动、停止、标记完成以及标记任何特定下载的下载进度。

const START_DOWNLOAD = "START_DOWNLOAD";
const startDownload = payload => ({ type: START_DOWNLOAD, payload });

const DOWNLOAD_PROGRESS = "DOWNLOAD_PROGRESS";
const downloadProgress = payload => ({ type: DOWNLOAD_PROGRESS, payload });

const STOP_DOWNLOAD = "STOP_DOWNLOAD";
const stopDownload = payload => ({ type: STOP_DOWNLOAD, payload });

const COMPLETE_DOWNLOAD = "COMPLETE_DOWNLOAD";
const completeDownload = payload => ({ type: COMPLETE_DOWNLOAD payload });

这些操作将包含一个标识下载的 id，并且可以使用以下 reducer 修改 redux 状态：

const downloadReducer = (state = initialState, action) => {
  switch (action.type) {
    case STOP_DOWNLOAD:
      return {
        ...state,
        [action.payload.id]: {
          state: "IDLE",
        },
      };

    case START_DOWNLOAD:
      return {
        ...state,
        [action.payload.id]: {
          state: "IN_PROGRESS",
          progress: 0,
        },
      };

    case DOWNLOAD_PROGRESS:
      return {
        ...state,
        [action.payload.id]: {
          state: "IN_PROGRESS",
          progress: action.payload.progress,
        },
      };

    case COMPLETE_DOWNLOAD:
      return {
        ...state,
        [action.payload.id]: {
          state: "DONE",
          progress: 100,
        },
      };

    default:
      return state;
  }
};

现在是关于如何使用 redux observable 管理这些操作的异步调度的问题。

例如我们可以这样做：

const downloadEpic = action$ =>
  action$.ofType(START_DOWNLOAD).mergeMap(action =>
    downloader
    .takeUntil(
      action$.filter(
        stop =>
        stop.type === STOP_DOWNLOAD &&
        stop.payload.id === action.payload.id,
      ),
    )
    .map(progress => {
      if (progress === 100) {
        return completeDownload({
          id: action.payload.id
        });
      } else {
        return downloadProgress({
          id: action.payload.id,
          progress
        });
      }
    }),
  );

这行得通。但是，如果我们想限制允许的活动下载数量怎么办。我们可以将 mergeMap 替换为 concatMap 以一次只允许一个活动下载。或者我们可以将 concurrent parameter 提供给 mergeMap 并准确指定我们希望一次允许多少次内部下载器可观察到的执行。

但是，这带来了我们现在无法停止排队下载的问题。

I have created a complete working example that you can try here.

如何以最惯用的方式使用 rxjs 和 redux observable 限制和排队下载？

Answer 1

我已经关注这个问题好几天了，但还没有时间用完整的代码给你一个可靠的答案，因为它相对复杂。所以我只会给你 tl;dr 版本，我希望它总比没有好 :)

我的直觉告诉我，我会让 UI 发送一个代表尝试下载的动作，而不是真正的保证。例如ATTEMPT_DOWNLOAD。史诗将监听此操作并检查当前的活动下载数量是否超过 >= 最大值，如果是，将发出一个操作以将下载排入队列而不是启动它。

您的减速器将存储那些活动的下载 ID 和排队的下载 ID。例如

{ active: ['123', '456', ...etc], queued: ['789'] }

您可以使用它来具体跟踪 active/queued 以及了解它们的数量 active.length、queued.length

当下载完成某个地方的史诗时，将检查是否有任何排队的下载，如果是，则出列一个。具体怎么做主要是个人喜好。例如如果史诗发出 DEQUEUE_DOWNLOAD 或其他什么。

如果发送取消操作，您的缩减程序将需要同时查看 active 和 queued 并删除其中存在的 ID。如果它实际上是活动的而不只是排队那么处理下载的史诗将监听取消操作并且它将停止它并执行与上面相同的出队检查。

这有点 hand-wavy，但重要的是这些：

你的 UI 组件在它尝试之前不应该知道事情将如何或何时排队，尽管它可以通过查看 redux 状态（例如显示 "download queued" 或随便）
注意不要在多个地方重复下载状态。例如如果它处于活动状态，则您的商店中只有一个事实来源表明它是
epics 运行在减速器之后，利用它来发挥你的优势。
可能会有一个史诗，他唯一的工作就是监听 true 下载请求并执行它，而对排队或类似的事情一无所知。这部史诗将被其他多个 epics 重复使用。通过直接将其作为函数调用或通过发出像您真正的 START_DOWNLOAD

并不总是很清楚业务逻辑应该位于何处。例如减速器应该主要是 getters/setters 还是应该有更多自以为是的逻辑并做出决定？在这些情况下没有很多规则。尽量保持一致。

顺便说一句，这完全是我的直觉。我可能已经发现（或者您可能仍然会发现）在实践中这不是一个好的解决方案。只是给出我的初步想法！

使用惯用的 redux observable 排队和取消事件

Queue and cancel events using idiomatic redux observable

rxjs

redux

rxjs5

redux-observable