Redux Observable - Как отправить действие, чтобы начать отдельный эпос, затем дождаться ответа (или тайм-аута)

Question

Таким образом, у меня в основном есть подключение к веб-сокету , это позволяет мне отправлять общие сообщения через 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 })

Я изо всех сил пытался воплотить это в код, я думаю, что самая неловкая часть всего эпоса - это то, что я хочу выпустить действие в серединемоя эпопея и жди. Мне это не совсем подходит ... Ани-паттерн? Но я не совсем уверен, как мне это делать.

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 и опубликовал ниже, если это поможет другим.

Обратите внимание, что эта реализация предполагает реализацию websocket из другого ответа . Это также предполагает, что реализация веб-сокета сервера примет «request_id» и ответит тем же «request_id», так что сообщения запроса и ответа могут быть связаны. Вероятно, также стоит отметить, что «epicLinkId» предназначен только для клиентской части и просто позволяет двум создаваемым эпикам связываться друг с другом, без этого вы сможете вызвать 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 };

Где 2 экспортированных эпоса из documents.js пробиваются в объединение Эпики.