Couchbase query.

//The below code is executed per HTTP request in vertx verticle.

com.couchbase.client.java.document.json.JsonObject couchBody = com.couchbase.client.java.document.json.JsonObject.empty();
final String fullName = "John Smith";

couchBody.put("fullName", fullName);

final Context context = vertx.getOrCreateContext();

JsonDocument doc = JsonDocument.create("123456789", couchBody);

final AtomicInteger count = new AtomicInteger();

bucket.async().upsert(doc, PersistTo.MASTER).subscribe(d -> {

    String sql = "SELECT * FROM " + BUCKET_NAME + " WHERE fullName =$1";

    N1qlParams params = N1qlParams.build().adhoc(false);
    ParameterizedN1qlQuery query = N1qlQuery.parameterized(sql, JsonArray.create()
            .add(fullName), params);

    bucket.async().query(query)
    .subscribe(r -> {
        r.errors().subscribe(e -> {
            System.out.println(e);
            context.runOnContext((v) -> {
                h.reply(new JsonObject().put("queryErrors", e.toString()));
            });
        }, re -> {
            re.printStackTrace();
            context.runOnContext((v) -> {
                h.reply(new JsonObject().put("queryErrors", re.getCause().toString()));
            });
        });

        r.rows().map(rows -> rows.value()).subscribe(
                row -> {
                     //System.out.println("My row: " + row);
                    count.incrementAndGet();
                }, re -> {
                    re.printStackTrace();
                    context.runOnContext((v) -> {
                        h.reply(new JsonObject().put("queryError", re.getCause().toString()));
                    });
                }, () -> {
                    context.runOnContext((v) -> {
                        h.reply(new JsonObject().put("result", count.get()));
                    });
                });
    });
}, t-> {
    context.runOnContext((v) -> {
        h.reply(new JsonObject().put("insertError", t.toString()));
    });
});