RXJava2 verwaltet Subskriptionen - java, rx-java2, publishsubject

Ich muss klären, welcher der beste Ansatz für die Verwaltung des spezifischen Szenarios mit RxJava2 ist (die gesamte App-Struktur basiert darauf):

In meiner App können viele Leute Änderungen vornehmendas gleiche Dokument, also muss ich jede Änderung an alle weitergeben, die das Dokument betrachten. Aber dieses Objekt ist sehr komplex und schwer, also muss ich es aus dem Speicher entfernen, wenn die letzte Person es schließt. Und mehr: Das Dokument kann ein Kind aus einem anderen Dokument sein, so dass jede Änderung im übergeordneten Dokument an alle Kinder gesendet werden muss.

Was ich bisher gemacht habe: Ich habe einen Manager erstellt, so dass jeder Dokumentantrag dazu kommt. Wenn jemand an dem Dokument arbeiten muss, sehe ich in einer Map nach, ob das Dokument bereits geöffnet ist. Ist dies nicht der Fall, erstelle ich eine BaseDocument-Instanz, die die Daten aus dem Dokument empfängt, und ein PublishSubject, um die Ereignisse zu verteilen und zu dieser Map hinzuzufügen. Dann abonniere ich den Observer des Benutzers auf dem PublishSubject, um die Änderungen zu erhalten. Immer wenn der Benutzer etwas ändern muss, sendet er die Änderung an das BaseDocument, nimmt die Änderung vor und sendet die neue Version über onNext () an alle. So weit, ist es gut.

Mein Problem ist, dass ich nicht kontrollieren kann, wenn jemanddisposed the document observer, so kann ich nicht kontrollieren, wann das Dokument nicht mehr benötigt wird, damit ich alle nicht gespeicherten Änderungen persistieren und das Objekt zerstören kann. Ich kann keine Subscribe-Liste oder ähnliches finden, außer "hasObservers ()" und Ich möchte keinen Timer hinzufügen, um ihn abzurufen, wenn alles erledigt ist.

Meine "wundersame Antwort" wäre ein Rückrufheißt, wenn der letzte Abonnent disponiert, also könnte ich einfach das Haus putzen und das ganze Objekt wegwerfen, aber ich kann sowas nicht finden. Also, wie kann ich die Abonnements verwalten?

Antworten:

Einer der Ansätze wäre, die Anzahl der Abonnenten zu zählen und zu disponieren. Und wenn die Zahl 0 erreicht, entfernen Sie das Dokument. Es würde ungefähr so ​​aussehen:

int numberOfSubscribers = 0;

...

public Observable<T> expose(){
return subject.asObservable()
.doOnSubscribe(()-> numberOfSubscribers++)
.doOnDispose(()-> {
numberOfSubscribers--;
if (numberOfSubscribers == 0){
//remove the object
}
});

Natürlich müssen Sie hier Unterstützung für das Concurrency-Problem hinzufügen (synchronized / atomic int), dies ist nur ein Entwurf.

Hoffe das hilft :)