以前はforkjoin（）操作であったオブザーバブルのグループを連鎖する方法

Question

私はmysqlにordersテーブルを持っていますが、各オーダーには引用符、請求書などの関連する多数のドキュメントがあります。したがって、 "document_id"主キーと "order_id"外部キーを持つ "documents"と呼ばれる第2のテーブル。同様のやり方で、私は、技術者が各車両に対して行うさまざまな検査のための別のケース、次に車両画像用の別の検査を行います。私はこれに似たJSONを返す必要があるノードおよびExpressを使用してWebサービス...

[ 
    { 
    "order_id": 1003, 
    "customer_id": 8000, 
    "csi": 90, 
    "date_admitted": "2016-10-28T05:00:00.000Z", 
    "plates": "YZG-5125", 
    ... 
    documents: { 
     "type": "invoice", 
     "number": "1234", 
     ... 
    }, 
    checks: { 
     "scanner": "good", 
     "battery": "average", 
     ... 
    }, 
    vehicle_pictures: { 
     "title": "a title...", 
     "path": "the file path" 
     ... 
    } 
    }, 
    { 
    ... 
    }, 
    ... 
] 

あなたが見ることができるように作成しています、各注文の3件の問い合わせ、チェックのいずれかを実行する必要があります別のドキュメントと3番目の写真の場合、これらのサブ結果を注文に追加して、最後にレスポンスに配列を返す必要があります。

これは、同期プログラミングの古い世界では非常に簡単な作業ですが、mysqlライブラリの接続オブジェクトのquery（）メソッドの非同期性のために、この脅威は本当に地獄になります。

私は単一の注文を処理する必要がある状況では、サーバー上でforkJoin（）を使用してRxJSライブラリを使用すれば、3つの結果をすべて一度に処理すれば十分でしょう。 3つのクエリを管理するためにforkJoinを使用すると、すべてがプロセスを取得し、最後にはすべてがきれいに組み立てられたres.json（result）を呼び出すことができます。

注：node-mysql-libmysqlclientなどの同期ライブラリパッケージを使用する代わりに、RxJSを使用してこの問題を解決したいと考えています。理由は、基本的に、ノードJSのような非同期言語でこれを行う「正しい」方法は非同期になるということです。またObservablesは非同期ソリューションコンテストの絶対勝者であるように見えますが、私が開発したすべてのソリューションで一貫していたいので、RxJSを使用し、非同期、q約束、またはその他のライブラリを使用したくないので、この質問は主にRxJSマスター。

また、このような点で私が見つけたすべての質問には、古典的な "純粋主義"の回答があります。ノードを使用している場合は、非同期を使用する必要があり、だから、この位置を守る人にとっては、Nodeの同期が意味をなさないケースの1つなので、これは難しいことですが、RxJSを使ってこれを実現する方法を本当に知りたいのですが、私は確信していません。

Answer 1

私が正しく理解していれば、非同期操作でデータベースから追​​加データを収集するために使用したいデータがあります。元のデータと、後続のクエリが返す追加情報からなる結合データセットを構築したいとします。

前述のとおり、続行する前に複数の操作が完了するまで待つには、forkJoinを使用します。データシーケンスの各項目に対してこれを実行し、結果を元のストリームにマージするにはswitchMapを使用する必要があります。あなたがこれまでに試してみました何

const data = [ 
    { id: 1, init: 'a' }, 
    { id: 2, init: 'b' }, 
    { id: 3, init: 'c' } 
] 

function getA(id) { 
    return Rx.Observable.timer(1000) 
    .map(() => { 
     return { id, a: 'abc' } 
    }) 
    .toPromise(); 
} 

function getB(id) { 
    return Rx.Observable.timer(1500) 
    .map(() => { 
     return { id, b: 'def' } 
    }) 
    .toPromise(); 
} 

Rx.Observable.interval(5000) 
    .take(data.length) 
    .map(id => data[id]) 
    .do(data => { console.log(`query id ${data.id}`)}) 
    .switchMap((data) => { 
    return Rx.Observable.forkJoin(getA(data.id), getB(data.id), (a, b) => { 
     console.log(`got results for id ${data.id}`); 
     return Object.assign({}, data, a, b); 
    }); 
    }) 
    .subscribe(x => console.log(x));