implemented websocket server and client support

Author: Juraj Veverka

hazelcast-cluster/it-tests/build.gradle

@@ -20,6 +20,9 @@ targetCompatibility = 10
 dependencies {
    compile 'org.slf4j:slf4j-api:1.8.0-beta2'
    compile 'org.slf4j:slf4j-simple:1.8.0-beta2'
+
+   compile 'org.eclipse.jetty.websocket:websocket-client:9.4.11.v20180605'
+
    testCompile 'org.testng:testng:6.14.3'
 }
 

hazelcast-cluster/it-tests/src/main/java/itx/hazelcast/cluster/client/ClientApp.java

@@ -0,0 +1,46 @@
+package itx.hazelcast.cluster.client;
+
+import itx.hazelcast.cluster.client.wsclient.SessionListener;
+import itx.hazelcast.cluster.client.wsclient.WsClient;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class MainClient {
+
+    public static void main(String[] args) throws Exception {
+
+        CountDownLatch cl = new CountDownLatch(1);
+        SessionListener sessionListener = new SessionListener() {
+
+            @Override
+            public void onSessionReady() {
+                cl.countDown();
+            }
+
+            @Override
+            public void onTextMessage(String message) {
+
+            }
+
+            @Override
+            public void onByteMessage(byte[] message) {
+
+            }
+
+            @Override
+            public void onSessionClose() {
+
+            }
+
+        };
+
+        WsClient clientApp = new WsClient("ws://localhost:8080/data/websocket", sessionListener);
+        clientApp.start();
+        cl.await(10, TimeUnit.SECONDS);
+        clientApp.sendTextMessage("hello world");
+        clientApp.sendByteMessage("hi there".getBytes());
+        clientApp.stop();
+    }
+
+}

hazelcast-cluster/it-tests/src/main/java/itx/hazelcast/cluster/client/MainClient.java

@@ -0,0 +1,13 @@
+package itx.hazelcast.cluster.client.wsclient;
+
+public interface SessionListener {
+
+    void onSessionReady();
+
+    void onTextMessage(String message);
+
+    void onByteMessage(byte[] message);
+
+    void onSessionClose();
+
+}

hazelcast-cluster/it-tests/src/main/java/itx/hazelcast/cluster/client/wsclient/SessionListener.java

@@ -0,0 +1,87 @@
+package itx.hazelcast.cluster.client.wsclient;
+
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.StatusCode;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
+import org.eclipse.jetty.websocket.api.annotations.WebSocket;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+@WebSocket(maxTextMessageSize = 64 * 1024)
+public class SimpleWebSocket {
+
+    final private static Logger LOG = LoggerFactory.getLogger(SimpleWebSocket.class);
+
+    private final CountDownLatch openLatch;
+    private final CountDownLatch closeLatch;
+    private Session session;
+    private SessionListener sessionListener;
+
+    public SimpleWebSocket(SessionListener sessionListener) {
+        this.openLatch = new CountDownLatch(1);
+        this.closeLatch = new CountDownLatch(1);
+        this.sessionListener = sessionListener;
+    }
+
+    public boolean awaitClose(int duration, TimeUnit unit) throws InterruptedException {
+        return this.closeLatch.await(duration,unit);
+    }
+
+    public boolean awaitOpen(int duration, TimeUnit unit) throws InterruptedException {
+        return this.openLatch.await(duration,unit);
+    }
+
+    @OnWebSocketClose
+    public void onClose(int statusCode, String reason) {
+        LOG.info("Connection closed: {} - {}", statusCode, reason);
+        this.session = null;
+        this.closeLatch.countDown(); // trigger latch
+        this.sessionListener.onSessionClose();
+    }
+
+    @OnWebSocketConnect
+    public void onConnect(Session session) {
+        LOG.info("Got connect: {}",session);
+        this.session = session;
+        this.sessionListener.onSessionReady();
+        this.openLatch.countDown();
+    }
+
+    @OnWebSocketMessage
+    public void onMessage(String msg) {
+        LOG.info("Got msg: {}",msg);
+        this.sessionListener.onTextMessage(msg);
+    }
+
+    @OnWebSocketMessage
+    public void onMessage(InputStream stream) {
+        try {
+            this.sessionListener.onByteMessage(stream.readAllBytes());
+        } catch (IOException e) {
+            LOG.error("Error: ", e);
+        }
+    }
+
+    public void close() {
+        if (session != null) {
+            this.session.close(StatusCode.NORMAL, "Closing");
+        }
+    }
+
+    public void sendTextMessage(String message) throws IOException {
+        session.getRemote().sendString(message);
+    }
+
+    public void sendByteMessage(byte[] message) throws IOException {
+        session.getRemote().sendBytes(ByteBuffer.wrap(message));
+    }
+
+}

