Redux Observables：具有多个事件的自定义观察者？

Question

您好，我正在尝试将 redux-observables 与 react native 和一个名为 Phoenix 的 websocket 包装器一起使用，它基本上允许您在通过 websocket 接收到某些消息时执行回调。

这是我正在尝试做的基本设置：

import 'rxjs';
import { Observable } from 'rxjs';
import { Socket, Channel } from 'phoenix';

import * as channelActions from '../ducks/channel';

export default function connectSocket(action$, store) {
  return action$.ofType(channelActions.SETUP)
    .mergeMap(action => {
      return new Observable(observer => {
        const socket = new Socket('http://localhost:4000/socket');

        const userId = store.getState().user.id;
        const channel = socket.channel(`user:${userId}`);

        channel
          .join()
          .receive('ok', response => observer.next({ type: 'join', payload: 'something' }))
          .receive('error', reason => observer.next({ type: 'error', payload: 'reason' }))

        channel.on('rooms:add', room => observer.next({ type: 'rooms:add', payload: '123' }))
        channel.on('something:else', room => observer.next({ type: 'something:else', payload: '123' }))
      });
    })
    .map(action => {
      switch (action.type) {
        case 'join':
          return channelActions.join(action.payload);
        case 'error':
          return channelActions.error(action.payload);
        case 'rooms:add':
          return channelActions.add(action.payload);
        case 'something:else':
          return channelActions.something(action.payload);
      }
    });
};

如您所见，这种方法存在几个问题/问题：

1. 一个观察者正在触发不同的事件。除了.map() 函数中的 switch 语句之外，我不知道如何拆分它们

2. 我不知道在哪里打电话给observer.complete()，因为我希望它继续侦听所有这些事件，而不仅仅是一次。

3. 如果我可以将channel 常量提取到一个单独的文件中，该文件可用于几个可以解决这些问题的史诗。但是channel依赖于socket，而socket又依赖于来自redux状态的user。

所以我很困惑如何解决这个问题。我认为提取全局 channel 对象的能力可以修复它，但这也取决于 redux 状态中的用户 ID。任何帮助表示赞赏。

附：如果值得的话，我的用例与这个人 (https://github.com/MichalZalecki/connect-rxjs-to-react/issues/1) 非常相似。其中一位响应者建议使用Rx.Subject，但我不知道从哪里开始...

Answer 1

似乎你让你的生活变得更加困难。您已经映射了事件，那么为什么要组合和重新映射它们？将每个事件映射到自己的流中并将结果合并在一起。

// A little helper function to map your nonstandard receive
// function into its own stream
const receivePattern = (channel, signal, selector) => 
  Observable.fromEventPattern(
    h => channel.receive(signal, h),
    h => {/*However you unsubscribe*/},
    selector
  )

export default function connectSocket(action$, store) {
  return action$.ofType(channelActions.SETUP)
    // Observable.create is usually a crutch pattern
    // Use defer or using here instead as it is more semantic and handles
    // most of the stream lifecycle for you
    .mergeMap(action => Observable.defer(() => {
        const socket = new Socket('http://localhost:4000/socket');
        const userId = store.getState().user.id;
        const channel = socket.channel(`user:${userId}`);
        const joinedChannel = channel.join();

        // Convert the ok message into a channel join
        const ok = receivePattern(joinedChannel, 'ok')
          // Just an example of doing some arbitrary mapping since you seem to be doing it 
          // in your example
          .mapTo('something')
          .map(channelActions.join);

        // Convert the error message
        const error = receivePattern(joinedChannel, 'error')
          .map(channelActions.error);

        // Since the listener methods follow a standard event emitter interface
        // You can use fromEvent to capture them
        // Rather than the more verbose fromEventPattern we used above
        const addRooms = Observable.fromEvent(channel, 'rooms:add')
          .map(channelActions.add);

        const somethingElse = Observable.fromEvent(channel, 'something:else')
          .map(channelActions.somethingElse);

        // Merge the resulting stream into one
        return Observable.merge(ok, error, addRooms, somethingElse);
      });
  );
};