Welcher rxjs-Operator ist wie concatmap, wartet aber auf jede Anfrage, bevor er die nächste absetzt? - eckig, rxjs, rxjs5

Zum Beispiel sagen wir, ich möchte etwas api machenAnrufe. Die API, mit der ich mich beschäftige, ist anfällig für Race Conditions. Wenn ich also 3 api-Aufrufe gleichzeitig mache und das gleiche Stück Daten auf dem Server aktualisiere, könnte es einen Teil der Daten verlieren.

Daher möchte ich meine Anfragen in die Warteschlange stellen und dann eine davon abwarten, bis die Antwort zurückkommt, bevor die nächste Anfrage ausgelöst wird.

Grundsätzlich brauche ich sowas wie concatMap aber dasProblem mit ConcatMap ist, dass es alle Anfragen gleichzeitig auslöst. Ich brauche concatMap, um einfach zu warten, bevor ich die nächste Anfrage abschicke. Ich benutze rxjs 5.

Hier ist ein Plunker mit angular2, wo Sie auf Schaltflächen klicken können. Wenn Sie 1 Sek. Klicken, wird eine Observable erstellt, die nach 1 Sekunde zurückkehrt. Es gibt 2 Sek. Und 3 Sek. Tasten.

https://plnkr.co/edit/6F4JrVueQX8PjPinZqIk?p=preview

@Component({
selector: "my-app",
template: `
<div>
<h2>Wait:</h2>

<button (click)="start(1)">1 sec</button>
<button (click)="start(2)">2 sec</button>
<button (click)="start(3)">3 sec</button>
</div>
`,
})
export class App {


constructor() {
}

start(wait) {

const waitSecs = parseInt(wait) * 1000;

of("clicked").delay(waitSecs).subscribe(
val => console.log(wait)
)

// Expected behavior:
// I click 3 sec, 2 sec, and 1 sec right after another.  Then
// the console log should output 3, 2, 1.  Right now it"s
// outputting 1, 2, 3.

}
}

Mein ideales Verhalten mit dieser App wäre, nachdem ich 3 Sek., 2 Sek. Und 1 Sek. Direkt hintereinander geklickt habe. Dann sollte die Konsole 3, 2, 1 ausgeben. Jetzt ist es "s Ausgabe 1, 2, 3.

Antworten:

Da Sie nur eine Anfrage ausführen möchten, können Sie die mergeMap() Betreiber mit der concurrency Setzen Sie den Wert auf 1. Beachten Sie, dass sich zu sendende Objekte in Ihrem mergeMap-Operator befinden, was zu Speicherproblemen führen kann.

Rx.Observable.from([3,5,2,1])
.mergeMap(
i => {
return Rx.Observable.of(i)
.do(i => console.log(`starting request for ${i}`))
.delay(1000)
.do(null,null, () => console.log(`finished request for ${i}`));
},
null,
1 /* concurrency limited to 1 */
)
.subscribe(val => console.log(`received value: ${val}`));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.2/Rx.js"></script>

Du solltest etwas Code posten. concatMap sollte für Ihren Anwendungsfall funktionieren. Wenn Sie das Brennen vor dem Abonnement beobachten, bedeutet dies, dass Sie verwenden müssen defer Operator, der zum Subskriptionszeitpunkt feuert, aber ohne Codebeispiele ist es schwer mehr zu sagen.

Nützliche Links zu vergangenen Fragen:

• Wie man das zweite beobachtbare * nur * nach dem ersten startet * ist * komplett * in rxjs gemacht
• Rx.js wartet auf den Abschluss des Rückrufs
• Rxjs, Verständnis verzögern