remove jersey from actual SDK. only keep as test dependency

client/CHANGES.txt

@@ -1,5 +1,9 @@
 CHANGES
 
+4.0.1 (Sep 4, 2020)
+- Remove jersey. Use custom SSE implementation
+- Bumped guava version to 29
+
 4.0.0 (Aug 19, 2020)
 - Deprecated Java 7 support. Java 8 is the minimum supported version for this and future releases.
 - Added support for the new Split streaming architecture. When enabled (default), the SDK will not poll for updates but instead receive notifications every time there's a change in your environments, allowing to process those much quicker. If disabled or in the event of an issue, the SDK will fallback to the known polling mechanism to provide a seamless experience.

client/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.split.client</groupId>
         <artifactId>java-client-parent</artifactId>
-        <version>4.0.0</version>
+        <version>4.0.1</version>
     </parent>
     <artifactId>java-client</artifactId>
     <packaging>jar</packaging>
@@ -47,12 +47,6 @@
                                     <include>org.apache.httpcomponents:httpcore</include>
                                     <include>com.google.code.gson:gson</include>
                                     <include>com.google.guava:guava</include>
-                                    <include>org.yaml.snakeyaml.*</include>
-                                    <include>org.glassfish.*</include>
-                                    <include>jakarta.*</include>
-                                    <include>org.javassist:javassist</include>
-                                    <include>com.sun.activation:jakarta.activation</include>
-                                    <include>org.jvnet.*</include>
                                 </includes>
                             </artifactSet>
                             <transformers>
@@ -68,26 +62,6 @@
                                     <pattern>com.google</pattern>
                                     <shadedPattern>split.com.google</shadedPattern>
                                 </relocation>
-                                <relocation>
-                                    <pattern>org.glassfish</pattern>
-                                    <shadedPattern>split.org.glassfish</shadedPattern>
-                                </relocation>
-                                <relocation>
-                                    <pattern>jakarta</pattern>
-                                    <shadedPattern>split.jakarta</shadedPattern>
-                                </relocation>
-                                <relocation>
-                                    <pattern>org.javassist</pattern>
-                                    <shadedPattern>split.org.javassist</shadedPattern>
-                                </relocation>
-                                <relocation>
-                                    <pattern>com.sun.activation</pattern>
-                                    <shadedPattern>split.com.sun.activation</shadedPattern>
-                                </relocation>
-                                <relocation>
-                                    <pattern>org.jvnet</pattern>
-                                    <shadedPattern>split.org.jvnet</shadedPattern>
-                                </relocation>
                             </relocations>
                             <filters>
                                 <filter>
@@ -96,9 +70,7 @@
                                         <exclude>META-INF/license/**</exclude>
                                         <exclude>META-INF/*</exclude>
                                         <exclude>META-INF/maven/**</exclude>
-                                        <!-- Disabled to allow Jersey implementation mappings/injector to work
-                                           <exclude>META-INF/services/**</exclude>
-                                        -->
+                                        <exclude>META-INF/services/**</exclude>
                                         <exclude>LICENSE</exclude>
                                         <exclude>NOTICE</exclude>
                                         <exclude>/*.txt</exclude>
@@ -149,7 +121,7 @@
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
-            <version>18.0</version>
+            <version>29.0-jre</version>
         </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
@@ -170,16 +142,6 @@
             <artifactId>snakeyaml</artifactId>
             <version>1.21</version>
         </dependency>
-        <dependency>
-            <groupId>org.glassfish.jersey.media</groupId>
-            <artifactId>jersey-media-sse</artifactId>
-            <version>2.31</version>
-        </dependency>
-        <dependency>
-            <groupId>org.glassfish.jersey.inject</groupId>
-            <artifactId>jersey-hk2</artifactId>
-            <version>2.31</version>
-        </dependency>
 
         <!-- Test deps -->
         <dependency>
@@ -211,6 +173,18 @@
             <version>1.7.21</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.glassfish.jersey.media</groupId>
+            <artifactId>jersey-media-sse</artifactId>
+            <version>2.31</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.glassfish.jersey.inject</groupId>
+            <artifactId>jersey-hk2</artifactId>
+            <version>2.31</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.glassfish.jersey.containers</groupId>
             <artifactId>jersey-container-grizzly2-http</artifactId>

client/src/main/java/io/split/client/SplitClientConfig.java

@@ -275,7 +275,7 @@ public static final class Builder {
         private int _authRetryBackoffBase = 1;
         private int _streamingReconnectBackoffBase = 1;
         private String _authServiceURL = "https://auth.split.io/api/auth";
-        private String _streamingServiceURL = "https://streaming.split.io/event-stream";
+        private String _streamingServiceURL = "https://streaming.split.io/sse";
 
         public Builder() {
         }

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

@@ -72,6 +72,6 @@ private AuthenticationResponse getSuccessResponse(String jsonContent) {
             expiration = response.getExpiration();
         }
 
-        return new AuthenticationResponse(response.isPushEnabled(), response.getToken(), channels, 3000/*expiration*/, false);
+        return new AuthenticationResponse(response.isPushEnabled(), response.getToken(), channels, expiration, false);
     }
 }

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

