Hazelcast and Web Sockets: Guest Post by Mark Addy of C2B2

This is a guest post by Mark Addy of C2B2 partners. Please find the original post here.

Hazelcast & Websockets

Over the past few years we have written many blogs demonstrating how Data Grid eventing frameworks can be hooked up to Websocket clients to deliver updates in real-time.  One Data Grid product we haven’t looked at previously is [Hazelcast][2] and recently this looks to be gaining lots of traction in the market place and also interest from our customers. Hazelcast has a developer friendly API and can be set up with the minimal amount of effort, it also boasts an almost identical set of features when running in client-server mode compared to the traditional embedded cache architecture. As we have discussed before the advantages in terms of de-coupling, independent tuning, scalability etc offered by client-server architectures is one of the more attractive features offered by Data Grids. We are going to setup and run a very basic Hazelcast Data Grid in client-server mode. A background thread will generate some updates to fictitious Stock Market data held in the grid and we’ll use Hazelcast’s event listener framework to publish these events to our clients.

JSR-356 is available as part of the EE7 release and as a result vendors are incorporating this into their compliant Application and Web Server distributions. Glassfish, Wildfly and Tomcat all come with implementations and this post is going to use Tomcat’s Web Socket capabilities for hosting the Web Application detailed in this post. JSR-356 Web Sockets have been available in Tomcat since the later releases of 7.0.x and are also present as standard in the new 8.0.x Web Container. In this post we’ll write a simple Web Application exposing a Server side Web Socket endpoint. The Web Application will register an event listener on our Hazelcast Data Grid and listen for events. Any Web Socket clients connecting to the application will receive push notification on these events and most importantly display these in an aesthetically pleasing graph.

Lastly JSR-356 also mandates the capability to define Web Socket Client endpoints as well as Server endpoints. This allows “any” client application to define and create a Web Socket connection to a Server side endpoint. We’ll take a look at this too and show how we can connect a legacy Swing application to our Web Socket Application running to Tomcat to receive the same push events from the Data Grid.

All the code for this prototype can be found here, use the following command to clone the repository:

git clone https://github.com/mark-addy/hazelcast-websocket-demo.git  

Build everything by executing the following command from the parent hazelcast-websocket-demo project root directory:

mvn clean install -DskipTests

Now we’ll step through everything.

hazelcast-shared

The hazelcast-shared project contains a Java class for the cached Stock records held in the data grid and a class to hold the results of a call to get all the Stock records currently held in the grid.  All the other projects depend on this one.

hazelcast-cluster

I’ve cheated just a little bit in setting up our Hazelcast cluster. Rather than running multiple JVM’s I’m creating the cluster in a single Java process, populating the cluster with some made up Stock Market data and then starting a background thread to randomly update prices of those stocks in the cluster. It’s all set up as a JUnit test and this will run forever unless you send some input to the console!

To programmatically create a Hazelcast cluster node you just need to use the following syntax.

Config config = new Config();  
 HazelcastInstance instance = Hazelcast.newHazelcastInstance(config);

By default the first cluster node will attempt to bind to all interfaces and listen on port 5701.  Subsequent nodes search starting at 5701 for the next available port.

The Hazelcast cluster needs to be running for the Web Application to start-up so once you’ve installed this project run the class StockMapEntryListenerTest as a JUnit Test.

hazelcast-web

The Web Application contains a ServletContextListener which instantiates a singleton instance holding a instance of a “HazelcastClient” which we will use to connect to the cluster created in the previous step. I’m doing this here so that the relatively expensive operation of creating a client connection to the server side cluster happens as part of the Web Application start up life-cycle.

@WebListener public class StockServletContextListener implements ServletContextListener {  
private static final Logger LOG = Logger.getLogger(StockServletContextListener.class.getName());  
@Override public void contextDestroyed(ServletContextEvent arg0) {  
LOG.info("Servlet Context Destroyed - Shutting down Hazelcast Client");  
ClientInstance.getInstance().getClient().shutdown();  
}  
@Override public void contextInitialized(ServletContextEvent servletContextEvent) {  
LOG.info("Servlet Context Initialized - Creating Hazelcast Client");  
ClientInstance.getInstance();  
}  
}

In the ClientInstance class you can see that the code to create a Hazelcast client using the vanilla settings is very straight forward. All we need to pass into the client configuration is the IP address of at least one active member of the server side cluster. Provided we have supplied a valid address the client will connect and also fail-over to another cluster member should the initial connection fail.

