Por que o Couchbase escolheu o RxJava para o novo SDK Java

Esta postagem do blog explica nossos motivos e motivação por trás da escolha do RxJava como um dos componentes integrais do nosso novo SDK Java.

Motivação

Há muitas maneiras de projetar uma API e cada uma tem seu próprio conjunto de benefícios (e desvantagens). No processo de criação de nossas novas APIs, uma das principais perguntas era como expô-la ao usuário. 

Uma pergunta que não precisávamos nos fazer era: deve ser síncrono ou assíncrono? Acreditamos firmemente que APIs assíncronas são a única maneira sensata de obter o desempenho e a escalabilidade de que você frequentemente precisa, e também é muito mais fácil ir de assíncrono para síncrono do que o contrário. O SDK estável atual (1.4.3 no momento em que escrevo) já faz uso intenso de Futures de várias maneiras para fornecer respostas assíncronas, e isso remonta a 2006/7, quando o spymemcached introduziu originalmente o conceito em sua API.

É bem sabido que a interface Future do Java é muito limitada em comparação com outras soluções (como os futures do Scala). Além disso, também é um pouco mais complicado programar se você precisar construir fluxos de dados assíncronos onde uma computação depende da outra e você quer que tudo seja assíncrono. Em versões recentes, adicionamos suporte a ouvintes, o que melhora bastante a situação, mas ainda não é uma solução ideal.

Ao longo dos últimos anos, outras bibliotecas e padrões surgiram, os quais acompanhamos de perto. Um dos conceitos maduros é conhecido como Reactive Extensions, originado na Microsoft e no .NET. Ele se baseia na ideia de que as aplicações devem ser orientadas a eventos e reagir a esses eventos de maneira assíncrona. Ele define um conjunto muito rico de operadores sobre o que você pode fazer com os dados (modificá-los, combiná-los, filtrá-los e assim por diante). Recentemente, a Netflix o portou para o Java e o apelidou de RxJava (observe que, embora o projeto atualmente viva sob o namespace da Netflix, ele será movido para “io.reactivex” mais cedo ou mais tarde). Ele é muito estável e também fornece adaptadores para outras linguagens da JVM, como Scala, Groovy e JRuby, o que se alinha bem com nossos planos de ampliar o suporte também.

O Conceito

A ideia principal do Rx gira em torno de Observables e seus observadores. Se você ainda não se deparou com esse conceito, pode pensar no Observable como o primo assíncrono e baseado em push (ou, mais formalmente, um dual) de um Iterable. Mais especificamente, aqui está a relação entre eles:

Evento	Iterável (pull)	Observável (push)
recuperar dados	T next()	onNext(T)
encontrar erro	lança uma exceção	onError(Exception)
concluído	devoluções	onCompleted()

Toda vez que os dados são enviados para um Observable, cada observe que está inscrito nele recebe os dados em seu método onNext(). Se o observável for concluído eventualmente (o que não precisa ser sempre o caso), o método onCompleted é chamado. Agora, em qualquer lugar do processo, se ocorrer um erro, o método onError é chamado e o Observable também é considerado concluído.

Se você gosta de gramática, o contrato se parece com isto: 

OnNext* (OnCompleted | OnError)?

Observe especificamente que não há distinção se apenas 1 ou N dados são retornados, isso pode ser normalmente inferido a partir dos métodos que você chama e de como ele está documentado. Isso não altera seu fluxo de programação de qualquer maneira. Visto que isso é um pouco abstrato, vamos ver um exemplo concreto. Na classe CouchbaseCluster, existe um método chamado openBucket que inicializa todos os recursos necessários e então retorna uma instância de Bucket para você trabalhar. Agora você pode imaginar que abrir sockets, obter uma configuração e assim por diante leva algum tempo, então este é um candidato perfeito. A API de bloqueio se pareceria com:

interface Cluster {
        Bucket openBucket(String name, String password);
}

Como podemos torná-lo assíncrono? Precisamos envolvê-lo em um Observable:

interface Cluster {
        Observável

openBucket(String name, String password);
}

Então, agora retornamos um observável que eventualmente retornará com uma instância de bucket que podemos usar. Vamos adicionar um observador:

cluster.abrirBucket().Inscrever-se(novo ObservadorBalde>() {
    @Override
    público vazio aoConcluir() {
        Sistema.fora.println(“Observável concluído!”);
    }

    @Override
    público vazio aoErro(Lançável e) {
        Sistema.erro.println(“Algo aconteceu”);
        e.printStackTrace();
    }

    @Override
    público vazio emSeguida(Balde balde) {
        Sistema.fora.println(“Bucket recebido: “ + balde);
    }
});

