Redux Observable-如何发送动作以启动单独的史诗，然后等待该史诗响应（或超时）

Question

所以我基本上有一个websocket connection，这使我可以通过WEBSOCKET_MESSAGE_SEND发送通用消息，并通过WEBSOCKET_MESSAGE_RECEIVED操作接收它们。

但是在某些情况下，我想以与Ajax REST调用类似的方式发出请求。例如，为一个用户请求文档列表，我可能想要一部史诗片：

1. [接收一个动作，例如({ type: GET_DOCUMENTS })
2. 生成随机密钥以跟踪当前请求，我们将其称为'request_id'
3. 发送({ type: WEBSOCKET_MESSAGE_SEND, request_id })动作。
4. 等待任何一个
   1. 动作({ type: WEBSOCKET_MESSAGE_RECEIVED, request_id, message }) **必须与匹配的“ request_id”一起使用，否则应将其忽略。
      • ->发出动作，例如({ type: GET_DOCUMENTS_SUCCESS, documents: message })
   2. 超时，例如10秒
      • ->发出动作，例如({ type: GET_DOCUMENTS_TIMEOUT })

我一直在努力将其放入代码中，我认为整个史诗中最尴尬的部分是我想在史诗中间发出一个动作并等待。这对我来说不太合适... ani模式？但是我不太确定该怎么做。

Answer 1

是的。没有很好的方法在史诗中间发出动作。将史诗一分为二如何？

const getDocumentsEpic = action$ =>
    action$.pipe(
        ofType("GET_DOCUMENTS"),
        map(() => {
          const requestId = generateRequestId();
          return {
            type: "WEBSOCKET_MESSAGE_SEND",
            requestId
          };
        })
    );

const websocketMessageEpic = action$ =>
    action$.pipe(
        ofType("WEBSOCKET_MESSAGE_SEND"),
        switchMap(requestId => {
          return action$.pipe(
              ofType("WEBSOCKET_MESSAGE_RECEIVED"),
              filter(action => action.requestId === requestId),
              timeout(10000),
              map(({ message }) => ({
                type: "GET_DOCUMENTS_SUCCESS",
                documents: message
              })),
              catchError(() => of({ type: "GET_DOCUMENTS_TIMEOUT" }))
          );
        })
    );

Answer 2

[我觉得我可能需要多次重复此过程，这似乎是相当数量的重复样板，我应该创建一种根据需求生成史诗的方法。因此，我扩展了@sneas很棒的答案，并在下面发布，以防它对其他人有所帮助。

注意，此实现假定other answer中的websocket实现。它还假定服务器Websocket实现将接受“ request_id”并使用相同的“ request_id”进行响应，以便可以链接请求和响应消息。可能还值得注意的是，“ epicLinkId”仅在客户端，并且仅使所创建的2个史诗能够相互链接，否则，您只能调用一次createNotifyReqResEpics()。

createNotifyReqResEpics.js（基于上述代码的帮助器）

import { ofType } from 'redux-observable';
import { of } from 'rxjs';
import { map, switchMap, filter, timeout, catchError, first } from 'rxjs/operators';
import { notificationActionTypes } from '../actions';

const generateRequestId = () => Math.random().toString(16).slice(2);

export default ({
    requestFilter,
    requestMessageMapper,
    responseMessageMapper
}) => {

    if (typeof requestFilter !== "function")
        throw new Error("Invalid function passed into createNotifyReqResEpics 'requestFilter' argument.");
    if (typeof requestMessageMapper !== "function")
        throw new Error("Invalid function passed into createNotifyReqResEpics 'requestMessageMapper' argument.");
    if (typeof responseMessageMapper !== "function")
        throw new Error("Invalid function passed into createNotifyReqResEpics 'responseMessageMapper' argument.");

    const epicLinkId = generateRequestId();

    const websocketSendEpic = action$ =>
        action$.pipe(
            filter(requestFilter),
            map(action => ({
                epic_link_id: epicLinkId,
                type: notificationActionTypes.WEBSOCKET_MESSAGE_SEND,
                message: {
                    request_id: generateRequestId(),
                    ...requestMessageMapper(action)
                }
            }))
        );

    const websocketReceiveEpic = action$ =>
        action$.pipe(
            ofType(notificationActionTypes.WEBSOCKET_MESSAGE_SEND),
            filter(action => action.epic_link_id === epicLinkId),
            switchMap(sendAction =>
                action$.pipe(
                    ofType(notificationActionTypes.WEBSOCKET_MESSAGE_RECEIVED),
                    filter(receiveAction => receiveAction.request_id === sendAction.request_id),
                    first(),
                    timeout(10000),
                    map(receiveAction => responseMessageMapper(false, receiveAction.message)),
                    catchError(errorMessage => of(responseMessageMapper(errorMessage && errorMessage.message, null))))));

    return [websocketSendEpic, websocketReceiveEpic];
};

documents.js（史诗）

import EventTypes from '../shared-dependencies/EventTypes';
import { documentActionTypes, refreshDocumentsError, refreshDocumentsSuccess } from '../actions';
import { createNotifyReqResEpics } from '../utils';

const [getDocumentsReqEpic, getDocumentsRespEpic] = createNotifyReqResEpics({
    requestFilter: action => action.type === documentActionTypes.REFRESH_DOCUMENTS_REQUEST,
    requestMessageMapper: action => ({ eventType: EventTypes.get_user_documents_req }),
    responseMessageMapper: (error, action) => error ? refreshDocumentsError(error) : refreshDocumentsSuccess(action.result)
});

export { getDocumentsReqEpic, getDocumentsRespEpic };

从documents.js导出的2个史诗在其中进入了combinEpics。