1

質問は簡単です。どうすれば redux-observable を EventSource で使用できますか?

RxJ では次のようになります。

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)
});
4

1 に答える 1

7

これは、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
        }))
    );
于 2016-12-06T00:09:34.870 に答える