Exemplos de implementações de assinantes personalizados

reactive streams

Esta biblioteca é uma implementação da especificação de fluxos reativos. A API de fluxo reativo consiste nos seguintes componentes:

1. Editora

2. Assinante

3. inscrição

Um Publisher é um provedor de um número potencialmente ilimitado de elementos sequenciados, publicados de acordo com a demanda recebida de seu Subscriber ou múltiplas instâncias de Subscriber.

Em resposta a uma chamada para Publisher.subscribe(Subscriber), as possíveis sequências de invocação para métodos na classe Subscriber são fornecidas pelo seguinte protocolo:

onSubscribe onNext* (onError | onComplete)?

Isso significa que onSubscribe é sempre sinalizado, seguido por um número possivelmente ilimitado de onNext sinais, conforme solicitado por Subscriber. Isso é seguido por um sinal onError se houver uma falha ou um sinal onComplete quando não houver mais elementos disponíveis, desde que o Subscription não seja cancelado.

Assinantes

A API do driver Java Reactive Streams espelha a API do driver Java Sync e quaisquer métodos que fazem com que a E/S da rede retorne um tipo Publisher<T> , em que T é o tipo de resposta para a operação.

Implementações de assinantes personalizados

Na documentação do Java Reactive Streams, implementamos diferentes tipos de Subscriber . Embora seja um cenário artificial para fluxos reativos, bloqueamos os resultados de um exemplo antes de iniciar o próximo para garantir o estado do banco de dados. Para ver o código fonte de todas as implementações de assinatura personalizada, consulte SubscriberHelpers.java no código fonte do driver.

• ObservableSubscriber

  A classe de assinante base é a ObservableSubscriber<T>, um Subscriber que armazena os resultados do Publisher<T>. Ele também contém um método await() para que possamos bloquear os resultados para garantir o estado do banco de dados antes de prosseguir para o próximo exemplo.

• OperationSubscriber

  Uma implementação do ObservableSubscriber que chama imediatamente Subscription.request() quando ela é inscrita.

• PrintSubscriber

  Uma implementação do OperationSubscriber que imprime uma mensagem quando o método Subscriber.onComplete() é chamado.

• ConsumerSubscriber

  Uma implementação de OperationSubscriber que recebe um Consumer e chama Consumer.accept(result) quando Subscriber.onNext(T result) é chamado.

• PrintToStringSubscriber

  Uma implementação de ConsumerSubscriber que imprime a versão de string de result quando o método Subscriber.onNext() é chamado.

• PrintDocumentSubscriber

  Uma implementação do ConsumerSubscriber que imprime a versão JSON de um tipo Document quando o método Subscriber.onNext() é chamado.

Exemplos de bloqueio e não bloqueio

Como nossos tipos de Subscriber contêm uma trava que só é liberada quando o método onComplete() do Subscriber é chamado, podemos usar essa trava para bloquear outras ações chamando o método await . Os dois exemplos a seguir usam nosso PrintDocumentSubscriber de solicitação automática .

O primeiro não está bloqueando e o segundo bloqueia aguardando a conclusão do Publisher :

// Create a publisher
Publisher<Document> publisher = collection.find();
// Non-blocking
publisher.subscribe(new PrintDocumentSubscriber());
Subscriber<Document> subscriber = new PrintDocumentSubscriber();
publisher.subscribe(subscriber);
subscriber.await(); // Block for the publisher to complete

Editores, assinantes e assinaturas

Em geral, os tipos Publisher, Subscriber e Subscription compreendem uma API de baixo nível e espera-se que os usuários e as bibliotecas criem APIs mais expressivas sobre eles, em vez de usar apenas essas interfaces. Como uma biblioteca que implementa apenas essas interfaces, os usuários se beneficiarão desse ecossistema em crescimento, que é um princípio fundamental de design dos fluxos reativos.