Threading and tests.

Author: kinyoklion

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingSynchronizerImpl.java

@@ -38,11 +38,12 @@ public PollingSynchronizerImpl(
     private void doPoll() {
         try {
             FDv2SourceResult res = poll(selectorSource.getSelector(), false).get();
-            switch(res.getResultType()) {
+            boolean shouldShutdown = false;
+            switch (res.getResultType()) {
                 case CHANGE_SET:
                     break;
                 case STATUS:
-                    switch(res.getStatus().getState()) {
+                    switch (res.getStatus().getState()) {
                         case INTERRUPTED:
                             break;
                         case SHUTDOWN:
@@ -54,6 +55,7 @@ private void doPoll() {
                                 task.cancel(true);
                             }
                             internalShutdown();
+                            shouldShutdown = true;
                             break;
                         case GOODBYE:
                             // We don't need to take any action, as the connection for the poll
@@ -63,7 +65,11 @@ private void doPoll() {
                     }
                     break;
             }
-            resultQueue.put(res);
+            if (shouldShutdown) {
+                shutdownFuture.complete(res);
+            } else {
+                resultQueue.put(res);
+            }
         } catch (InterruptedException | ExecutionException e) {
             // TODO: Determine if handling is needed.
         }

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamingSynchronizerImpl.java

@@ -52,7 +52,8 @@ class StreamingSynchronizerImpl implements Synchronizer {
     private final String payloadFilter;
     private final IterableAsyncQueue<FDv2SourceResult> resultQueue = new IterableAsyncQueue<>();
     private final CompletableFuture<FDv2SourceResult> shutdownFuture = new CompletableFuture<>();
-    private final AtomicBoolean closed = new AtomicBoolean(false);
+    private boolean closed = false;
+    private final Object closeLock = new Object();
     private final FDv2ProtocolHandler protocolHandler = new FDv2ProtocolHandler();
     private volatile EventSource eventSource;
     final Duration initialReconnectDelay;
@@ -152,8 +153,10 @@ private Thread getRunThread() {
                 // So we have to assume something is wrong that we can't recover from at this point,
                 // and just let the thread terminate. That's better than having the thread be killed
                 // by an uncaught exception.
-                if (closed.get()) {
-                    return; // ignore any exception that's just a side effect of stopping the EventSource
+                synchronized (closeLock) {
+                    if (closed) {
+                        return;
+                    }
                 }
                 logger.error("Stream thread ended with exception: {}", LogValues.exceptionSummary(e));
                 logger.debug(LogValues.exceptionTrace(e));
@@ -181,29 +184,34 @@ private Thread getRunThread() {
     @Override
     public CompletableFuture<FDv2SourceResult> next() {
         // If we are already closed, don't start the stream.
-        if (!started.getAndSet(true) && !closed.get()) {
-            startStream();
+        synchronized (closeLock) {
+            if (!closed) {
+                if (!started.getAndSet(true)) {
+                    startStream();
+                }
+            }
         }
+
         return CompletableFuture.anyOf(shutdownFuture, resultQueue.take())
                 .thenApply(result -> (FDv2SourceResult) result);
     }
 
     @Override
     public void close() {
-        if (closed.getAndSet(true)) {
-            return; // already shutdown
-        }
-
-        shutdownFuture.complete(FDv2SourceResult.shutdown());
+        synchronized (closeLock){
+            closed = true;
 
-        // If the synchronizer was never started, then the event source could be null.
-        if (eventSource != null) {
-            try {
-                eventSource.close();
-            } catch (Exception e) {
-                logger.debug("Error closing event source during shutdown: {}", LogValues.exceptionSummary(e));
+            // If the synchronizer was never started, then the event source could be null.
+            if (eventSource != null) {
+                try {
+                    eventSource.close();
+                } catch (Exception e) {
+                    logger.debug("Error closing event source during shutdown: {}", LogValues.exceptionSummary(e));
+                }
             }
         }
+
+        shutdownFuture.complete(FDv2SourceResult.shutdown());
     }
 
     private boolean handleEvent(StreamEvent event) {
@@ -338,7 +346,7 @@ private boolean handleError(StreamException e) {
                     "will retry");
 
             if (!recoverable) {
-                resultQueue.put(FDv2SourceResult.terminalError(errorInfo));
+                shutdownFuture.complete(FDv2SourceResult.terminalError(errorInfo));
                 return false;
             } else {
                 // Queue as INTERRUPTED to indicate temporary failure

lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingInitializerImplTest.java

@@ -339,7 +339,7 @@ public void internalErrorWithInvalidDataKind() throws Exception {
 
         // Create a response with malformed payload-transferred event. `state->states`.
         // This will trigger JSON_ERROR internal error which maps to INVALID_DATA
-        String malformedPutObjectJson = "{\n" +
+        String malformedPayloadTransferred = "{\n" +
             "  \"events\": [\n" +
             "    {\n" +
             "      \"event\": \"server-intent\",\n" +
@@ -367,7 +367,7 @@ public void internalErrorWithInvalidDataKind() throws Exception {
             "}";
 
         FDv2Requestor.FDv2PayloadResponse response = new FDv2Requestor.FDv2PayloadResponse(
-            com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event.parseEventsArray(malformedPutObjectJson),
+            com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event.parseEventsArray(malformedPayloadTransferred),
             okhttp3.Headers.of()
         );
 

lib/sdk/server/src/test/java/com/launchdarkly/sdk/server/PollingSynchronizerImplTest.java

@@ -733,8 +733,8 @@ public void internalErrorWithInvalidDataKindContinuesPolling() throws Exception
         when(requestor.Poll(any(Selector.class))).thenAnswer(invocation -> {
             int count = callCount.incrementAndGet();
             if (count == 1) {
-                // First call returns response with malformed put-object which triggers INTERNAL_ERROR (INVALID_DATA)
-                String malformedPutObjectJson = "{\n" +
+                // First call returns response with malformed payload transfer. state->states
+                String malformedPayloadTransferred = "{\n" +
                     "  \"events\": [\n" +
                     "    {\n" +
                     "      \"event\": \"server-intent\",\n" +
@@ -750,7 +750,7 @@ public void internalErrorWithInvalidDataKindContinuesPolling() throws Exception
                     "    {\n" +
                     "      \"event\": \"payload-transferred\",\n" +
                     "      \"data\": {\n" +
-                    "        \"state\": \"(p:payload-1:100)\",\n" +
+                    "        \"states\": \"(p:payload-1:100)\",\n" +
                     "        \"version\": 100\n" +
                     "      }\n" +
                     "    },\n" +
@@ -762,7 +762,7 @@ public void internalErrorWithInvalidDataKindContinuesPolling() throws Exception
                     "}";
 
                 return CompletableFuture.completedFuture(new FDv2Requestor.FDv2PayloadResponse(
-                    com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event.parseEventsArray(malformedPutObjectJson),
+                    com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event.parseEventsArray(malformedPayloadTransferred),
                     okhttp3.Headers.of()
                 ));
             } else {