public class ClientInstance {  
      private final HazelcastInstance client;  
      private final IMap<String, StockRecord> stockMap;  
      private ClientInstance() {  
           ClientConfig clientConfig = new ClientConfig();  
           clientConfig.addAddress("127.0.0.1:5701");  
           client = HazelcastClient.newHazelcastClient(clientConfig);  
           stockMap = client.getMap("stock-map");  
      }  
      private static class ClientInstanceHolder {  
           private static final ClientInstance INSTANCE = new ClientInstance();  
      }  
      public static ClientInstance getInstance() {  
           return ClientInstanceHolder.INSTANCE;  
      }  
      public HazelcastInstance getClient() {  
           return client;  
      }  
      public IMap<String, StockRecord> getMap() {  
           return stockMap;  
      }  
    

}

Note that the server side Hazelcast cluster must be up and running before attempting to deploy / start the Web Application, if not then initialization of the singleton will fail and the Web Application won’t start.  This is because the default Hazelcast client implementation will try three times to connect to the cluster before failing.  You obviously have control over the number of attempts etc  if you want to make your client more resilient to the absence of any available server side nodes to connect to.

Lastly we have the Web Socket server side endpoint.  Tomcat has now put its proprietary Web Socket implementation into maintenance mode and replaced it with a JSR-356 implementation, here is our JSR-356 code:

@ServerEndpoint(value = "/websocket/stock")  
public class StockWebSocket implements EntryListener<String, StockRecord> {  
private static final Log LOG = LogFactory.getLog(StockWebSocket.class);  
private static final Set<StockWebSocket> connections = new CopyOnWriteArraySet<StockWebSocket>();  
private Session session;  
private String listenerId = null;  
public StockWebSocket() {  
listenerId = ClientInstance.getInstance().getMap().addEntryListener(this, true);  
}  
@OnOpen  
public void start(Session session) {  
this.session = session;  
connections.add(this);  
}  
@OnClose  
public void end() {  
if (listenerId != null) {  
ClientInstance.getInstance().getMap().removeEntryListener(listenerId);  
}  
connections.remove(this);  
}  
@OnMessage  
public void incoming(String message) {  
if (message.equals("open")) {  
Set<String> keys = ClientInstance.getInstance().getMap().keySet();  
StockResponse response = new StockResponse();  
response.setStocks(keys);  
send(this, ResponseSerializer.getInstance().serialize(response));  
}  
}  
@OnError  
public void onError(Throwable t) throws Throwable {  
}  
private static void send(StockWebSocket client, String message) {  
try {  
synchronized (client) {  
client.session.getBasicRemote().sendText(message);  
}  
} catch (IOException e) {  
LOG.debug("Failed to send message to client", e);  
connections.remove(client);  
try {  
client.session.close();  
} catch (IOException ioException) {  
}  
}  
}  
@Override  
public void entryAdded(EntryEvent<String, StockRecord> event) {  
}  
@Override  
public void entryRemoved(EntryEvent<String, StockRecord> event) {  
}  
@Override  
public void entryUpdated(EntryEvent<String, StockRecord> event) {  
send(this, ResponseSerializer.getInstance().serialize(event.getValue()));  
}  
@Override  
public void entryEvicted(EntryEvent<String, StockRecord> event) {  
}  
}

The Web Socket implements com.hazelcast.core.EntryListener and therefore has access to the associated event callbacks.

• When a client connects to the Web Socket we retrieve the singleton Hazelcast client instance and register ourself as a listener for all events in the “stock-map” cache.
• Clients send a simple “open” text message when they connect.  The Web Socket responds with an instance of StockResponse containing a Collection of all the Stock symbols currently in the Data Grid.  Ok, so in the real world we wouldn’t really ask the Web Socket for a complete list of all the records in the Data Grid!
• When the entryUpdated callback occurs, we serialize the updated cache value (StockRecord) to JSON and send this to the associated client.
• When the client disconnects the Web Socket we retrieve the singleton Hazelcast client instance and remove ourself from the listeners.

String addEntryListener(EntryListener<K,V> listener,  
             boolean includeValue)  
 String addEntryListener(EntryListener<K,V> listener,  
             K key,  
             boolean includeValue)  
 String addEntryListener(EntryListener<K,V> listener,  
             Predicate<K,V> predicate,  
             boolean includeValue)  
 String addEntryListener(EntryListener<K,V> listener,  
             Predicate<K,V> predicate,  
             K key,  
             boolean includeValue)

