redux-observable - 使用redux-observable并订阅Websocket

Question

试图弄清楚如何使史诗般的前进，该史诗将订阅websocket，然后在发出的事件从websocket滚入时调度一些 Action 。

我看到的示例使用的是Multiplex，实际上并未在websocket上调用subscribe，而在更改它方面我有些困惑。

我是这样开始的。但是我认为Redux可以观察到

const socket$ = Observable.webSocket<DataEvent>(
  "ws://thewebsocketurl"
);

const bankStreamEpic = (action$, store) =>
  action$.ofType(START_BANK_STREAM).mergeMap(action => {
    console.log("in epic mergeMap");
    socket$
      .subscribe(
        e => {
          console.log("dispatch event " + e);
         distributeEvent(e);
        },
        e => {
          logger.log("AmbassadorsDataService", "Unclean socket closure");
        },
        () => {
          logger.log("AmbassadorsDataService", "Socket connection closed");
        }
      )
  });

   function distributeEvent(event: DataEvent) : void {
        //this.logger.log('AmbassadorsDataService', 'Event Received: ' + event.command + ' and id: ' + event.id);
        if(event.source === '/ambassadors/bank') {
            if( event.command === 'REMOVE') {
                removeDataEvent(event);
            }
            else if(event.command == 'ADD') {
                loadDataEvent(event);
            }
        }
    }

它抛出一个错误:
未捕获的TypeError:您在期望流的位置提供了“undefined”。您可以提供一个Observable，Promise，Array或Iterable。

任何帮助，将不胜感激!

谢谢

Answer 1

在redux-observable中，您几乎永远不会(除非您知道我为什么要说“几乎”)，除非您自己调用subscribe。而是将Observable链接起来，中间件和其他运算符将为您处理订阅。

如果您要做的只是为收到的每个事件调度一个 Action ，那就很简单:

const socket$ = Observable.webSocket<DataEvent>(
  "ws://thewebsocketurl"
);

const bankStreamEpic = (action$, store) =>
  action$.ofType('START_BANK_STREAM')
    .mergeMap(action =>
      socket$
        .map(payload => ({
          type: 'BANK_STREAM_MESSAGE',
          payload
        }))
    );

根据从套接字接收的消息的内容是什么，您可能(也可能不需要)进行更多的自定义，但是实际上，将其他逻辑放入 reducer 中可能会更好，因为它可能与副作用无关。

您可能会想要一种停止流的方法，它只是一个 takeUntil:

const socket$ = Observable.webSocket<DataEvent>(
  "ws://thewebsocketurl"
);

const bankStreamEpic = (action$, store) =>
  action$.ofType('START_BANK_STREAM')
    .mergeMap(action =>
      socket$
        .map(payload => ({
          type: 'BANK_STREAM_MESSAGE',
          payload
        }))
        .takeUntil(
          action$.ofType('STOP_BANK_STREAM')
        )
    );

我使用 mergeMap是因为您这样做了，但在这种情况下，我认为 switchMap更合适，因为每个具有多个的似乎是多余的，除非您需要多个并且您的问题只是忽略了每个的独特之处。