hazelcast-cluster/it-tests/src/main/java/itx/hazelcast/cluster/client/wsclient/SimpleWebSocket.java

@@ -0,0 +1,46 @@
+package itx.hazelcast.cluster.client.wsclient;
+
+import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
+import org.eclipse.jetty.websocket.client.WebSocketClient;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.concurrent.TimeUnit;
+
+public class WsClient {
+
+    private WebSocketClient client;
+    private String destUri;
+    private SessionListener sessionListener;
+    private SimpleWebSocket socket;
+
+    public WsClient(String destUri, SessionListener sessionListener) {
+        this.destUri = destUri;
+        this.sessionListener = sessionListener;
+    }
+
+    public void start() throws Exception {
+        client = new WebSocketClient();
+        client.start();
+        URI echoUri = new URI(destUri);
+        ClientUpgradeRequest request = new ClientUpgradeRequest();
+        socket = new SimpleWebSocket(sessionListener);
+        client.connect(socket, echoUri, request);
+        socket.awaitOpen(10, TimeUnit.SECONDS);
+    }
+
+    public void sendTextMessage(String message) throws IOException {
+        socket.sendTextMessage(message);
+    }
+
+    public void sendByteMessage(byte[] message) throws IOException {
+        socket.sendByteMessage(message);
+    }
+
+    public void stop() throws Exception {
+        socket.close();
+        socket.awaitClose(10, TimeUnit.SECONDS);
+        client.stop();
+    }
+
+}

hazelcast-cluster/it-tests/src/main/java/itx/hazelcast/cluster/client/wsclient/WsClient.java

