The Simplicity of RxJava With N1QL Queries and Couchbase

Have you ever tried to query Couchbase using N1QL and the Java SDK's synchronous API? It works well and isn't particularly difficult, but I find it to be a little messy, and well, synchronous. There are a few different ways to address this. You could write better code, or you can take a look at RxJava, which has recently started to grow on me.

In case you're unfamiliar with RxJava, it is a reactive extension set for Java similar to Rx.NET and RxJS. The great thing is that the Couchbase Java SDK integrates very well with this extension set. We're going to take a look at a few snippets of Java code that takes full advantage of RxJava with N1QL.

Let's say that we have the very simple synchronous query code that follows:

public static List<Map<String, Object>> fetch(final Bucket bucket) {
    String queryStr = "SELECT META().id, title, description, type " +
            "FROM `" + bucket.name() + "` " +
            "WHERE type = 'task'";
    N1qlQueryResult result = bucket.query(N1qlQuery.simple(queryStr));
    List<Map<String, Object>> content = new ArrayList<Map<String, Object>>();
    for(N1qlQueryRow row : result) {
        content.add(row.value().toMap());
    }
    return content;
}

In the above example, we have a simple N1QL query. After executing the query, we get a set of results, which we loop through and convert into a Java Map and add to a List. Finally, we return the parsed results. This parsing is common when working with Spring Boot applications, but not limited to.

While the above code wasn't particularly difficult, there was a lot going on. With RxJava, the above code can be converted into the following:

public static List<Map<String, Object>> getAll(final Bucket bucket) {
    String queryStr = "SELECT META().id, title, description, type " +
            "FROM `" + bucket.name() + "` " +
            "WHERE type = 'task'";
    return bucket.async().query(N1qlQuery.simple(queryStr))
            .flatMap(AsyncN1qlQueryResult::rows)
            .map(result -> result.value().toMap())
            .toList()
            .timeout(10, TimeUnit.SECONDS)
            .toBlocking()
            .single();
}

Alright, you got me! The RxJava code has more lines, but does that mean it is more complicated or messy?

RxJava uses observables, which can be considered values over time. They are asynchronous and can contain a chain of operations as data flows through the pipeline. For example, in our above code, we execute the N1QL query asynchronously. The .flatMap will do asynchronous transformations against the data that is returned in the query from AsyncN1qlQueryResult to type AsyncN1qlQueryRow and the .map will transform each of the rows returned into a Java Map. Because we want a List of the Java Map we can call toList, which will do exactly that. Because we don't want to work with an observable in the end, we call toBlocking, which converts the result into an iterable value, or in our case, something the parent function or maybe front-end can work with. We finally end with .single because we know we will only ever be returning one List.

RxJava is powerful and can be applied to things much more complicated than the example above. The point here is that it is simple to use and provides an event-based approach to asynchronous data manipulation.