RXJava2 gerencia as assinaturas - java, rx-java2, publishsubject

Preciso esclarecer qual é a melhor abordagem para gerenciar o cenário específico usando o RxJava2 (toda a estrutura do aplicativo é baseada nele):

No meu aplicativo, muitas pessoas podem fazer alteraçõeso mesmo documento, por isso preciso entregar todas as alterações para todos que estiverem vendo o documento. Mas esse objeto é muito complexo e pesado, então preciso removê-lo da memória quando a última pessoa o fechar. E mais: o documento pode ser um filho de outro documento, portanto, todas as alterações no documento pai devem ser enviadas para todos os filhos.

O que eu fiz até agora: Eu criei um gerenciador para que cada solicitação de documento chegue a ele. Sempre que alguém precisa trabalhar no documento, vejo em um Mapa se o documento já está aberto. Se não estiver, criei uma instância BaseDocument, que recebe os dados do documento e um PublishSubject para distribuir os eventos e adicioná-los ao mapa. Então eu assino o Observer do usuário no PublishSubject para obter as alterações. Sempre que o usuário precisa mudar alguma coisa, ele envia a mudança para o BaseDocument, faz a mudança e envia a nova versão através de onNext () para todos. Por enquanto, tudo bem.

Meu problema é que eu não consigo controlar quando alguémdescarta o observador de documentos, então eu não posso controlar quando o documento não é mais necessário, então eu posso persistir quaisquer alterações não salvas e destruir o objeto. Eu não consigo encontrar qualquer lista de assinaturas ou algo assim, exceto "hasObservers ()" e Eu não quero adicionar um temporizador para pesquisa para fechar, se tudo estiver feito.

Minha "resposta milagrosa" seria um retorno de chamadaé chamado quando o último assinante se desfaz, então eu poderia apenas limpar a casa e jogar todo o objeto fora, mas não consigo encontrar nada como isso. Então, como posso gerenciar as assinaturas?

Respostas:

Uma das abordagens seria contar o número de inscritos e descartes. E se o número atingir 0, remova o documento. Seria algo parecido com isto:

int numberOfSubscribers = 0;

...

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

É claro que você precisa adicionar suporte para o problema de simultaneidade aqui (synchronized / atomic int), isso é apenas um rascunho.

Espero que isto ajude :)