In meiner Android App habe ich einen Moderator, der Benutzerinteraktionen abwickelt, einen Anforderungsmanager enthält und bei Bedarf Benutzereingaben über den Anforderungsmanager an den Anforderungsmanager sendet.
Der Anforderungsmanager selbst enthält die Server-API und verarbeitet Serveranforderungen mit diesem RxJava.
Ich habe einen Code, der immer dann eine Anfrage an den Server sendet, wenn ein Benutzer eine Nachricht eingibt und die Antwort vom Server anzeigt:
%Vor%Aber jetzt muss ich irgendwie Schlange stehen. Der Benutzer kann eine neue Nachricht senden, bevor der Server geantwortet hat. Jede Nachricht aus der Warteschlange sollte sequenziell verarbeitet werden. I.e. Die zweite Nachricht wird gesendet, nachdem wir eine Antwort auf die erste Nachricht erhalten haben und so weiter.
Falls ein Fehler auftritt, sollten keine weiteren Anfragen bearbeitet werden.
Ich muss auch die Antworten in einem RecyclerView anzeigen.
Ich habe keine Ahnung, wie ich den obigen Code ändern kann, um die oben beschriebene Handhabung zu erreichen
Ich sehe Art von Problem. Auf der einen Seite kann diese Warteschlange jederzeit vom Benutzer aktualisiert werden, auf der anderen Seite, wenn der Server eine Antwort gesendet hat, sollte die Nachricht aus der Warteschlange entfernt werden.
Vielleicht gibt es einen Rxjava-Operator oder eine spezielle Art, die ich gerade verpasst habe.
Ich sah eine ähnliche Antwort hier, aber die "Warteschlange" dort ist konstant. N sequenzielle api-Aufrufe mit RxJava und Retrofit machen
Ich bin sehr dankbar für jede Lösung oder jeden Link
Ich finde keine elegante native-RxJava-Lösung. Also werde ich eine Subscriber anpassen, um deine Arbeit zu erledigen.
Für Ihre 3 Punkte:
Für die sequentielle Ausführung erstellen wir einen einzelnen Thread-Scheduler
Scheduler sequential = Schedulers.from(Executors.newFixedThreadPool(1));
Um alle Anfragen zu stoppen, wenn ein Fehler auftritt, sollten wir alle Anfragen zusammen abbonieren, anstatt jedes Mal ein Flowable zu erstellen. Also definieren wir folgende Funktionen (hier ist die Anfrage Integer und Antwort String ):
void sendRequest(Integer request)
Flowable<String> reciveResponse()
und definieren Sie ein Feld, um die Zuordnung von Anfrage- und Antwortfluss zu ermöglichen:
FlowableProcessor<Integer> requestQueue = UnicastProcessor.create();
Um die nicht gesendete Anfrage erneut auszuführen, definieren wir die Rerun-Funktion:
void rerun()
Dann können wir es verwenden:
%Vor%Lasst uns sie jetzt implementieren.
Wenn wir eine Anfrage senden, schieben wir sie einfach in requestQueue
Erstens sollten wir, um die Anfrage nacheinander auszuführen, die Arbeit auf sequential :
Zweitens, um die Anfrage zu stoppen, wenn ein Fehler auftritt. Es ist ein Standardverhalten. Wenn wir nichts tun, wird ein Fehler das Abonnement und alle weiteren Elemente werden nicht ausgegeben.
Drittens, um die nicht gesendeten Anfragen erneut auszuführen. Erstens weil der native Operator den Stream abbrechen wird, wie MapSubscriber do (RxJava-2.1.0-FlowableMap # 63):
Wir sollten den Fehler umbrechen. Hier verwende ich meine % co_de Um die mögliche Ausnahme zu umbrechen, können Sie jede andere Implementierung verwenden, die die Ausnahme umbrechen kann, statt sie zu werfen:
%Vor%
Und dann ist es die benutzerdefinierte Try .
Er fordert und sendet normalerweise Gegenstände. Wenn ein Fehler auftritt (tatsächlich ist ein fehlgeschlagener OnErrorStopSubscriber implements Subscriber<Try<T>>, Subscription -Empfang), hat er dort angehalten und wird keine Downstream-Anfrage anfordern oder ausgeben. Nach dem Aufruf von Try method kehrt es zum laufenden Statu zurück und wird normal ausgegeben. Die Klasse umfasst etwa 80 Zeilen. Sie können den Code auf meinem GitHub sehen .
Jetzt können wir unseren Code testen:
%Vor%und Ausgabe:
%Vor% Sie können sehen, dass es nacheinander läuft. Und gestoppt, wenn ein Fehler auftritt. Nach dem Aufruf rerun method behandelt es weiter die linke nicht gesendete Anfrage.
Ich schlage vor, asynchrone beobachtbare Methoden zu erstellen, hier ein Beispiel:
%Vor%das erste Observable sendet Anfrage, wenn er Antwort erhalten wird, die zweite und Sie können Kette, können Sie jede Methode zur Behandlung von Fehlern oder Erfolg, diese Probe wie Warteschlange.
hier das Ergebnis für die Ausführung:
%Vor%Für diese Art von Verhalten verwende ich die Flowable-Backpressure-Implementierung. Erstellen Sie einen äußeren Stream, der für Ihren API-Anforderungsstream übergeordnet ist, flatMap die API-Anforderung mit maxConcurrency = 1 und implementieren Sie eine Art von Pufferstrategie, sodass Ihr Flowable keine Ausnahme auslöst.
%Vor%Es puffert Benutzereingaben bis zu einem bestimmten Schwellenwert und löscht sie dann (wenn Sie dies nicht tun, wird eine Ausnahme ausgelöst, aber es ist sehr unwahrscheinlich, dass der Benutzer einen solchen Puffer überschreitet), wird er nacheinander 1 zu 1 ausgeführt wie eine Schlange. Versuchen Sie nicht, dieses Verhalten selbst zu implementieren, wenn es Operatoren für ein Verhalten in der Bibliothek selbst gibt.
Oh, ich habe vergessen zu erwähnen, dass deine sendRequest() -Methode Flowable zurückgeben muss, oder du kannst sie in Flowable konvertieren.
Hoffe, das hilft!
Meine Lösungen wären wie folgt (ich habe vorher in Swift etwas ähnliches gemacht):