RXJava2 gestionar suscripciones - java, rx-java2, publishsubject

Necesito aclarar cuál es el mejor enfoque para administrar el escenario específico utilizando RxJava2 (toda la estructura de la aplicación se basa en él):

En mi aplicación, muchas personas pueden hacer cambios enel mismo documento, por lo que debo entregar cada cambio a todos los que miren el documento. Pero este objeto es muy complejo y pesado, por lo que debo eliminarlo de la memoria cuando la última persona lo cierre. Y más: el documento puede ser un hijo de otro documento, por lo que cada cambio en el documento padre debe enviarse a todos los niños.

Lo que he hecho hasta ahora: He creado un administrador para que cada solicitud de documento llegue a él. Cuando alguien necesita trabajar en el documento, observo en un mapa si el documento ya está abierto. Si no lo es, creo una instancia de BaseDocument, que recibe los datos del documento y un PublishSubject para distribuir los eventos y agregarlos a ese mapa. Luego me suscribo al observador del usuario en PublishSubject para obtener los cambios. Cuando el usuario necesita cambiar algo, envía el cambio a BaseDocument, realiza el cambio y envía la nueva versión a través de onNext () a todos. Hasta ahora tan bueno.

Mi problema es que no puedo controlar cuando alguiendesecha el observador de documentos, por lo que no puedo controlar más cuando ya no se necesita, así que puedo conservar los cambios no guardados y destruir el objeto. No puedo encontrar ninguna lista de suscripciones ni nada parecido, excepto "hasObservers ()" y No quiero agregar un temporizador para encuestarlo para que se cierre si todo está listo.

Mi "respuesta milagrosa" sería una devolución de llamada quese llama cuando el último suscriptor dispone, por lo que solo puedo limpiar la casa y tirar todo el objeto, pero no puedo encontrar nada como esto. Entonces, ¿cómo puedo gestionar las suscripciones?

Respuestas

Uno de los enfoques sería contar el número de suscripciones y eliminaciones. Y si el número llega a 0, elimine el documento. Se vería algo como esto:

int numberOfSubscribers = 0;

...

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

Por supuesto, debe agregar soporte para el problema de concurrencia aquí (sincronizado / atómico int), esto es solo un borrador.

Espero que esto ayude :)