Observe que esses métodos são chamados em uma thread diferente, então, se você deixar o código assim e encerrar sua thread principal logo em seguida, provavelmente não verá nada. Embora agora você possa escrever todo o resto do seu código no método onNext, essa provavelmente não é a melhor maneira de fazer isso. Como o bucket é algo 
quero abrir de forma síncrona, você pode bloquear nisso e então prosseguir com o resto do seu código. Todo Observable pode ser convertido em um observable bloqueante, o que se parece com um Iterable:

BlockingObservable

blockingObservable = cluster.openBucket().toBlocking();

Você encontrará muitos métodos para iterar sobre os dados recebidos de forma bloqueante, mas também existem métodos de Atalho se você espera apenas um único valor (o que sabemos ser o nosso caso):

Bucket bucket = cluster.openBucket().toBlocking().single();

O que acontece aqui internamente é que o valor chamado em onNext é armazenado para nós e retornado uma vez que onComplete é chamado. Se onError for chamado, o throwable é lançado diretamente e você pode capturá-lo.

Unificando APIs

Agora, o que você já viu mal arranha a superfície. A abertura da caçamba também poderia muito bem ser tratada com um Future sozinho. Onde os Observables realmente brilham é quando você precisa trabalhar com mais de um resultado retornado. Nesse caso, um Future não serve mais e Futuroou algo semelhante não possui o mesmo contrato. Como os Observables implicam que mais de um T pode ser retornado, as APIs podem parecer iguais mesmo que às vezes um e às vezes mais de um T sejam retornados.

Novamente, vamos analisar um exemplo concreto. O SDK expõe um método get que retorna um documento. Ele se parece com isto:

interface Bucket {
        Observável

get(String id);
}

Mas também oferecemos suporte a Consultas (Views, N1QL) que potencialmente retornam mais de um resultado (ou mesmo nenhum). Graças ao contrato do Observable, podemos construir uma API assim:

interface Bucket {
        Observável

query(ViewQuery query);
}

Viu? O contrato diz implicitamente “se você passar uma consulta, você recebe N ViewResults de volta”, já que você sabe como um Observable precisa se comportar. E, para ter uma visão mais ampla, aqui estão ainda mais métodos que se comportam intuitivamente da maneira que você espera.

interface Balde {
    D estende Documento>> ObservávelD> inserir(Documento D);
    D estende Documento>> ObservávelD> atualização upsert(Documento D);
    D estende Documento>> ObservávelD> substituir(Documento D);

    ObservávelResultadoDaVisualização> consulta(Consulta de View);
    ObservávelResultado da Consulta> consulta(Consulta consulta);
    ObservávelResultado da Consulta> consulta(String consulta);

    ObservávelBooleano> descarga();
}

Torne meu dataflow assíncrono!

Até agora vimos o que os Observables podem fazer por nós e como eles nos ajudam a fornecer APIs coesas, simples e ainda assim assíncronas. Mas os Observables realmente se destacam por seus aspectos de composibilidade. Você pode fazer muitas coisas com Observables, e não podemos cobrir todas elas nesta postagem. O RxJava possui uma documentação de referência muito boa que pode ser encontrada aqui, então dê uma olhada. Ele usa diagramas de mármore para mostrar como os fluxos de dados assíncronos funcionam, algo que também queremos fornecer como parte de nossa documentação no futuro.

Vamos considerar um exemplo prático: você quer carregar um documento do Couchbase (que é um objeto JSON completo com detalhes do usuário), mas quer apenas fazer algo com o primeiro nome (firstname) mais adiante no seu código. Podemos usar a função map para mapear do JsonDocument para a String do primeiro nome:

balde
    .obter(“user::1”)
    .mapa(novo Func1JsonDocument, String>() {
        @Override
        público String ligar(JsonDocument jsonDocument) {
            retornar documentoJson.conteúdo().obterString(“nome”);
        }
    })
    .Inscrever-se(novo Ação 1String>() {
        @Override
        público vazio ligar(String nome) {
            Sistema.fora.println(nome);
        }
    });

