implemented integration test using websocket client

Author: Juraj Veverka

hazelcast-cluster/README.md

@@ -22,4 +22,17 @@ First server-node started will use port 8080, second server-node in cluster will
 third will use 8082 an so on.
 
 ## Run integration tests
-TODO
+In order to run integration tests, JDK 11 and tmux has to be installed.
+1. Build project and distribution.
+   ```
+   gradle clean build installDist
+   ```
+2. Start 3-node cluster in new terminal using script below.
+   Wait till cluster is formed.
+   ```
+   ./start-3-node-cluster.sh
+   ```    
+3. While cluster is running, run integration tests in new terminal window.
+   ```
+   gradle clean test -Dtest.profile=integration
+   ```

hazelcast-cluster/common/build.gradle

@@ -21,7 +21,7 @@ repositories {
 sourceCompatibility = 11
 targetCompatibility = 11
 
-def grpcVersion = '1.13.2'
+def grpcVersion = '1.14.0'
 mainClassName = 'itx.examples.grpc.service.Main'
 
 dependencies {
@@ -42,7 +42,7 @@ dependencies {
 
 protobuf {
     protoc {
-        artifact = 'com.google.protobuf:protoc:3.3.0'
+        artifact = 'com.google.protobuf:protoc:3.6.1'
     }
     plugins {
         grpc {

hazelcast-cluster/common/src/main/proto/messages.proto

@@ -14,3 +14,44 @@ message Address {
     string hostName = 1;
     int32 port = 2;
 }
+
+message SubscribeToTopicRequest {
+    string topicId = 1;
+    int64 requestId = 2;
+}
+
+message SubscribeToTopicResponse {
+    int64 requestId = 2;
+    string subscriptionId = 3;
+}
+
+message UnsubscribeFromTopicRequest {
+    int64 requestId = 1;
+    string subscriptionId = 2;
+}
+
+message UnsubscribeFromTopicResponse {
+    int64 requestId = 1;
+}
+
+message DataMessage {
+    string topicId = 1;
+    string message = 2;
+}
+
+message DataMessageEvent {
+    string topicId = 1;
+    string message = 2;
+    string subscriptionId = 3;
+}
+
+message MessageWrapper {
+    oneof payload {
+        SubscribeToTopicRequest subscribeToTopicRequest = 1;
+        SubscribeToTopicResponse subscribeToTopicResponse = 2;
+        UnsubscribeFromTopicRequest unsubscribeFromTopicRequest = 3;
+        UnsubscribeFromTopicResponse unsubscribeFromTopicResponse = 4;
+        DataMessage dataMessage = 5;
+        DataMessageEvent dataMessageEvent = 6;
+    }
+}

hazelcast-cluster/it-tests/build.gradle

@@ -12,6 +12,10 @@ test {
    testLogging {
       events "passed", "skipped", "failed"
    }
+
+   if (System.properties['test.profile'] != 'integration') {
+      exclude '**/*ITTest*'
+   }
 }
 
 sourceCompatibility = 10

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

@@ -1,12 +1,14 @@
 package itx.hazelcast.cluster.client.wsclient;
 
+import itx.hazelcast.cluster.dto.MessageWrapper;
+
 public interface SessionListener {
 
     void onSessionReady();
 
     void onTextMessage(String message);
 
-    void onByteMessage(byte[] message);
+    void onMessageWrapper(MessageWrapper messageWrapper);
 
     void onSessionClose();
 

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

@@ -1,5 +1,6 @@
 package itx.hazelcast.cluster.client.wsclient;
 
+import itx.hazelcast.cluster.dto.MessageWrapper;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.StatusCode;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
@@ -49,10 +50,10 @@ public void onClose(int statusCode, String reason) {
 
     @OnWebSocketConnect
     public void onConnect(Session session) {
-        LOG.info("Got connect: {}",session);
+        LOG.info("onConnect");
         this.session = session;
-        this.sessionListener.onSessionReady();
         this.openLatch.countDown();
+        this.sessionListener.onSessionReady();
     }
 
     @OnWebSocketMessage
@@ -62,9 +63,10 @@ public void onMessage(String msg) {
     }
 
     @OnWebSocketMessage
-    public void onMessage(InputStream stream) {
+    public void onMessage(InputStream inputStream) {
         try {
-            this.sessionListener.onByteMessage(stream.readAllBytes());
+            MessageWrapper messageWrapper = MessageWrapper.parseFrom(inputStream.readAllBytes());
+            this.sessionListener.onMessageWrapper(messageWrapper);
         } catch (IOException e) {
             LOG.error("Error: ", e);
         }

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

@@ -1,5 +1,6 @@
 package itx.hazelcast.cluster.client.wsclient;
 
+import itx.hazelcast.cluster.dto.MessageWrapper;
 import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
 import org.eclipse.jetty.websocket.client.WebSocketClient;
 
@@ -33,8 +34,8 @@ public void sendTextMessage(String message) throws IOException {
         socket.sendTextMessage(message);
     }
 
-    public void sendByteMessage(byte[] message) throws IOException {
-        socket.sendByteMessage(message);
+    public void sendMessageWrapper(MessageWrapper messageWrapper) throws IOException {
+        socket.sendByteMessage(messageWrapper.toByteArray());
     }
 
     public void stop() throws Exception {

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

@@ -0,0 +1,57 @@
+package itx.hazelcast.cluster.client.tests;
+
+import itx.hazelcast.cluster.client.wsclient.SessionListener;
+import itx.hazelcast.cluster.dto.MessageWrapper;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class BlockingSessionListener implements SessionListener {
+
+    private final CountDownLatch clStart;
+    private CountDownLatch clMessageWrapper;
+    private final List<MessageWrapper> messageBuffer;
+
+    public BlockingSessionListener() {
+        this.clStart = new CountDownLatch(1);
+        this.messageBuffer = new ArrayList<>();
+    }
+
+    @Override
+    public void onSessionReady() {
+        clStart.countDown();
+    }
+
+    public void awaitForSessionReady(long time, TimeUnit timeUnit) throws InterruptedException {
+        clStart.await(time, timeUnit);
+    }
+
+    public void resetMessageBuffer(int expectedMessageCount) {
+        clMessageWrapper = new CountDownLatch(expectedMessageCount);
+        messageBuffer.clear();
+    }
+
+    public List<MessageWrapper> awaitMessages(long time, TimeUnit timeUnit) throws InterruptedException {
+        clMessageWrapper.await(time, timeUnit);
+        return Collections.unmodifiableList(messageBuffer);
+    }
+
+    @Override
+    public void onTextMessage(String message) {
+    }
+
+    @Override
+    public void onMessageWrapper(MessageWrapper messageWrapper) {
+        messageBuffer.add(messageWrapper);
+        clMessageWrapper.countDown();
+    }
+
+    @Override
+    public void onSessionClose() {
+
+    }
+
+}

hazelcast-cluster/it-tests/src/test/java/itx/hazelcast/cluster/client/tests/BlockingSessionListener.java

@@ -0,0 +1,84 @@
+package itx.hazelcast.cluster.client.tests;
+
+import itx.hazelcast.cluster.client.wsclient.WsClient;
+import itx.hazelcast.cluster.dto.DataMessage;
+import itx.hazelcast.cluster.dto.DataMessageEvent;
+import itx.hazelcast.cluster.dto.MessageWrapper;
+import itx.hazelcast.cluster.dto.SubscribeToTopicRequest;
+import itx.hazelcast.cluster.dto.SubscribeToTopicResponse;
+import itx.hazelcast.cluster.dto.UnsubscribeFromTopicRequest;
+import itx.hazelcast.cluster.dto.UnsubscribeFromTopicResponse;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class SingleNodeITTest {
+
+    @Test
+    public void testSingleNodeClientMessage() throws Exception {
+        String topic = "topic1";
+        BlockingSessionListener sessionListener = new BlockingSessionListener();
+
+        // start client
+        WsClient client = new WsClient("ws://localhost:8080/data/websocket", sessionListener);
+        client.start();
+        sessionListener.awaitForSessionReady(10, TimeUnit.SECONDS);
+
+        // subscribe to topic and wait for subscription
+        sessionListener.resetMessageBuffer(1);
+        SubscribeToTopicRequest subscribeToTopicRequest = SubscribeToTopicRequest.newBuilder()
+                .setRequestId(1)
+                .setTopicId(topic)
+                .build();
+        MessageWrapper messageWrapper = MessageWrapper.newBuilder()
+                .setSubscribeToTopicRequest(subscribeToTopicRequest)
+                .build();
+
+        client.sendMessageWrapper(messageWrapper);
+
+        List<MessageWrapper> messageWrappers = sessionListener.awaitMessages(10, TimeUnit.SECONDS);
+        SubscribeToTopicResponse subscribeToTopicResponse = messageWrappers.get(0).getSubscribeToTopicResponse();
+        Assert.assertNotNull(subscribeToTopicResponse);
+        Assert.assertEquals(subscribeToTopicResponse.getRequestId(), subscribeToTopicRequest.getRequestId());
+        Assert.assertNotNull(subscribeToTopicResponse.getSubscriptionId());
+
+        // send data message to topic and wait for message event from same topic
+        sessionListener.resetMessageBuffer(1);
+        DataMessage dataMessage = DataMessage.newBuilder()
+                .setTopicId(topic)
+                .setMessage("hello world")
+                .build();
+        messageWrapper = MessageWrapper.newBuilder()
+                .setDataMessage(dataMessage)
+                .build();
+        client.sendMessageWrapper(messageWrapper);
+
+        messageWrappers = sessionListener.awaitMessages(10, TimeUnit.SECONDS);
+        DataMessageEvent dataMessageEvent = messageWrappers.get(0).getDataMessageEvent();
+        Assert.assertNotNull(dataMessageEvent);
+        Assert.assertEquals(dataMessage.getMessage(), dataMessageEvent.getMessage());
+        Assert.assertEquals(dataMessage.getTopicId(), dataMessageEvent.getTopicId());
+        Assert.assertEquals(subscribeToTopicResponse.getSubscriptionId(), dataMessageEvent.getSubscriptionId());
+
+        // unsubscribe from topic and wait for confirmation
+        sessionListener.resetMessageBuffer(1);
+        UnsubscribeFromTopicRequest unsubscribeFromTopicRequest = UnsubscribeFromTopicRequest.newBuilder()
+                .setRequestId(2)
+                .setSubscriptionId(subscribeToTopicResponse.getSubscriptionId())
+                .build();
+        messageWrapper = MessageWrapper.newBuilder()
+                .setUnsubscribeFromTopicRequest(unsubscribeFromTopicRequest)
+                .build();
+        client.sendMessageWrapper(messageWrapper);
+        messageWrappers = sessionListener.awaitMessages(10, TimeUnit.SECONDS);
+        UnsubscribeFromTopicResponse unsubscribeFromTopicResponse = messageWrappers.get(0).getUnsubscribeFromTopicResponse();
+        Assert.assertNotNull(unsubscribeFromTopicResponse);
+        Assert.assertEquals(unsubscribeFromTopicRequest.getRequestId(), unsubscribeFromTopicResponse.getRequestId());
+
+        // stop client
+        client.stop();
+    }
+
+}