@@ -20,6 +20,10 @@ dependencies {
    compile "org.eclipse.jetty:jetty-server:${jettyVersion}"
    compile "org.eclipse.jetty:jetty-servlet:${jettyVersion}"
 
+   compile "org.eclipse.jetty.websocket:websocket-api:${jettyVersion}"
+   compile "org.eclipse.jetty.websocket:websocket-server:${jettyVersion}"
+   compile "org.eclipse.jetty.websocket:websocket-client:${jettyVersion}"
+
    compile "com.fasterxml.jackson.core:jackson-databind:2.9.6"
    compile "javax.xml.bind:jaxb-api:2.3.0"
 

hazelcast-cluster/server-node/build.gradle

@@ -14,6 +14,7 @@
 import itx.hazelcast.cluster.server.rest.RestApplication;
 import itx.hazelcast.cluster.server.services.DataService;
 import itx.hazelcast.cluster.server.services.DataServiceImpl;
+import itx.hazelcast.cluster.server.websocket.WsServlet;
 import org.eclipse.jetty.server.HttpConfiguration;
 import org.eclipse.jetty.server.HttpConnectionFactory;
 import org.eclipse.jetty.server.Server;
@@ -45,22 +46,23 @@ public ServerApp() {
     public void startServer() throws Exception {
         LOG.info("starting hazelcast ...");
         executorService = Executors.newSingleThreadExecutor();
-
+        // Register serializers in hazelcast configuration
         SerializerConfig sc = new SerializerConfig()
                 .setImplementation(new InstanceInfoSerializer())
                 .setTypeClass(InstanceInfo.class);
         Config config = new Config();
         config.getSerializationConfig().addSerializerConfig(sc);
-
+        // Create hazelcast instance
         hazelcastInstance = Hazelcast.newHazelcastInstance(config);
         MembershipListener membershipListener = new MembershipListenerImpl();
+        // listen on cluster events
         Cluster cluster = hazelcastInstance.getCluster();
         cluster.addMembershipListener(membershipListener);
-
+        // populate clusterInfo map
         Map<String, InstanceInfo> clusterInfo = hazelcastInstance.getMap( "clusterInfo" );
         InstanceInfo instanceInfo = createInstanceInfo(cluster.getLocalMember(), clusterInfo);
         clusterInfo.put(instanceInfo.getId(), instanceInfo);
-
+        // register leadership listener
         GateKeepingListener gateKeepingListener = new GateKeepingListenerImpl();
         GateKeeperRunnable gateKeeperRunnable = new GateKeeperRunnable(executorService, hazelcastInstance, gateKeepingListener);
         executorService.submit(gateKeeperRunnable);
@@ -76,13 +78,15 @@ public void startServer() throws Exception {
         ServletContainer restServletContainer = new ServletContainer(resourceConfig);
         ServletHolder restServletHolder = new ServletHolder(restServletContainer);
         context.addServlet(restServletHolder, "/rest");
-
+        // Register websocket handlers
+        ServletHolder webSocketHolder = new ServletHolder(new WsServlet());
+        context.addServlet(webSocketHolder, "/websocket");
+        // Setup http connectors
         HttpConfiguration httpConfig = new HttpConfiguration();
         HttpConnectionFactory httpConnectionFactory = new HttpConnectionFactory(httpConfig);
         ServerConnector http = new ServerConnector(server, httpConnectionFactory);
         http.setPort(instanceInfo.getWebServerPort());
         server.addConnector(http);
-
         server.start();
         LOG.info("init sequence done.");
     }
@@ -93,20 +97,22 @@ public void stopServer() throws Exception {
             server.stop();
         }
         if (hazelcastInstance != null) {
-            ISemaphore semaphore = hazelcastInstance.getSemaphore("gatekeeper");
-            semaphore.release();
+            //ISemaphore semaphore = hazelcastInstance.getSemaphore("gatekeeper");
+            //semaphore.release();
             //hazelcastInstance.shutdown();
         }
         executorService.shutdown();
     }
 
     private InstanceInfo createInstanceInfo(Member member, Map<String, InstanceInfo> clusterinfo) throws UnknownHostException {
         LOG.info("clusterinfo size {}", clusterinfo.size());
-        int webServerPort = 8080;
+        int webServerPort = 8080; //let's start with 8080 and check if this port is free
         InetSocketAddress inetSocketAddress = member.getAddress().getInetSocketAddress();
         for (InstanceInfo i: clusterinfo.values()) {
             if (inetSocketAddress.getHostName().equals(i.getAddress().getHostName())) {
-                webServerPort++;
+                if (webServerPort == i.getWebServerPort()) {
+                    webServerPort = i.getWebServerPort() + 1;
+                }
             }
         }
 

hazelcast-cluster/server-node/src/main/java/itx/hazelcast/cluster/server/hazelcast/ServerApp.java

@@ -0,0 +1,39 @@
+package itx.hazelcast.cluster.server.websocket;
+
+import org.eclipse.jetty.websocket.servlet.ServletUpgradeRequest;
+import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
+import org.eclipse.jetty.websocket.servlet.WebSocketCreator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class WebSocketCreatorImpl implements WebSocketCreator, WebSocketUnregisterService {
+
+    private static final Logger LOG = LoggerFactory.getLogger(WebSocketCreatorImpl.class);
+
+    private final Map<Long, WsEndpoint> wsEndPoints;
+    private final AtomicLong atomicLong;
+
+    public WebSocketCreatorImpl() {
+        this.wsEndPoints = new ConcurrentHashMap<>();
+        this.atomicLong = new AtomicLong(0);
+    }
+
+    @Override
+    public Object createWebSocket(ServletUpgradeRequest req, ServletUpgradeResponse resp) {
+        LOG.info("createWebSocket ...");
+        Long id = atomicLong.getAndIncrement();
+        WsEndpoint wsEndpoint = new WsEndpoint(id, this);
+        wsEndPoints.put(id, wsEndpoint);
+        return wsEndpoint;
+    }
+
+    @Override
+    public void unregister(Long id) {
+        wsEndPoints.remove(id);
+    }
+
+}

hazelcast-cluster/server-node/src/main/java/itx/hazelcast/cluster/server/websocket/WebSocketCreatorImpl.java

@@ -0,0 +1,7 @@
+package itx.hazelcast.cluster.server.websocket;
+
+public interface WebSocketUnregisterService {
+
+    void unregister(Long id);
+
+}