Merge pull request #1 from manpreet333/feature/master/add-system-test

Added system test to demonstrate usage & test the complete functionality

Author: preet-dev

sseclient/build.gradle

@@ -18,6 +18,10 @@ dependencies {
     testCompile group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-jsr310', version: '2.10.2'
     testCompile group: 'commons-io', name: 'commons-io', version: '2.8.0'
     testCompile group: 'org.testng', name: 'testng', version: '7.1.0'
+    testCompile group: 'org.slf4j', name: 'slf4j-api', version: '1.7.30'
+    testCompile group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.3'
+    testCompile group: 'org.projectlombok', name: 'lombok', version: '1.18.12'
+    testAnnotationProcessor group: 'org.projectlombok', name: 'lombok', version: '1.18.12'
 }
 
 jar {

sseclient/src/test/java/org/apache/client/sse/SseTest.java

@@ -0,0 +1,47 @@
+package org.apache.client.sse;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.client.sse.utils.TestUtils;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+@Slf4j
+public class SseTest {
+
+    private final String SSEUri = "http://abrandao.com/lab/Javascript/Javascript_SSEvents/sserver_cpu.php";
+    private final int MAX_CONCURRENT_SSE_STREAMS = 6;
+    private CloseableHttpAsyncClient asyncClient;
+    private ApacheHttpSseClient sseClient;
+
+    @Test(groups = { "system" })
+    public void testSSEClient() throws ExecutionException, InterruptedException {
+        //Initialize client
+        asyncClient = HttpAsyncClients.createDefault();
+        asyncClient.start();
+        sseClient = new ApacheHttpSseClient(asyncClient, Executors.newFixedThreadPool(MAX_CONCURRENT_SSE_STREAMS));
+
+        //Initialize request
+        SseRequest sseRequest = new SseRequest(SSEUri);
+        Future<SseResponse> sseResponseFuture = sseClient.execute(sseRequest);
+
+        //Add a 10 second delay for some events to be processed by the streaming client
+        TestUtils.sleep(10);
+
+        //Validate events exists and log them
+        SseResponse response = sseResponseFuture.get();
+        SseEntity responseEntity = response.getEntity();
+        Assert.assertTrue(responseEntity.hasMoreEvents(), "Expected event stream to have some events");
+        BlockingQueue<Event> eventList = responseEntity.getEvents();
+        for (Event eachEvent: eventList) {
+            TestUtils.logEvent(eachEvent);
+        }
+    }
+
+}

sseclient/src/test/java/org/apache/client/sse/utils/TestUtils.java

@@ -0,0 +1,25 @@
+package org.apache.client.sse.utils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.client.sse.Event;
+
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+public class TestUtils {
+
+    public static void sleep(int seconds) {
+        try {
+            TimeUnit.SECONDS.sleep(seconds);
+        } catch (InterruptedException excObj) {
+            log.error("Exception occurred while waiting for " + seconds + " seconds", excObj);
+        }
+    }
+
+    public static void logEvent(Event event) {
+        log.info("EVENT ID: " + event.getId());
+        log.info("EVENT TYPE: " + (event.getEvent() == null ? "" : event.getEvent()));
+        log.info("EVENT RETRY ms: " + event.getRetry());
+        log.info("EVENT DATA: " + event.getData());
+    }
+}