如何使用 Redux Observable 获取 EventSource

Question

问题很简单，如何将 redux-observable 与 EventSource 一起使用？

与 RxJs 类似：

const observable = Observable.create(observer => {
  const eventSource = new EventSource('/model-observable');
  return () => {
    eventSource.close();
  };
});

observable.subscribe({
  next: data => {
    this.zone.run(() => this.someStrings.push(data));
  },
  error: err => console.error('something wrong occurred: ' + err)
});

Answer 1

这听起来更像是一个关于如何连接到 EventSource 的通用 RxJS 问题。这可以通过多种方式完成。如果您只关心消息（而不是错误/打开）：

import { fromEvent } from 'rxjs/observable/fromEvent';

const fromEventSource = url => {
  return new Observable(observer => {
    const source = new EventSource(url);
    const message$ = fromEvent(source, 'message');
    const subscription = message$.subscribe(observer);

    return () => {
      subscription.unsubscribe();
      source.close();
    };
  });
};

如果您关心打开和/或错误，则需要更多代码来将所有内容集中在一起：

import { Observable } from 'rxjs/Observable';
import { Subscriber } from 'rxjs/Subscriber';

const fromEventSource = (url, openObserver) => {
  return new Observable(observer => {
    const open = new Subscriber(openObserver);
    const source = new EventSource(url);

    const onOpen = event => {
      open.next(event);
      open.complete();
    };

    const onError = event => {
      if (event.readyState === EventSource.CLOSED) {
        observer.complete();
      } else {
        observer.error(event);
      }
    };

    const onMessage = event => {
      observer.next(event.data);
    };

    source.addEventListener('open', onOpen, false);
    source.addEventListener('error', onError, false);
    source.addEventListener('message', onMessage, false);

    return () => {
      source.removeEventListener('open', onOpen, false);
      source.removeEventListener('error', onError, false);
      source.removeEventListener('message', onMessage, false);
      source.close();
    };
  });
};

fromEventSource('http://some-url.com')
  .subscribe(value => console.log(value));

redux-observable 中的用法是这样的：

const somethingEpic = action$ =>
  action$.ofType(SOMETHING)
    .mergeMap(() =>
      fromEventSource('http://some-url.com')
        .map(message => ({
          type: MESSAGE,
          payload: message
        }))
        .catch(e => Observable.of({
          type: SOMETHING_ERROR,
          payload: e,
          error: true
        }))
    );