Merge pull request #171 from splitio/release/4.1.3

Release/4.1.3

Author: mredolatti

client/CHANGES.txt

@@ -1,5 +1,9 @@
 CHANGES
 
+4.1.3 (Dec 2, 2020)
+- Fix Issue when closing SSE Connection
+- Updated log-level for some messages
+
 4.1.2 (Nov 25, 2020)
 - Updated junit from 4.12 to 4.13.1
 - Updated HttpClient from 4.5.2 to 5.0.3

client/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.split.client</groupId>
         <artifactId>java-client-parent</artifactId>
-        <version>4.1.2</version>
+        <version>4.1.3</version>
     </parent>
     <artifactId>java-client</artifactId>
     <packaging>jar</packaging>

client/src/main/java/io/split/engine/sse/AuthApiClientImp.java

@@ -46,14 +46,14 @@ public AuthenticationResponse Authenticate() {
                 return getSuccessResponse(jsonContent);
             }
 
-            _log.debug(String.format("Problem to connect to : %s. Response status: %s", _target, statusCode));
+            _log.error(String.format("Problem to connect to : %s. Response status: %s", _target, statusCode));
             if (statusCode >= HttpStatus.SC_BAD_REQUEST && statusCode < HttpStatus.SC_INTERNAL_SERVER_ERROR) {
                 return new AuthenticationResponse(false,false);
             }
 
             return new AuthenticationResponse(false,true);
         } catch (Exception ex) {
-            _log.debug(ex.getMessage());
+            _log.error(ex.getMessage());
             return new AuthenticationResponse(false,true);
         }
     }

client/src/main/java/io/split/engine/sse/EventSourceClientImp.java

@@ -65,15 +65,15 @@ public boolean start(String channelList, String token) {
         try {
             return _sseClient.open(buildUri(channelList, token));
         } catch (URISyntaxException e) {
-            _log.debug("Error building Streaming URI: " + e.getMessage());
+            _log.error("Error building Streaming URI: " + e.getMessage());
             return false;
         }
     }
 
     @Override
     public void stop() {
         if (!_sseClient.isOpen()) {
-            _log.debug("Event Source Client is closed.");
+            _log.info("Event Source Client is closed.");
             return;
         }
         _sseClient.close();

client/src/main/java/io/split/engine/sse/client/SSEClient.java

@@ -1,13 +1,10 @@
 package io.split.engine.sse.client;
 
 import com.google.common.base.Strings;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.hc.client5.http.classic.methods.HttpGet;
-import org.apache.hc.client5.http.config.RequestConfig;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
-import org.apache.hc.client5.http.impl.classic.HttpClients;
-import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
-import org.apache.hc.core5.util.Timeout;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -19,6 +16,8 @@
 import java.net.SocketException;
 import java.net.URI;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
@@ -45,11 +44,16 @@ private enum ConnectionState {
     private final static long CONNECT_TIMEOUT = 30000;
     private static final Logger _log = LoggerFactory.getLogger(SSEClient.class);
 
+    private final ExecutorService _connectionExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
+            .setDaemon(true)
+            .setNameFormat("SPLIT-SSEConnection-%d")
+            .build());
     private final CloseableHttpClient _client;
     private final Function<RawEvent, Void> _eventCallback;
     private final Function<StatusMessage, Void> _statusCallback;
     private final AtomicReference<ConnectionState> _state = new AtomicReference<>(ConnectionState.CLOSED);
     private final AtomicReference<CloseableHttpResponse> _ongoingResponse = new AtomicReference<>();
+    private final AtomicReference<HttpGet> _ongoingRequest = new AtomicReference<>();
 
     public SSEClient(Function<RawEvent, Void> eventCallback,
                      Function<StatusMessage, Void> statusCallback,
@@ -61,23 +65,21 @@ public SSEClient(Function<RawEvent, Void> eventCallback,
 
     public synchronized boolean open(URI uri) {
         if (isOpen()) {
-            _log.debug("SSEClient already open.");
+            _log.info("SSEClient already open.");
             return false;
         }
 
         _statusCallback.apply(StatusMessage.INITIALIZATION_IN_PROGRESS);
 
         CountDownLatch signal = new CountDownLatch(1);
-        Thread thread = new Thread(() -> connectAndLoop(uri, signal));
-        thread.setDaemon(true);
-        thread.start();
+        _connectionExecutor.submit(() -> connectAndLoop(uri, signal));
         try {
             if (!signal.await(CONNECT_TIMEOUT, TimeUnit.SECONDS)) {
                 return false;
-            };
+            }
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
-            _log.debug(e.getMessage());
+            _log.info(e.getMessage());
             return false;
         }
         return isOpen();
@@ -91,9 +93,10 @@ public synchronized void close() {
         if (_state.compareAndSet(ConnectionState.OPEN, ConnectionState.CLOSED)) {
             if (_ongoingResponse.get() != null) {
                 try {
+                    _ongoingRequest.get().abort();
                     _ongoingResponse.get().close();
                 } catch (IOException e) {
-                    _log.debug(String.format("Error closing SSEClient: %s", e.getMessage()));
+                    _log.info(String.format("Error closing SSEClient: %s", e.getMessage()));
                 }
             }
         }
@@ -124,7 +127,7 @@ private void connectAndLoop(URI uri, CountDownLatch signal) {
                     _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
                     return;
                 } catch (IOException exc) { // Other type of connection error
-                    _log.debug(exc.getMessage());
+                    _log.info(String.format("SSE connection ended abruptly: %s. Retying", exc.getMessage()));
                     _statusCallback.apply(StatusMessage.RETRYABLE_ERROR);
                     return;
                 }
@@ -145,17 +148,18 @@ private void connectAndLoop(URI uri, CountDownLatch signal) {
     }
 
     private boolean establishConnection(URI uri, CountDownLatch signal) {
-        HttpGet request = new HttpGet(uri);
+
+        _ongoingRequest.set(new HttpGet(uri));
 
         try {
-            _ongoingResponse.set(_client.execute(request));
+            _ongoingResponse.set(_client.execute(_ongoingRequest.get()));
             if (_ongoingResponse.get().getCode() != 200) {
                 return false;
             }
             _state.set(ConnectionState.OPEN);
             _statusCallback.apply(StatusMessage.CONNECTED);
         } catch (IOException exc) {
-            _log.debug(String.format("Error establishConnection: %s", exc));
+            _log.error(String.format("Error establishConnection: %s", exc));
             return false;
         } finally {
             signal.countDown();

pom.xml

@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>io.split.client</groupId>
     <artifactId>java-client-parent</artifactId>
-    <version>4.1.2</version>
+    <version>4.1.3</version>
     <dependencyManagement>
         <dependencies>
             <dependency>

testing/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>io.split.client</groupId>
         <artifactId>java-client-parent</artifactId>
-        <version>4.1.2</version>
+        <version>4.1.3</version>
     </parent>
 
     <artifactId>java-client-testing</artifactId>