feat: Add FDv2 Streaming Synchronizer

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

@@ -124,8 +124,20 @@ protected CompletableFuture<FDv2SourceResult> poll(Selector selector, boolean on
                     case NONE:
                         break;
                     case INTERNAL_ERROR: {
+                        FDv2ProtocolHandler.FDv2ActionInternalError internalErrorAction = (FDv2ProtocolHandler.FDv2ActionInternalError) res;
+                        DataSourceStatusProvider.ErrorKind kind = DataSourceStatusProvider.ErrorKind.UNKNOWN;
+                        switch (internalErrorAction.getErrorType()) {
+                            case MISSING_PAYLOAD:
+                            case JSON_ERROR:
+                                kind = DataSourceStatusProvider.ErrorKind.INVALID_DATA;
+                                break;
+                            case UNKNOWN_EVENT:
+                            case IMPLEMENTATION_ERROR:
+                            case PROTOCOL_ERROR:
+                                break;
+                        }
                         DataSourceStatusProvider.ErrorInfo info = new DataSourceStatusProvider.ErrorInfo(
-                                DataSourceStatusProvider.ErrorKind.UNKNOWN,
+                                kind,
                                 0,
                                 "Internal error occurred during polling",
                                 new Date().toInstant());

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.
         }