Merge pull request #205 from launchdarkly/eb/ch51696/stream-store-status

(5.0 - #7) add stream logic for data store outages + improve test code

Author: eli-darkly

src/main/java/com/launchdarkly/sdk/server/DataStoreStatusProviderImpl.java

@@ -18,10 +18,8 @@ public Status getStoreStatus() {
   }
 
   @Override
-  public void addStatusListener(StatusListener listener) {
-    if (delegateTo != null) {
-      delegateTo.addStatusListener(listener);
-    }
+  public boolean addStatusListener(StatusListener listener) {
+    return delegateTo != null && delegateTo.addStatusListener(listener);
   }
 
   @Override

src/main/java/com/launchdarkly/sdk/server/DataStoreUpdatesImpl.java

@@ -4,6 +4,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.launchdarkly.sdk.server.DataModelDependencies.KindAndKey;
 import com.launchdarkly.sdk.server.interfaces.DataStore;
+import com.launchdarkly.sdk.server.interfaces.DataStoreStatusProvider;
 import com.launchdarkly.sdk.server.interfaces.DataStoreTypes.DataKind;
 import com.launchdarkly.sdk.server.interfaces.DataStoreTypes.FullDataSet;
 import com.launchdarkly.sdk.server.interfaces.DataStoreTypes.ItemDescriptor;
@@ -32,10 +33,12 @@ final class DataStoreUpdatesImpl implements DataStoreUpdates {
   private final DataStore store;
   private final FlagChangeEventPublisher flagChangeEventPublisher;
   private final DataModelDependencies.DependencyTracker dependencyTracker = new DataModelDependencies.DependencyTracker();
-
+  private final DataStoreStatusProvider dataStoreStatusProvider;
+  
   DataStoreUpdatesImpl(DataStore store, FlagChangeEventPublisher flagChangeEventPublisher) {
     this.store = store;
     this.flagChangeEventPublisher = flagChangeEventPublisher;
+    this.dataStoreStatusProvider = new DataStoreStatusProviderImpl(store);
   }
 
   @Override
@@ -78,6 +81,11 @@ public void upsert(DataKind kind, String key, ItemDescriptor item) {
     }
   }
 
+  @Override
+  public DataStoreStatusProvider getStatusProvider() {
+    return dataStoreStatusProvider;
+  }
+
   private boolean hasFlagChangeEventListeners() {
     return flagChangeEventPublisher != null && flagChangeEventPublisher.hasListeners();
   }

src/main/java/com/launchdarkly/sdk/server/FeatureRequestor.java

@@ -1,9 +1,21 @@
 package com.launchdarkly.sdk.server;
 
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.launchdarkly.sdk.server.DataModel.VersionedData;
+import com.launchdarkly.sdk.server.interfaces.DataStoreTypes.DataKind;
+import com.launchdarkly.sdk.server.interfaces.DataStoreTypes.FullDataSet;
+import com.launchdarkly.sdk.server.interfaces.DataStoreTypes.ItemDescriptor;
+import com.launchdarkly.sdk.server.interfaces.DataStoreTypes.KeyedItems;
+
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.AbstractMap;
 import java.util.Map;
 
+import static com.launchdarkly.sdk.server.DataModel.FEATURES;
+import static com.launchdarkly.sdk.server.DataModel.SEGMENTS;
+
 interface FeatureRequestor extends Closeable {
   DataModel.FeatureFlag getFlag(String featureKey) throws IOException, HttpErrorException;
 
@@ -19,5 +31,23 @@ static class AllData {
       this.flags = flags;
       this.segments = segments;
     }
+
+    FullDataSet<ItemDescriptor> toFullDataSet() {
+      return new FullDataSet<ItemDescriptor>(ImmutableMap.of(
+          FEATURES, toKeyedItems(FEATURES, flags),
+          SEGMENTS, toKeyedItems(SEGMENTS, segments)
+          ).entrySet());
+    }
+
+    static KeyedItems<ItemDescriptor> toKeyedItems(DataKind kind, Map<String, ? extends VersionedData> itemsMap) {
+      ImmutableList.Builder<Map.Entry<String, ItemDescriptor>> builder = ImmutableList.builder();
+      if (itemsMap != null) {
+        for (Map.Entry<String, ? extends VersionedData> e: itemsMap.entrySet()) {
+          ItemDescriptor item = new ItemDescriptor(e.getValue().getVersion(), e.getValue());
+          builder.add(new AbstractMap.SimpleEntry<>(e.getKey(), item));
+        }
+      }
+      return new KeyedItems<>(builder.build());
+    }
   }
 }

src/main/java/com/launchdarkly/sdk/server/PollingProcessor.java

@@ -64,7 +64,7 @@ public Future<Void> start() {
     scheduler.scheduleAtFixedRate(() -> {
       try {
         FeatureRequestor.AllData allData = requestor.getAllData();
-        dataStoreUpdates.init(DefaultFeatureRequestor.toFullDataSet(allData));
+        dataStoreUpdates.init(allData.toFullDataSet());
         if (!initialized.getAndSet(true)) {
           logger.info("Initialized LaunchDarkly client.");
           initFuture.set(null);