Redux Observable/RxJS：カスタムオブザーバブルを作成するには？

Question

私はReduxの-観察可能な叙事詩でのWebSocketのセットアップをやろうとしている、と私はこの男に似たアプローチで行くよ：https://github.com/MichalZalecki/connect-rxjs-to-react/issues/1

しかし、それは配線物事アップで私の最初の刺し傷のように見えるではありません

=== UPDAT：

import 'rxjs'; 
import Observable from 'rxjs'; 

import * as scheduleActions from '../ducks/schedule'; 

export default function connectSocket(action$, store) { 
    return action$.ofType(scheduleActions.CANCEL_RSVP) 
    .map(action => { 
     new Observable(observer => { 
     // do websocket stuff here 
     observer.next('message text'); 
     }); 
    }) 
    .map(text => { 
     console.log("xxxxxxxxxxxxx: ", text); 
     return scheduleActions.rsvpCancelled(1); 
    }); 
}; 

は、しかし、私はObject is not a constructorエラーを取得しています：それは上記の男と同じように見えるにもかかわらず、作業E ===

{Observable}エクスポートを破棄するための提案のように見えます！

だけでなく、問題はObservableクラスは名前の輸出ではなく、デフォルトのエクスポートとして利用可能である、RxJSのV5では

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

import * as scheduleActions from '../ducks/schedule'; 

export default function connectSocket(action$, store) { 
    return action$.ofType(scheduleActions.CANCEL_RSVP) 
    .map(action => { 
     new Observable(observer => { 
     // do websocket stuff here 
     observer.next('message text'); 
     }); 
    }) 
    .map(text => { 
     console.log("xxxxxxxxxxxxx: ", text); // prints undefined 
     return scheduleActions.rsvpCancelled(1); 
    }); 
}; 

Answer 1

... textは、次の方法に渡るようには見えないということです。

定期 rxjsからのインポート

import { Observable } from 'rxjs'; 

も（観察可能なプロトタイプにすべての演算子を追加する）RxJSのすべてをインポートします。これはin the docs hereと記載されています。あなたがより明示的なことを好むとだけObservable自体をインポートしたい場合は、rxjs/Observableでそれを直接インポートすることができます。

import { Observable } from 'rxjs/Observable'; 

---

は別に、あなたはあなたのカスタムが観測マッピングしている途中でカップルの問題を抱えています。

創刊

あなたが実際にそれを返していません。 hehe。 return文がない（または中括弧を削除して、矢印関数の暗黙的な戻り値を使用する）ことができます。

第二の問題

あなたが観測を返却する際、正規.map()オペレータが特別な何もしていません。カスタムObservableをサブスクライブしてフラットにしたい場合は、ある種のフラット化を行う演算子を使用する必要があります。

最も一般的な2つは、mergeMap（別名flatMap）またはswitchMapです。

action$.ofType(scheduleActions.CANCEL_RSVP) 
    .mergeMap(action => { 
    return new Observable(observer => { 
     // do websocket stuff here 
     observer.next('message text'); 
    }); 
    }) 

必要な演算子は、目的の動作によって異なります。まだ慣れていない場合はdocumentation on the various operatorsを確認するか、mergeMapとswitchMapに直接ジャンプしてください。

---

あなたは冒険している場合は、RxJS v5のは、あなたがObservable.webSocket()で試すことができます箱から出してのWebSocketをサポートしていません。それはうまく説明されていませんが、単体テストを見ることもできます。単純な読み込み専用の単方向ストリーミングの場合は、URLを提供し、購読してください。それは実際に信じられないほど強力です、もしあなたがそれを使用する方法を理解することができます。単一のソケットを介して、複雑な複数の入出力チャネルを双方向、多重化することをサポートします。 Netflixで数千rpsの内部ツールを使用しています。