@@ -1,6 +1,8 @@
 package io.split.engine.sse;
 
 import com.google.common.annotations.VisibleForTesting;
+import io.split.engine.sse.client.RawEvent;
+import io.split.engine.sse.client.SSEClient;
 import io.split.engine.sse.dtos.SegmentQueueDto;
 import io.split.engine.sse.exceptions.EventParsingException;
 import io.split.engine.sse.workers.SplitsWorker;
@@ -9,39 +11,34 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.ws.rs.client.Client;
-import javax.ws.rs.client.ClientBuilder;
-import javax.ws.rs.client.WebTarget;
-import javax.ws.rs.sse.InboundSseEvent;
+import java.net.URI;
 import java.net.URISyntaxException;
-import java.util.concurrent.TimeUnit;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
 public class EventSourceClientImp implements EventSourceClient {
     private static final Logger _log = LoggerFactory.getLogger(EventSourceClient.class);
 
     private final String _baseStreamingUrl;
-    private final Client _client;
     private final NotificationParser _notificationParser;
     private final NotificationProcessor _notificationProcessor;
-    private final SplitSseEventSource _splitSseEventSource;
+    private final SSEClient _sseClient;
     private final PushStatusTracker _pushStatusTracker;
 
     @VisibleForTesting
     /* package private */ EventSourceClientImp(String baseStreamingUrl,
                                                NotificationParser notificationParser,
                                                NotificationProcessor notificationProcessor,
-                                               Client client,
                                                PushStatusTracker pushStatusTracker) {
         _baseStreamingUrl = checkNotNull(baseStreamingUrl);
         _notificationParser = checkNotNull(notificationParser);
         _notificationProcessor = checkNotNull(notificationProcessor);
-        _client = checkNotNull(client);
-        _splitSseEventSource = new SplitSseEventSource(
-                inboundEvent -> { onMessage(inboundEvent); return null; },
-                status -> { onSSeStatusChange(status); return null; });
         _pushStatusTracker = pushStatusTracker;
+
+        _sseClient = new SSEClient(
+                inboundEvent -> { onMessage(inboundEvent); return null; },
+                status -> { _pushStatusTracker.handleSseStatus(status); return null; });
+
     }
 
     public static EventSourceClientImp build(String baseStreamingUrl,
@@ -52,18 +49,17 @@ public static EventSourceClientImp build(String baseStreamingUrl,
         return new EventSourceClientImp(baseStreamingUrl,
                 new NotificationParserImp(),
                 NotificationProcessorImp.build(splitsWorker, segmentWorker, pushStatusTracker),
-                ClientBuilder.newBuilder().readTimeout(70, TimeUnit.SECONDS).build(),
                 pushStatusTracker);
     }
 
     @Override
     public boolean start(String channelList, String token) {
-        if (_splitSseEventSource.isOpen()) {
-            _splitSseEventSource.close();
+        if (_sseClient.isOpen()) {
+            _sseClient.close();
         }
 
         try {
-            return _splitSseEventSource.open(buildTarget(channelList, token));
+            return _sseClient.open(buildUri(channelList, token));
         } catch (URISyntaxException e) {
             _log.error("Error building Streaming URI: " + e.getMessage());
             return false;
@@ -72,25 +68,25 @@ public boolean start(String channelList, String token) {
 
     @Override
     public void stop() {
-        if (!_splitSseEventSource.isOpen()) {
+        if (!_sseClient.isOpen()) {
             _log.warn("Event Source Client is closed.");
             return;
         }
-        _splitSseEventSource.close();
+        _sseClient.close();
     }
 
-    private WebTarget buildTarget(String channelList, String token) throws URISyntaxException {
-        return _client.target(new URIBuilder(_baseStreamingUrl)
+    private URI buildUri(String channelList, String token) throws URISyntaxException {
+        return new URIBuilder(_baseStreamingUrl)
                 .addParameter("channels", channelList)
                 .addParameter("v", "1.1")
                 .addParameter("accessToken", token)
-                .build());
+                .build();
     }
 
-    private void onMessage(InboundSseEvent event) {
+    private void onMessage(RawEvent event) {
         try {
-            String type = event.getName();
-            String payload = event.readData();
+            String type = event.event();
+            String payload = event.data();
             if (payload.length() > 0) {
                 _log.debug(String.format("Payload received: %s", payload));
                 switch (type) {
@@ -110,8 +106,4 @@ private void onMessage(InboundSseEvent event) {
             _log.warn(String.format("Error onMessage: %s", e.getMessage()));
         }
     }
-
-    private void onSSeStatusChange(SseStatus status) {
-        _pushStatusTracker.handleSseStatus(status);
-    }
 }

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

@@ -1,5 +1,6 @@
 package io.split.engine.sse;
 
+import io.split.engine.sse.client.SSEClient;
 import io.split.engine.sse.dtos.ControlNotification;
 import io.split.engine.sse.dtos.ErrorNotification;
 import io.split.engine.sse.dtos.OccupancyNotification;
@@ -8,6 +9,6 @@ public interface PushStatusTracker {
     void handleIncomingControlEvent(ControlNotification controlNotification);
     void handleIncomingOccupancyEvent(OccupancyNotification occupancyNotification);
     void handleIncomingAblyError(ErrorNotification notification);
-    void handleSseStatus(SseStatus newStatus);
+    void handleSseStatus(SSEClient.StatusMessage newStatus);
     void forcePushDisable();
 }

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

@@ -1,6 +1,7 @@
 package io.split.engine.sse;
 
 import io.split.engine.common.PushManager;
+import io.split.engine.sse.client.SSEClient;
 import io.split.engine.sse.dtos.ControlNotification;
 import io.split.engine.sse.dtos.ControlType;
 import io.split.engine.sse.dtos.ErrorNotification;
@@ -16,7 +17,7 @@ public class PushStatusTrackerImp implements PushStatusTracker {
     private static final Logger _log = LoggerFactory.getLogger(PushStatusTracker.class);
 
     private final AtomicBoolean _publishersOnline = new AtomicBoolean(true);
-    private final AtomicReference<SseStatus> _sseStatus = new AtomicReference<>(SseStatus.DISCONNECTED);
+    private final AtomicReference<SSEClient.StatusMessage> _sseStatus = new AtomicReference<>(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS);
     private final AtomicReference<ControlType> _backendStatus = new AtomicReference<>(ControlType.STREAMING_RESUMED);
     private final LinkedBlockingQueue<PushManager.Status> _statusMessages;
 
@@ -26,33 +27,40 @@ public PushStatusTrackerImp(LinkedBlockingQueue<PushManager.Status> statusMessag
 
     public synchronized void reset() {
         _publishersOnline.set(true);
-        _sseStatus.set(SseStatus.DISCONNECTED);
+        _sseStatus.set(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS);
         _backendStatus.set(ControlType.STREAMING_RESUMED);
     }
 
     @Override
-    public void handleSseStatus(SseStatus newStatus) {
-        _log.debug(String.format("handleSseStatus new status: %s", newStatus.toString()));
-        _log.debug(String.format("handleSseStatus current status: %s", _sseStatus.get().toString()));
+    public void handleSseStatus(SSEClient.StatusMessage newStatus) {
+        _log.debug(String.format("Current status: %s. New status: %s", _sseStatus.get().toString(), newStatus.toString()));
+
         switch(newStatus) {
             case CONNECTED:
-                if (_sseStatus.compareAndSet(SseStatus.DISCONNECTED, SseStatus.CONNECTED)
-                    || _sseStatus.compareAndSet(SseStatus.RETRYABLE_ERROR, SseStatus.CONNECTED)) {
+                if (_sseStatus.compareAndSet(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS, SSEClient.StatusMessage.CONNECTED)
+                    || _sseStatus.compareAndSet(SSEClient.StatusMessage.RETRYABLE_ERROR, SSEClient.StatusMessage.CONNECTED)) {
                     _statusMessages.offer(PushManager.Status.STREAMING_READY);
                 }
                 break;
             case RETRYABLE_ERROR:
-                if (_sseStatus.compareAndSet(SseStatus.CONNECTED, SseStatus.RETRYABLE_ERROR)) {
+                if (_sseStatus.compareAndSet(SSEClient.StatusMessage.CONNECTED, SSEClient.StatusMessage.RETRYABLE_ERROR)) {
                     _statusMessages.offer(PushManager.Status.STREAMING_BACKOFF);
                 }
                 break;
             case NONRETRYABLE_ERROR:
-                if (_sseStatus.compareAndSet(SseStatus.CONNECTED, SseStatus.NONRETRYABLE_ERROR)
-                    || _sseStatus.compareAndSet(SseStatus.RETRYABLE_ERROR, SseStatus.NONRETRYABLE_ERROR)) {
+                if (_sseStatus.compareAndSet(SSEClient.StatusMessage.CONNECTED, SSEClient.StatusMessage.NONRETRYABLE_ERROR)
+                    || _sseStatus.compareAndSet(SSEClient.StatusMessage.RETRYABLE_ERROR, SSEClient.StatusMessage.NONRETRYABLE_ERROR)) {
                     _statusMessages.offer(PushManager.Status.STREAMING_OFF);
                 }
                 break;
-            case DISCONNECTED: // Restore initial status
+            case FORCED_STOP:
+                if (_sseStatus.compareAndSet(SSEClient.StatusMessage.CONNECTED, SSEClient.StatusMessage.FORCED_STOP)
+                    || _sseStatus.compareAndSet(SSEClient.StatusMessage.RETRYABLE_ERROR, SSEClient.StatusMessage.FORCED_STOP)
+                    || _sseStatus.compareAndSet(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS, SSEClient.StatusMessage.FORCED_STOP)) {
+                    _statusMessages.offer(PushManager.Status.STREAMING_DOWN);
+                }
+                break;
+            case INITIALIZATION_IN_PROGRESS: // Restore initial status
                 reset();
                 break;
         }
@@ -61,6 +69,7 @@ public void handleSseStatus(SseStatus newStatus) {
     @Override
     public void handleIncomingControlEvent(ControlNotification controlNotification) {
         _log.debug(String.format("handleIncomingOccupancyEvent: %s", controlNotification.getControlType()));
+
         if (_backendStatus.get().equals(ControlType.STREAMING_DISABLED)) {
             return;
         }
@@ -87,6 +96,7 @@ public void handleIncomingControlEvent(ControlNotification controlNotification)
     @Override
     public void handleIncomingOccupancyEvent(OccupancyNotification occupancyNotification) {
         _log.debug(String.format("handleIncomingOccupancyEvent: publishers=%d", occupancyNotification.getMetrics().getPublishers()));
+
         int publishers = occupancyNotification.getMetrics().getPublishers();
         if (publishers <= 0 && _publishersOnline.compareAndSet(true, false) && _backendStatus.get().equals(ControlType.STREAMING_RESUMED)) {
             _statusMessages.offer(PushManager.Status.STREAMING_DOWN);
@@ -98,6 +108,7 @@ public void handleIncomingOccupancyEvent(OccupancyNotification occupancyNotifica
     @Override
     public void handleIncomingAblyError(ErrorNotification notification) {
         _log.debug(String.format("handleIncomingAblyError: %s", notification.getMessage()));
+
         if (_backendStatus.get().equals(ControlType.STREAMING_DISABLED)) {
             return; // Ignore
         }
@@ -112,8 +123,9 @@ public void handleIncomingAblyError(ErrorNotification notification) {
     @Override
     public synchronized void forcePushDisable() {
         _log.debug("forcePushDisable");
+
         _publishersOnline.set(false);
-        _sseStatus.set(SseStatus.DISCONNECTED);
+        _sseStatus.set(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS);
         _backendStatus.set(ControlType.STREAMING_DISABLED);
         _statusMessages.offer(PushManager.Status.STREAMING_OFF);
     }