RxJS/ReactiveX適切なモジュールの通信

Question

私はリアクティブプログラミングにはかなり新しいですが、すでに愛しています。しかし、私の脳をそれに切り替えるのはまだ難しいです。私は、「被験者の使用を避ける」、「不純な機能を避ける」、そしてもちろん「必須コードを避ける」などの推奨事項に従っています。

私が達成するのが難しいのは、1つのモジュールが "アクション" /観測可能と登録し、それに反応することができる単純なクロスモジュール通信です。単純なメッセージバスはおそらく動作しますが、これは避けようとしている主題と命令的なコードスタイルの使用を強制します。第二のモジュールは、それが観測「myAction」を「トリガー」加入した瞬間にそう現在

// some sandbox 
class Api { 
    constructor() { 
    this.actions = {}; 
    } 

    registerAction(actionName, action) { 
    // I guess this part will have to be changed 
    this.actions[actionName] = action.publishReplay(10).refCount(); 
    //this.actions[actionName].connect(); 
    } 

    getAction(actionName) { 
    return this.actions[actionName]; 
    } 
} 

const api = new Api(); 

// ------------------------------------------------------------------- 
// module 1 
let myAction = Rx.Observable.create((obs) => { 
    console.log("EXECUTING"); 
    obs.next("42 " + Date.now()); 
    obs.complete(); 
}); 

api.registerAction("myAction", myAction); 

let myTrigger = Rx.Observable.interval(1000).take(2); 

let executedAction = myTrigger 
.flatMap(x => api.getAction("myAction")) 
.subscribe(
    (x) => { console.log(`executed action: ${x}`); }, 
    (e) => {}, 
() => { console.log("completed");}); 

// ------------------------------------------------------------------- 
// module 2 
api.getAction("myAction") 
    .subscribe(
    (x) => { console.log(`SECOND executed action: ${x}`); }, 
    (e) => {}, 
() => { console.log("SECOND completed");}); 

：だからここ

は、私が一緒に遊んだ、単純な出発点です。そして実際の生活のシナリオでは、ajaxコールになる可能性があります。モジュール1から「myAction」が正しく呼び出されるまで、すべての加入者を遅延/待機させる方法はありますか？そして、やはり - 被験者を使って簡単に行うことができますが、私は推奨された方法に従ってそれをやろうとしています。

Answer 1

ここで私が思ったものよりはるかに簡単な解決法があります。単純に2つの観測値を使用します。同様の効果はスケジューラーとsubscribeOnで達成できます。

// some sandbox 
class Action { 
    constructor(name, observable) { 
    this.name = name; 
    this.observable = observable; 
    this.replay = new Rx.ReplaySubject(10); 
    } 
} 

function actionFactory(action, param) { 

    return Rx.Observable.create(obs => { 
    action.observable 
    .subscribe(x => { 
     obs.next(x); 
     action.replay.next(x); 
    }, (e) => {},() => obs.complete); 
    }); 
} 

class Api { 
    constructor() { 
    this.actions = {}; 
    } 

    registerAction(actionName, action) { 
    let generatedAction = new Action(actionName, action); 

    this.actions[actionName] = generatedAction; 

    return actionFactory.bind(null, generatedAction); 
    } 

    getAction(actionName) { 
    return this.actions[actionName].replay; 
    } 
} 

const api = new Api(); 

// ------------------------------------------------------------------- 
// module 1 
let myAction = Rx.Observable.create((obs) => { 
    obs.next("42 " + Date.now()); 
    obs.complete(); 
}); 

let myRegisteredAction$ = api.registerAction("myAction", myAction); 

let myTrigger = Rx.Observable.interval(1000).take(1).delay(1000); 

let executedAction = myTrigger 
.map(x => { return { someValue: x} }) 
.concatMap(x => myRegisteredAction$(x)) 
.subscribe(
    (x) => { console.log(`MAIN: ${x}`); }, 
    (e) => { console.log("error", e)}, 
() => { console.log("MAIN: completed");}); 


// ------------------------------------------------------------------- 
// module 2 
var sub = api.getAction("myAction") 
    .subscribe(
    (x) => { console.log(`SECOND: ${x}`); }, 
    (e) => {console.log("error : " + e)}, 
() => { console.log("SECOND: completed");}); 

Answer 2

私が正しく理解したら、api.getActionを呼び出すと、getActionの呼び出しが完了するまで、そのobservableの次の値を待つようにします。他の値を処理する前に

これは、concatMapを使用して非常に簡単に達成できるものです。 ConcatMapは、observable（あなたの場合はgetActionの呼び出し）を返す関数を取ります。 ConcatMapは、関数内で返されたオブザーバブルが完了するまで、次の値の処理を開始するまで待機します。

このようにコードを変更すると、正常に機能していれば機能するはずです。

let executedAction = myTrigger 
.concatMap(x => api.getAction("myAction")) 
.subscribe(
    (x) => { console.log(`executed action: ${x}`); }, 
    (e) => {}, 
() => { console.log("completed");}); 

myTriggerが新しい価値を持っている場合、観察が完了api.getActionから返されるまで、それは処理されません。