To run the Web Application you’ll need an instance of Tomcat, I used Tomcat 8.0.1 and you can get a copy from here.  Once you’ve installed Tomcat, ensure that your hazelcast cluster is up and running before attempting to deploy the Web Application.

• Build the Web Application (you should have done this right at the beginning)
• Copy the build artifact into the $TOMCAT_HOME/webapps directory.
• Start Tomcat (bin/startup.sh)

You should be able to access the Web Application at this URL:

http://localhost:8080/hazelcast-web/stockticker.jsp

Your browser will make a Web Socket connection to the Tomcat instance and send an “open” text message to retrieve all the Stock keys in the Data Grid.  The Java script in the browser then creates a chart “series” for each Stock and wait for updates to be pushed.  The screen should hopefully look similar to this after a few minutes:

We’re using HighCharts for the cool graphs.

hazelcast-client

Java’s Swing framework might not be as cool as HTML5 & Web Sockets but there are still a vast number of “legacy” Swing applications out in the real-world and we come across them regularly.

As already discussed JSR-356 permits Client Web Socket endpoints and Tomcat provides support for standalone clients to make Web Socket connections to server side endpoints.  This last project demonstrates this capability and allows us to write a Swing client that connects to the Tomcat Web Socket we have just created and render Stock price changes as they occur in a JFreeChart.

Traditionally Swing clients might use RMI or JMS to retrieve data from the server side but these methods can be problematic in environments with firewall restrictions so using Web Sockets might have some potential in certain use-cases.

Firstly we need to set up some dependencies for our client to make a Web Socket connection.  Below are the required libraries, taken from this projects POM:

<dependency>  
                <groupId>org.apache.tomcat</groupId>  
                <artifactId>tomcat-websocket-api</artifactId>  
                <version>8.0.1</version>  
           </dependency>  
           <dependency>  
                <groupId>org.apache.tomcat</groupId>  
                <artifactId>tomcat-websocket</artifactId>  
                <version>8.0.1</version>  
           </dependency>  
           <dependency>  
                <groupId>org.apache.tomcat</groupId>  
                <artifactId>tomcat-juli</artifactId>  
                <version>8.0.1</version>  
           </dependency>  
           <dependency>  
                <groupId>org.apache.tomcat</groupId>  
                <artifactId>tomcat-util</artifactId>  
                <version>8.0.1</version>  
           </dependency>  
           <dependency>  
                <groupId>org.apache.tomcat</groupId>  
                <artifactId>tomcat-coyote</artifactId>  
                <version>8.0.1</version>  
           </dependency>

 @ClientEndpoint  
 public class StockWebSocketClient {  
      private StockClient demo;  
      private Session session;  
      public StockWebSocketClient(StockClient demo) throws DeploymentException, IOException {  
           this.demo = demo;  
           WebSocketContainer container = ContainerProvider.getWebSocketContainer();  
           String uri = "ws://localhost:8080/hazelcast-web/websocket/stock";  
           container.connectToServer(this, URI.create(uri));  
      }  
      @OnOpen  
      public void onOpen(Session session) {  
           System.out.println("opened");  
           this.session = session;  
           send("open");  
      }  
      @OnError  
      public void onError(Throwable t) {  
           System.out.println("error " + t.getMessage());  
           t.printStackTrace();  
      }  
      @OnClose  
      public void onClose() {  
           System.out.println("close");  
      }  
      @OnMessage  
      public void onMessage(String message) {  
           System.out.println("message : " + message);  
           Object messageObject = ResponseSerializer.getInstance().deserialize(message);  
           if (messageObject instanceof StockRecord) {  
                demo.updatePrice((StockRecord)messageObject);  
           } else if (messageObject instanceof StockResponse) {  
                System.out.println("Stock Response received : " + Arrays.deepToString(((StockResponse)messageObject).getStocks().toArray()));  
                demo.renderGraph((StockResponse)messageObject);  
           }  
      }  
      private void send(String message) {  
           try {  
                synchronized (this) {  
                     session.getBasicRemote().sendText(message);  
                }  
           } catch (IOException e) {  
                try {  
                     session.close();  
                } catch (IOException ioException) {  
                }  
           }  
      }  
 }