Há dois aspectos importantes aqui: Cada método encadeado aqui também é executado de forma assíncrona, portanto, não está bloqueando a thread de origem. Assim que a chamada get para o Couchbase retorna, mapeamos o primeiro nome (firstname) do documento JSON e, finalmente, o imprimimos. Você não precisa fornecer um Observer completo; se estiver interessado apenas no valor onNext, você pode implementar apenas esse método (como mostrado aqui). Veja os métodos sobrecarregados para mais exemplos.

Observe também que estou mostrando deliberadamente classes anônimas no estilo Java 6/7 aqui. Também oferecemos suporte ao Java 8, mas falaremos mais sobre isso depois. Agora, como poderíamos estender esta cadeia se quisermos apenas imprimir o nome se ele começar com um “a”?

balde
    .obter(“user::1”)
    .mapa(novo Func1JsonDocument, String>() {
        @Override
        público String ligar(JsonDocument jsonDocument) {
            retornar documentoJson.conteúdo().obterString(“nome”);
        }
    })
    .filtro(novo Func1String, Booleano>() {
        @Override
        público Booleano ligar(String s) {
            retornar s.começaCom(“a”);
        }
    })
    .Inscrever-se(novo Ação 1String>() {
        @Override
        público vazio ligar(String nome) {
            Sistema.fora.println(nome);
        }
    });

É claro que uma simples instrução if seria suficiente, mas você pode imaginar que seu código para filtrar pode ser muito mais complexo (e provavelmente chamando outra coisa também). Como exemplo final sobre a transformação de observáveis, vamos fazer algo que acontece com muita frequência: você carrega um documento, modifica seu conteúdo e depois o salva de volta no couchbase:

balde
    .obter(“user::1”)
    .mapa(novo Func1JsonDocument, JsonDocument>() {
        @Override
        público Chamada JsonDocument(JsonDocument original) {
            original.conteúdo().colocar(“nome”, “AlgoDiferente”);
            retornar original;
        }
    })
    .flatMap(novo Func1JsonDocument, ObservableJsonDocument>>() {
        @Override
        público ObservávelJsonDocument> ligar(JsonDocument modificado) {
            retornar balde.substituir(modificado);
        }
    }).Inscrever-se();

O FlatMap se comporta muito como o map, a diferença é que ele retorna um observável por si só, tornando-se perfeitamente adequado para mapear sobre operações assíncronas.

Outro aspecto é que, com Observables, o tratamento de erros sofisticado está literalmente ao seu alcance. Vamos implementar um exemplo que aplica um tempo limite (timeout) de 2 segundos e, se a chamada não retornar, entrega outra coisa em troca:

balde
    .obter(“user::1”)
    .tempo esgotado(2, TimeUnit.SEGUNDOS)
    .onErrorReturn(novo Func1Lançável, JsonDocument>() {
        @Override
        público Chamada JsonDocument(Lançável lançável) {
            retornar JsonDocument.criar(“user::anonymous”, JsonObject.vazio().colocar(“nome”, “john-doe”));
        }
    });

Aqui, um documento fictício é retornado (assumindo alguns padrões razoáveis para o nosso exemplo) se a chamada get não retornar em 2 segundos. Este é apenas um exemplo simples, mas você pode fazer muito com exceções, como tentar novamente, ramificar para outros observáveis e assim por diante. Consulte a documentação oficial (e a documentação do Rx) sobre como usá-las corretamente.

Espere, tem mais

Há muito mais recursos disponíveis, como combinar (mesclar, juntar, concatenar) diferentes observables, agrupar os resultados em intervalos de tempo, realizar efeitos colaterais e outros. Assim que você supera o obstáculo inicial (pequeno) de entender o conceito, parece muito natural e prometemos que você não vai querer voltar atrás (se estivermos errados, no entanto, você sempre pode bloquear em um Observable ou convertê-lo em um future).

O RxJava também tem um suporte razoável ao Java 8, então se você é um dos sortudos que já consegue usá-lo em seus projetos, você pode simplificar o exemplo acima para isto:

balde
    .get(“user::1”)
    .map(jsonDocument -> jsonDocument.content().getString(“firstname”))
    .filter(s -> s.startsWith(“a”))
    .subscribe(System.out::println);

Neat, right? RxJava also provides different language adaptors on top of it, at the time of writing Scala, Clojure, Groovy, JRuby and Kotlin. They can be used to provide even more language-specific integration and we are also planning to use some of them to enhance couchbase support for each of those languages as we see demand. Our topmost priority aside from the java SDK is definitely Scala, so be on the lookout for some announcements sooner than later!

We hope that you are now as excited as we are and looking forward to your feedback and questions through the usual channels!