Wie Gruppen von Observablen, die zuvor jeweils in einer forkjoin() -Operation waren

Question

Ich habe eine Reihenfolge Tabelle in MySQL, jeder Auftrag hat eine Reihe von Dokumenten zugeordnet, ob es sich um Angebote, Rechnungen usw. Es gibt daher eine zweite Tabelle mit dem Namen "Dokumente", die einen Primärschlüssel "Dokument_ID" und einen Fremdschlüssel "Order_ID" aufweist; In ähnlicher Weise habe ich einen anderen Fall für die verschiedenen Überprüfungen, die Techniker zu jedem Fahrzeug machen, dann einen anderen Tisch für Fahrzeugbilder. Ich bin einen Web-Service zu schaffen Knoten und Express verwendet, die eine json, die ähnliche dazu zurückkehren muss ...

[ 
    { 
    "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" 
     ... 
    } 
    }, 
    { 
    ... 
    }, 
    ... 
] 

Wie Sie sehen können, ist es notwendig, drei Anfragen für jeden Auftrag zu tun, einen für Schecks, ein anderes für Dokumente und ein drittes für Bilder, dann muss ich diese Unterergebnisse zur Reihenfolge hinzufügen, um schließlich das Array in der Antwort zurückzugeben.

Dies wäre eine sehr einfache Aufgabe in der alten Welt der synchronen Programmierung, aber aufgrund der asynchronen Natur der Methode query() im Verbindungsobjekt der mysql-Bibliothek, werden diese Bedrohungen zur wahren Hölle.

In einer Situation, in der ich eine einzige Bestellung bearbeiten müsste, würde die Verwendung der RxJS-Bibliothek auf dem Server mit einem forkJoin() genügen, um alle drei Ergebnisse gleichzeitig zu verarbeiten, was ich nicht sicher bin bestellen (mit einem forkJoin für die Verwaltung der 3 Abfragen), so wird alles Prozess und am Ende kann ich res.json (Ergebnis) mit allem sauber zusammengebaut aufrufen.

Hinweis: Ich möchte dies mit RxJS lösen, anstatt ein sync-Bibliothekspaket wie node-mysql-libmysqlclient zu verwenden. Der Grund dafür ist, dass der "richtige" Weg, dies in einer asynchronen Sprache wie Node JS zu tun, async ist. Ich möchte auch RxJS und nicht async, q Versprechen oder irgendeine andere Bibliothek verwenden, da Observables der absolute Gewinner im async solutions Wettbewerb sind und auch in allen von mir entwickelten Lösungen konsistent sein wollen, daher ist diese Frage hauptsächlich darauf ausgerichtet RxJS-Master.

Auch jede einzelne Frage, die ich in so ähnlich gefunden habe, hat die klassische "puristische" Antwort, die besagt, dass man bei Verwendung von Node "asynchron" und nicht in synchronen Lösungen denken sollte. Das ist eine Herausforderung für diejenigen, die diese Position verteidigen, da dies (glaube ich) einer der Fälle ist, in denen Sync in Node Sinn macht, aber ich möchte wirklich lernen, wie man das mit RxJS macht, anstatt zu denken, dass dies unmöglich ist. was ich sicher nicht bin.

Answer 1

Wenn ich die Dinge richtig verstanden habe, haben Sie einige Daten, die Sie verwenden möchten, um zusätzliche Daten aus der Datenbank über asynchrone Operationen zu sammeln. Sie möchten einen kombinierten Datensatz erstellen, der aus den ursprünglichen Daten und den zusätzlichen Informationen besteht, die die nachfolgenden Abfragen zurückgegeben haben.

Wie Sie bereits erwähnt haben, können Sie mit forkJoin warten, bis mehrere Vorgänge abgeschlossen sind, bevor Sie fortfahren. Sie müssen dies für jedes Element in der Datensequenz tun und dann switchMap verwenden, um das Ergebnis wieder in den ursprünglichen Datenstrom einzufügen.

Werfen Sie einen Blick auf die following example jsbin die zeigen, wie dies getan werden kann:

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