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