feat: Add FDv2 Streaming Synchronizer (#106)

<!-- CURSOR_SUMMARY -->
> [!NOTE]
> **Adds FDv2 streaming support and refines polling behavior**
> 
> - New `StreamingSynchronizerImpl` maintains an SSE connection via
`EventSource`, parses FDv2 events, emits `CHANGE_SET`/status results,
handles recoverable/non-recoverable errors, restarts on `GOODBYE`, and
lazily starts on `next()`; supports selector and `filter` query params
> - Polling updates: map specific `INTERNAL_ERROR` types
(`MISSING_PAYLOAD`, `JSON_ERROR`) to `INVALID_DATA`;
`PollingSynchronizerImpl` now completes a shutdown future on terminal
errors to stop further polling
> - API docs clarified for `Initializer` and `Synchronizer` (state
diagrams; `next()` usage on shutdown/terminal)
> - Extensive tests added for streaming behaviors (events, reconnection,
errors, selector/filter) and for polling internal-error kinds and
shutdown/continuation
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
d4dff45. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

---------

Co-authored-by: Todd Anderson <tanderson@launchdarkly.com>

Author: kinyoklion

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