<\/p>\n
In this post we’ll go reactive all the way!<\/p>\n
Some Couchbase customers use So we’ll see how to get going fast with a Couchbase This blog post assumes that you are familiar with the basics of Vert.x<\/a>. Here is a short table of content:<\/p>\n <\/p>\n Let’s start by creating a new Maven based project: create a root folder for your project and initialize a maven directory structure (or use your favorite Maven archetype). For instance you can use the following command: “ Now let’s initiate the As you can see we’ll be using We’ll base our The init phase that we’ll write right after this will show how to use the Vert.x configuration to determine at runtime which node(s) in the Couchbase cluster we’ll bootstrap from. Instantiation of the Add the following init method to the The main entry point to the Couchbase API is the Let’s see how we can start our Verticle by first opening the bucket we’ll use throughout the lifetime of the Verticle. We want to keep a reference to it, and use the Notice first we get the name of the bucket (and the associated password if relevant) dynamically, from the Vert.x configuration. We open the bucket asynchronously, establishing the connections and resources internal to the SDK.<\/p>\n The Then we subscribe to our That’s a pretty good start!<\/p>\n When the Verticle stops, the resources created by the SDK should be properly disposed of. The Also since Once again we’ll stop the Verticle asynchronously, using a (we chose to shutdown RxJava upon completion of the SDK disconnection)<\/p>\n Note<\/strong> You can tune the SDK by passing in a If you didn’t create a specific environment the SDK will internally create one and shut it down properly, the result of which is seen above in the Let’s create a quick If you execute this, here is what you should see (notice the difference in timestamp format? How to verify the error behavior? We could simply change the password to one that is wrong, just to check the logs, which then look like:<\/p>\n So we successfully deployed (and stopped) our first Couchbase Verticle! \/! don’t forget to change the password back to the correct one<\/em><\/p><\/blockquote>\n Let’s try to do a little bit more with this Verticle. How about we try to prepare sample data in Couchbase and serve it in a REST endpoint managed by Vert.x?<\/p>\n We’ll create two sample documents in Couchbase during the starting up of the Verticle, users Alice and Bob.<\/p>\n One can store JSON in Couchbase using two Here are Alice and Bob, created using both alternatives:<\/p>\n and<\/p>\n Now, the start method needs a little bit of adjustment. Instead of saving the reference to the bucket in the subscription, we’ll move that earlier in a The use of Let’s modify our verticle so that it doesn’t stop right away, instead spinning a HTTP Server that will try to retrieve a json document from the database and send it to a client when the route Here is a quick and dirty way of using Vert.x’s Note<\/strong>: In a real application, Now let’s spin a HTTP Server on Verticle startup. Chain in the following code in the We need to create the Let’s test it: run the application and go to this url: https:\/\/localhost:8080\/users\/user1<\/a>. You should see Alice’s JSON, served directly from Couchbase!<\/p>\nVert.x<\/code>, a framework for writing fully asynchronous applications, knowing that the Couchbase Java SDK<\/code> fits well into this picture, being asynchronous from the ground up and exposing a RxJava<\/code>-based asynchronous API.<\/p>\nVerticle<\/code> that spins up a connection to a Couchbase Cluster<\/code> and Bucket<\/code> then serves JSON documents from the database, using Java 8.<\/p>\n\n
\n
\n
\n<\/li>\n<\/ul>\nStarting a new Vert.x project<\/h2>\n
mkdir -p cbvertx\/src\/main\/java\/com\/couchbase\/demo\/vertx\/<\/code>“.<\/p>\npom.xml<\/code> at the root of the project:<\/p>\n\r\n\r\n 4.0.0\r\n\r\n com.couchbase.demo\r\n cbvertx\r\n 1.0-SNAPSHOT\r\n\r\n \r\n \r\n io.vertx\r\n vertx-core\r\n 3.1.0\r\n \r\n \r\n io.vertx\r\n vertx-rx-java\r\n 3.1.0\r\n \r\n \r\n com.couchbase.client\r\n java-client\r\n 2.2.2\r\n \r\n \r\n \r\n io.reactivex\r\n rxjava\r\n 1.0.15\r\n \r\n \r\n log4j\r\n log4j\r\n 1.2.17\r\n \r\n \r\n\r\n \r\n \r\n \r\n maven-compiler-plugin\r\n 3.3\r\n \r\n 1.8\r\n 1.8\r\n \r\n \r\n \r\n \r\n\r\n\r\n<\/code><\/pre>\nVert.x<\/code> version 3.1.0<\/code> and its extension for bindings in RxJava, Couchbase Java SDK<\/code> version 2.2.2<\/code> and RxJava<\/code> version 1.0.15<\/code>…<\/p>\nSkeleton of the Verticle<\/h2>\n
CouchbaseVerticle<\/code> on the AbstractVerticle<\/code> in io.vertx.rxjava.core<\/code> (from the vertx-rx-java extension). Let’s create its skeleton in the project:<\/p>\npackage com.couchbase.demo.vertx;\r\n\r\nimport java.util.ArrayList;\r\nimport java.util.List;\r\n\r\nimport com.couchbase.client.java.CouchbaseAsyncCluster;\r\nimport io.vertx.core.Context;\r\nimport io.vertx.core.Vertx;\r\nimport io.vertx.core.json.JsonArray;\r\nimport io.vertx.core.json.JsonObject;\r\nimport io.vertx.core.logging.Logger;\r\nimport io.vertx.core.logging.LoggerFactory;\r\nimport io.vertx.rxjava.core.AbstractVerticle;\r\n\r\npublic class CouchbaseVerticle extends AbstractVerticle {\r\n\r\n private static final Logger LOGGER = LoggerFactory.getLogger(CouchbaseVerticle.class);\r\n\r\n private CouchbaseAsyncCluster cluster;\r\n\r\n}<\/code><\/pre>\nCouchbaseCluster<\/code> is still lightweight enough that it can be done like this during init.<\/p>\nCouchbaseVerticle<\/code>:<\/p>\n@Override\r\npublic void init(Vertx vertx, Context context) {\r\n super.init(vertx, context);\r\n \/\/getting the configuration JSON\r\n JsonObject config = context.config();\r\n\r\n \/\/getting the bootstrap node, as a JSON array (default to localhost)\r\n JsonArray seedNodeArray = config.getJsonArray(\"couchbase.seedNodes\", new JsonArray().add(\"localhost\"));\r\n \/\/convert to a List\r\n List seedNodes = new ArrayList<>(seedNodeArray.size());\r\n for (Object seedNode : seedNodeArray) {\r\n seedNodes.add((String) seedNode);\r\n }\r\n \/\/use that to bootstrap the Cluster\r\n this.cluster = CouchbaseAsyncCluster.create(seedNodes);\r\n}<\/code><\/pre>\nAsynchronously Obtaining a Bucket<\/h2>\n
Bucket<\/code> interface for the sync API, or AsyncBucket<\/code> for the async API. Establishing the connection to the bucket (“opening” it) is much more heavyweight, so it should be done asynchronously.<\/p>\nstart(Future startFuture)<\/code> method to asynchronously notify Vert.x that the Verticle is ready:<\/p>\n private volatile AsyncBucket bucket;\r\n\r\n @Override\r\n public void start(Future startFuture) throws Exception {\r\n cluster.openBucket(config().getString(\"couchbase.bucketName\", \"default\"), config().getString(\"couchbase.bucketPassword\", \"\"))\r\n .doOnNext(openedBucket -> LOGGER.info(\"Bucket opened \" + openedBucket.name()))\r\n .subscribe(\r\n openedBucket -> bucket = openedBucket,\r\n startFuture::fail,\r\n startFuture::complete);\r\n }<\/code><\/pre>\ndoOnNext<\/code> method is used to log the opening of the bucket.<\/p>\nObservable<\/code> stream and describe how we want to “consume” the final data:<\/p>\n\n
Future#fail<\/code> method.<\/li>\nFuture#complete<\/code> method.<\/li>\n<\/ul>\nGracefully tearing down the SDK<\/h2>\n
Cluster<\/code> object has a disconnect<\/code> method that does this, recursively calling close<\/code> on every Bucket<\/code> it opened (close can be used to dispose of a single Bucket).<\/p>\n1.0.15<\/code> RxJava has a method for shutting down its internal Threads: Schedulers.shutdown<\/code>. This should be invoked only when there won’t be subsequent usage of RxJava in the application though, so it might be a better idea to do that upon Vert.x shutdown…<\/p>\nFuture<\/code> to notify the framework of stop completion:<\/p>\n @Override\r\n public void stop(Future stopFuture) throws Exception {\r\n cluster.disconnect()\r\n .doOnNext(isDisconnectedCleanly -> LOGGER.info(\"Disconnected Cluster (cleaned threads: \" + isDisconnectedCleanly + \")\"))\r\n .subscribe(\r\n isDisconnectedCleanly -> stopFuture.complete(),\r\n stopFuture::fail,\r\n Schedulers::shutdown);\r\n }<\/code><\/pre>\nCouchbaseEnvironment<\/code> upon creation of the Cluster<\/code>. In that case, it is up to you to also call shutdown<\/code> on the environment when tearing down the whole SDK (that is, when all Clusters you used the environment in, generally just the one, are shut down).<\/p>\nisDisconnectedCleanly<\/code> variable.<\/p><\/blockquote>\nSeeing it in action<\/h2>\n
main<\/code> that embeds Vert.x, deploys the Verticle and then stops. Note this is a pretty naive implementation with CountDownLatches, where you would usually rather use the command line or Launcher<\/code> as a main class.<\/p>\n public static void main(String[] args) throws InterruptedException {\r\n Vertx vertx = Vertx.vertx();\r\n\r\n final CountDownLatch startLatch = new CountDownLatch(1);\r\n vertx.deployVerticle(new CouchbaseVerticle(), event -> {\r\n if (event.succeeded())\r\n LOGGER.info(\"Verticle Deployed - \" + event.result());\r\n else\r\n LOGGER.error(\"Verticle deployment error\", event.cause());\r\n startLatch.countDown();\r\n });\r\n startLatch.await();\r\n\r\n final CountDownLatch stopLatch = new CountDownLatch(1);\r\n vertx.close(event -> {\r\n if (event.succeeded())\r\n LOGGER.info(\"Vert.x Stopped - \" + event.result());\r\n else\r\n LOGGER.error(\"Vert.x stopping error\", event.cause());\r\n stopLatch.countDown();\r\n });\r\n stopLatch.await();\r\n }<\/code><\/pre>\n2015-12-11<\/code> ones are from the SDK while Dec 11, 2015<\/code> ones are from Vert.x):<\/p>\n2015-12-11 16:21:20 INFO Node:135 - Connected to Node localhost\r\n2015-12-11 16:21:20 INFO ConfigurationProvider:263 - Opened bucket default\r\nDec 11, 2015 4:21:20 PM com.couchbase.demo.vertx.CouchbaseVerticle\r\nINFO: Bucket opened default\r\nDec 11, 2015 4:21:20 PM com.couchbase.demo.vertx.CouchbaseVerticle\r\nINFO: Verticle Deployed - caf06dd3-c8d1-4b89-8de0-58f09467b805\r\n2015-12-11 16:21:20 INFO ConfigurationProvider:284 - Closed bucket default\r\n2015-12-11 16:21:20 INFO Node:145 - Disconnected from Node localhost\r\nDec 11, 2015 4:21:22 PM com.couchbase.demo.vertx.CouchbaseVerticle\r\nINFO: Disconnected Cluster (cleaned threads: true)\r\nDec 11, 2015 4:21:22 PM com.couchbase.demo.vertx.CouchbaseVerticle\r\nINFO: Vert.x Stopped - null<\/code><\/pre>\n2015-12-11 16:25:45 WARN Endpoint:283 - [null][KeyValueEndpoint]: Authentication Failure.\r\n2015-12-11 16:25:45 WARN ResponseStatusConverter:129 - Unknown ResponseStatus with Protocol HTTP: 401\r\n2015-12-11 16:25:45 WARN ResponseStatusConverter:129 - Unknown ResponseStatus with Protocol HTTP: 401\r\nDec 11, 2015 4:25:45 PM com.couchbase.demo.vertx.CouchbaseVerticle\r\nSEVERE: Verticle deployment error\r\ncom.couchbase.client.java.error.InvalidPasswordException: Passwords for bucket \"default\" do not match.\r\n at com.couchbase.client.java.CouchbaseAsyncCluster$1.call(CouchbaseAsyncCluster.java:210)\r\n at com.couchbase.client.java.CouchbaseAsyncCluster$1.call(CouchbaseAsyncCluster.java:200)\r\n at rx.internal.operators.OperatorOnErrorResumeNextViaFunction$1.onError(OperatorOnErrorResumeNextViaFunction.java:99)\r\n at rx.internal.operators.OperatorMap$1.onError(OperatorMap.java:48)\r\n at rx.observers.Subscribers$5.onError(Subscribers.java:229)\r\n at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.pollQueue(OperatorObserveOn.java:197)\r\n at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber$2.call(OperatorObserveOn.java:170)\r\n at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)\r\n at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)\r\n at java.util.concurrent.FutureTask.run(FutureTask.java:266)\r\n at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)\r\n at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)\r\n at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\r\n at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\r\n at java.lang.Thread.run(Thread.java:745)\r\n\r\nDec 11, 2015 4:25:45 PM com.couchbase.demo.vertx.CouchbaseVerticle\r\nINFO: Vert.x Stopped - null<\/code><\/pre>\n
\nHigh five!<\/strong><\/p>\nGoing further<\/h2>\n
Creating sample data in Couchbase on startup<\/h2>\n
Document<\/code> implementations:<\/p>\n\n
JsonDocument<\/code> is the default one. It is based on a simple JSON representation provided by the SDK, the JsonObject<\/code>.<\/li>\nRawJsonDocument<\/code> is useful when you already have JSON marshalling\/unmarshalling in your application (or another way of representing JSON like Vert.x’s own JsonObject<\/code>). In this implementation what you pass in is the raw JSON String representation.<\/li>\n<\/ul>\nJsonDocument.create(\"user1\", com.couchbase.client.java.document.json.JsonObject.create()\r\n .put(\"name\", \"Alice\").put(\"age\", 26))<\/code><\/pre>\nRawJsonDocument.create(\"user2\", new JsonObject().put(\"name\", \"Bob\").put(\"age\", 31).encode())<\/code><\/pre>\ndoOnNext<\/code>. After that, we’ll create the documents and make an Observable out of them using Observable.just<\/code>. This can be forwarded to the SDK for insertion using flatMap<\/code>:<\/p>\n@Override\r\n public void start(Future startFuture) throws Exception {\r\n cluster.openBucket(config().getString(\"couchbase.bucketName\", \"default\"), config().getString(\"couchbase.bucketPassword\", \"\"))\r\n .doOnNext(openedBucket -> LOGGER.info(\"Bucket opened \" + openedBucket.name()))\r\n .doOnNext(openedBucket -> bucket = openedBucket)\r\n .flatMap(nowIgnoreBucket -> Observable.just(\r\n JsonDocument.create(\"user1\", com.couchbase.client.java.document.json.JsonObject.create()\r\n .put(\"name\", \"Alice\").put(\"age\", 26)),\r\n RawJsonDocument.create(\"user2\", new JsonObject().put(\"name\", \"Bob\").put(\"age\", 31).encode())\r\n ))\r\n .flatMap(doc -> bucket.upsert(doc))\r\n .subscribe(Actions.empty(),\r\n startFuture::fail,\r\n startFuture::complete);\r\n }<\/code><\/pre>\nupsert<\/code> here guarantees that the documents will be created, whether the key already exist in database or not.<\/p>\nServing JSON data from Couchbase<\/h2>\n
user\/{id}<\/code> is used:<\/p>\nLauncher<\/code> to start the program (which will not stop the Vert.x core right away). Replace the content of our main<\/code> method with the following:<\/p>\nString[] vertxArgs = new String[] { \"run\", \"com.couchbase.demo.vertx.CouchbaseVerticle\" };\r\nLauncher.main(vertxArgs);<\/code><\/pre>\nLauncher<\/code> would usually be made the main class of the jar and you’d pass in command line arguments directly.<\/p><\/blockquote>\nstart<\/code> method, right after the flatMap(doc -> bucket.upsert(doc))<\/code> call:<\/p>\n.last()\r\n.flatMap(ignore -> vertx.createHttpServer()\r\n .requestHandler(this::handle)\r\n .listenObservable(8080))<\/code><\/pre>\nhandle<\/code> method to set up that route:<\/p>\n\r\n private void handle(HttpServerRequest r) {\r\n String[] path = r.path().split(\"\/\");\r\n if (path.length == 3 && \"users\".equals(path[1])) {\r\n bucket.get(path[2], RawJsonDocument.class)\r\n .switchIfEmpty(Observable.error(new DocumentDoesNotExistException(path[2])))\r\n .subscribe(doc -> r.response()\r\n .putHeader(\"content-type\", \"application\/json\")\r\n .end(doc.content()),\r\n error -> {\r\n r.response()\r\n .putHeader(\"content-type\", \"application\/json\")\r\n .setStatusCode(500).setStatusMessage(error.toString())\r\n .end(\"{\"error\": \"\" + error.toString() + \"\"}\");\r\n });\r\n }\r\n }<\/code><\/pre>\n{\r\n \"name\": \"Alice\",\r\n \"age\": 26\r\n}<\/code><\/pre>\n