{"id":13061,"date":"2022-04-12T10:11:38","date_gmt":"2022-04-12T17:11:38","guid":{"rendered":"https:\/\/www.couchbase.com\/blog\/?p=13061"},"modified":"2024-05-08T07:18:22","modified_gmt":"2024-05-08T14:18:22","slug":"event-broker-integration-with-couchbase-solace-java","status":"publish","type":"post","link":"https:\/\/www.couchbase.com\/blog\/event-broker-integration-with-couchbase-solace-java\/","title":{"rendered":"Event Broker Integration with Couchbase, Solace & Java"},"content":{"rendered":"

In the previous <\/span>blog post<\/span><\/a> we discussed how Couchbase can seamlessly integrate into an <\/span>Event-Driven Architecture<\/span><\/i>. Using the Couchbase Eventing Service, document mutations can be published to a <\/span>Solace PubSub+<\/span><\/a> queue from where the data is made available to subscribing microservices in near real-time.<\/span><\/p>\n

This post focuses on how microservices as Topic subscribers can leverage Couchbase as a scalable & resilient datastore.<\/span><\/p>\n

Architecture overview & sample application<\/b><\/p>\n

In this sample application<\/a>, we subscribe to a Solace PubSub+ Topic, process the data, and update the document in Couchbase.<\/span><\/p>\n

In our example application, we store hotel details in Couchbase. Updates to the hotel configuration are published to a Solace PubSub+ topic. We build an application that subscribes to the Topic, retrieves the updated data and updates the existing hotel configuration in Couchbase.<\/span><\/p>\n

We use the <\/span>travel-sample<\/span><\/i> bucket that contains sample data such as <\/span>flights<\/span><\/i>, <\/span>routes<\/span><\/i> and others. In our example, we focus on the <\/span>hotel<\/span><\/i> documents stored in the <\/span>inventory<\/span><\/i> scope in the <\/span>hotel<\/span><\/i> collection. All hotel documents consist of a JSON document with various attributes.<\/span><\/p>\n

We will update only a small subset of attributes for this tutorial, such as <\/span>pets_allowed<\/span><\/i> and <\/span>free_internet<\/span><\/i>.<\/span><\/p>\n

\"Event-driven

The sample application subscribes to a Solace Topic and updates hotel data in Couchbase. Other applications may provide the hotel configuration changes. We use a simple service that publishes updates to the Topic.<\/p><\/div>\n

Caption: The sample application subscribes to a Solace Topic and updates hotel data in Couchbase. Other applications may provide the hotel configuration changes. We use a simple service that publishes updates to the Topic.<\/span><\/i><\/p>\n

Prerequisites<\/span><\/h2>\n

Please review the prerequisites for this sample application:<\/span><\/p>\n

Couchbase Capella<\/b><\/p>\n

Create a free <\/span>Couchbase Capella<\/span><\/a> trial account and follow the instructions to provision your trial cluster. This only takes a few minutes and gives you a fully-fledged Couchbase cluster, including the <\/span>Data<\/span><\/i> service and the sample dataset, <\/span>travel-sample,<\/span><\/i> that we will use throughout this blog post.<\/span><\/p>\n

You can use this tutorial when not using our managed Couchbase Capella offering. The <\/span>travel-sample<\/span><\/i> bucket is part of the Couchbase server product and can be imported using the Couchbase console.<\/span><\/p>\n

Solace PubSub+ Event Broker Cloud<\/b><\/p>\n

Sign up for a free <\/span>Solace cloud trial<\/span><\/a> and create a Service\/VPN. We will use the VPN to connect to a Solace Topic.<\/span><\/p>\n

 <\/p>\n

Code review – Solace subscriber & publisher<\/span><\/h2>\n

To subscribe to messages from the Solace Topic we use the Solace Java API. This example implementation is derived from examples in the Solace documentation. For implementation details consult the Solace documentation.<\/span><\/p>\n

final JCSMPProperties properties = new JCSMPProperties();\r\nproperties.setProperty(JCSMPProperties.HOST, \"tcps:\/\/yourhost.messaging.solace.cloud:55443\");\r\nproperties.setProperty(JCSMPProperties.USERNAME, \"solace-cloud-client\");\r\nproperties.setProperty(JCSMPProperties.PASSWORD, \"password\");\r\nproperties.setProperty(JCSMPProperties.VPN_NAME, \"couchbasedemo\");\r\nJCSMPSession session = JCSMPFactory.onlyInstance().createSession(properties);\r\nsession.connect();\r\n\r\nfinal Topic topic = JCSMPFactory.onlyInstance().createTopic(\"tutorial\/topic\");<\/pre>\n
We set up the <\/span>JCSMPProperties<\/span><\/i> with the relevant access information available in the Solace web interface. We then create a <\/span>JCSMPSession<\/span><\/i> and a <\/span>Topic<\/span><\/i>.<\/span><\/p>\n
To subscribe to the messages on the Topic we implement an <\/span>XMLMessageConsumer<\/span><\/i>:<\/span><\/p>\n
\/\/ ### Message Receiver - Subscribe to Topic ###\r\nfinal XMLMessageConsumer cons = session.getMessageConsumer(new XMLMessageListener() {\r\n  @Override\r\n  public void onReceive(BytesXMLMessage msg) {\r\n    if (msg instanceof TextMessage) {\r\n      System.out.printf(\"SUBSCRIBE: Message received: '%s'%n\", ((TextMessage) msg).getText());\r\n      UpdateHotelData.getInstance().upsertHotelData(((TextMessage) msg).getText());\r\n    } else {\r\n      System.out.println(\"Message received.\");\r\n    }\r\n  }\r\n\r\n  @Override\r\n  public void onException(JCSMPException e) {\r\n    System.out.printf(\"Consumer received exception: %s%n\", e);\r\n  }\r\n});\r\nsession.addSubscription(topic);\r\nSystem.out.println(\"SUBSCRIBE: Connected. Awaiting message...\");\r\ncons.start();<\/pre>\n
In the onReceive<\/i> method, we take the message, convert it to a string and then call the upsertHotelData<\/i> method of UpdateHotelData <\/i>class that contains the Couchbase implementation discussed a bit later in this blog post.<\/p>\n
We add the topic subscription to the session and start the message consumer.<\/span><\/p>\n
In order to be able to test the implementation, we also create a message publisher. For this, we first implement an <\/span>XMLMessageProducer<\/span><\/i>.<\/span><\/p>\n
\/\/ ### Message Producer - Send message to Topic\r\nXMLMessageProducer prod = session.getMessageProducer(new JCSMPStreamingPublishCorrelatingEventHandler() {\r\n  @Override\r\n  public void responseReceivedEx(Object key) {\r\n    System.out.println(\"Producer received response for msg: \" + key.toString());\r\n  }\r\n\r\n  @Override\r\n  public void handleErrorEx(Object key, JCSMPException cause, long timestamp) {\r\n    System.out.printf(\"Producer received error for msg: %s@%s - %s%n\", key.toString(), timestamp, cause);\r\n  }\r\n});<\/pre>\n
With the message producer in place, we can now send messages as seen below.\u00a0<\/span><\/p>\n
\/\/ Send messages. Here we loop through and send the same message multiple times.\r\nfor (int msgsSent = 0; msgsSent < 2; ++msgsSent) {\r\n  TextMessage msg = JCSMPFactory.onlyInstance().createMessage(TextMessage.class);\r\n  msg.setText(data);\r\n  System.out.printf(\"PUBLISH: Sending message '%s' to topic '%s'...%n\", data, topic.getName());\r\n  prod.send(msg, topic);\r\n  \/\/Gives us some time to follow the console logs\r\n  Thread.sleep(3000);\r\n}<\/pre>\n
Upserting the data to Couchbase<\/span><\/h3>\n
With the logic to subscribe and publish messages in place, let\u2019s now focus on how to update the hotel documents in Couchbase.<\/span><\/p>\n
For this example, we assume that the message retrieved from the Topic has the JSON format below. Including a <\/span>hotel_id<\/span><\/i> that we will use as the document id of the hotel documents stored in Couchbase, as well as a number of attributes that we can change:<\/span><\/p>\n
{\r\n\u00a0\u00a0\"hotel_id\": \"hotel_10025\",\r\n\u00a0\u00a0\"pets_ok\": true,\r\n\u00a0\u00a0\"free_breakfast\": true,\r\n\u00a0\u00a0\"free_internet\": true,\r\n\u00a0\u00a0\"free_parking\": true\r\n}<\/pre>\n
To establish a secure connection from our application to Couchbase Capella, we need to:<\/span><\/p>\n