RxJava and Kotlin #2

Subjects! Oque são, pra que servem, como são criados.

Um meio de adicionar valores à um observable em runtime para serem emitidos e subscritos.

No capítulo anterior, vimos o que são observables e como criar um observable. Nesse vamos aprender sobre subjects, para que servem e como criá-los.

Antes de começarmos, seria interessante vermos um pouco sobre dispose.

Como encerrar sua stream?

Vimos no epsódio passado como criar uma stream de observable, que se mantém ativa até o evento onError ou onComplete ser chamado. Contudo, você pode, de forma manual, encerrar sua stream.

O método subscribe() retorna um objeto do tipo Disposable(), que é utilizado para interromper ou cancelar a subscrição a qualquer momento, e assim, encerrar a stream.

Para não precisarmos cuidar de cada dispose separadamente, RxJava nos disponibiliza um meio de agrupar todos os disposes para que, de forma única, sejam descartados. Aqui um exemplo:

Como vemos acima, a subscrição ao observable é feita na linha 10 e o retorno dessa subscrição é salvo na variável disposable, depois, na linha 12, adicionamos o dispose ao compositeDisposable para quando o método onDestroy for chamado a stream ser encerrada e prevenir memory leaks ou crashes.

Por exemplo, imagine que você utilize um stream de eventos para atualizar algum componente da UI, se você não encerrar a stream quando a activity não estiver mais visível e tentar atualizar algum componete da UI, seu App provavelmente vai crashar.

Agora, vamos abordar o conceito de subjects.

O que são e para que servem os Subjects?

Subjects proporcionam um meio de adicionar algum valor à uma chain de observables para ser emitido até seu subscritor.

Os subjects agem ao mesmo tempo como observable e observer. Eles observam eventos, e a cada evento recebido, passam a agir como um observable, emitindo esses eventos até o subscritor.

Ao longo do artigo você verá muito a função onNext() sendo chamada. Ela serve para enviar enventos ao seu subject que, passando a agir como observable, emitirá esse evento até o subscritor.

No RxJava existem 4 tipos de subjects que serão vistos na sequência, são eles:

• PublishSubject
• BehaviorSubject
• ReplaySubject
• AsyncSubject

PublishSubject:

Como o nome nos dá a dica, esse subject pode ser entendido como um editor de jornal, pois, ele recebe um evento (observer pattern), resultado da função onNext(), e a partir daí ele se transforma em um observable e publica (emite) esse evento para seus subscritores.

Importante observar que este subject não armazena valores em um cache e emite apenas novos elementos a partir do momento em que são subscritos. Deste modo, os subscritores só receberão os eventos emitidos após a subscrição.

Aqui um exemplo para fixar o conteúdo:

Por que somente a String “Second onNext()” fora printada? Como você já aprendeu, esse tipo de subject não armazena nada em seu cache, emitindo apenas eventos criados após a subscrição. Assim, o evento emitido na linha 3, ocorreu antes da subscrição feita na linha 5, desta forma, apenas o evento da linha 7, que ocorreu após a subscrição, foi printado.

Caso seu observable termine com um erro, nenhum novo elemento será transmitido pela stream e o erro será propagado até o subscritor. A forma como fazer o tratamento de erros será abordado em um post específico.

BehaviorSubject:

BehaviorSuibject funciona basicamente como o PublishSubject, a não ser pelo fato de reemitir o último next event para o novo subscritor. Pode-se verificar esse comportamento ao observar o diagrama acima apresentado. O círculo verde é reemitido para o novo subscritor, embora o círculo verde tenha sido eimitido antes da subscrição.

Em outras palavras, BehaviorSubject guarda seu último onNext event no cache, que é repassado ao novo subscritor.

Utilizando o exmplo anterior podemos verificar que agora todos os valores são printados no console:

Outro benefício que obtemos ao utilizar esse tipo de subject é a possibilidade de acessar o último valor salvo no cache de maneira imperativa. Ainda tomando o exemplo acima emprestado, obtemos acesso à String Second onNext() chamando o seguinte código:

val lastValue: String? = behaviorSubject.value

O ponto de interrogação ao lado do tipo String indica que o valor recuperado pela função behaviorSubject.value pode ser nula.

Caso seu observable termine com um erro, nenhum novo elemento será transmitido pela stream, e o erro será propagado até o subscritor. A forma como fazer o tratamento de erros será abordado em um post específico.

ReplaySubject:

Assim como o BehaviorSubject, este tipo de subject também armazena dados no cache, contudo, não se restringe apenas ao último valor, como é o caso do BehaviorSubject, mas armazena tantos elementos quanto couberem no buffer que você especifica no momento da criação.

Tal característica é percebina se olharmos o gráfico acima, as esferas vermelha e verde são emitidas para o novo subscritor.

Apenas um aviso, o buffer criado pelo subject é mantido na memória, então cuidado ao utilizar grandes buffers para objetos que também façam uso pesado da memória, como imagens, por exemplo.

Vejamos:

Vamos lá!

Notamos que o tamanho do buffer foi definido em 2 quando da criação do nosso subject.

Antes de qualquer subscrição houveram 3 emissões. Por isso que o 1° e 2° subscritores, linha 6 e 8 respectivamente, somente receberam o segundo e terceiro evento.

Daí um quarto evento foi emitido (linha 10) e um 3° subscritor se juntou ao grupo (linha 12). Como o 1° e 2° subscritores não são mais novatos, eles não precisam receber os valores do cache, apenas as novas emissões. Diferente é o caso do 3° subscritor, que, como novato, deve receber tudo.

AsyncSubject:

Chegamos ao último tipo de subject disponível no RxJava. Este último, confesso, aprendi quando estava escrevendo este post e ainda não tive a necessidade de utilizá-lo.

O diagrama acima nos sugere que tal subject somente emitirá o último valor que antecede o evento onComplete para quem estiver subscrito. Dessa forma, embora você possa chamar o onNext várias vezes, quem estiver observando esses eventos não verá nada, até que o onComplete seja chamado.

Vamos ver a implementação:

Tal subject pode ser útil se você quiser manter o registro de elgum evento e apenas exibí-lo ao final, quando a operação estiver concluida. Exemplo seria o score de algum jogo, que, quando completado, é exibido.

TL;DR

Nesse epsódio vimos oque são os subjects, para que servem e como são criados.

Em resumo, podemos concluir que os subjects existem para atuar como uma ponte, utilizada para unir o mundo reativo com o mundo não reativo.

Também vimos a importância de cuidarmos do dispose da nossa stream para evitar memory leaks ou crashes, encerrando nossa stream sempre que a acitivty não estiver mais visível -onStop ou onDestroy.

É isso! Obrigado por ter lido até aqui, no próximo epsódio iniciarei a abordagem dos operadores, começando com os filtering operators. Não perca!