Support for indirect put events

Author: jkodumal

src/main/java/com/launchdarkly/client/FeatureRequestor.java

@@ -18,10 +18,8 @@
 
 import java.io.IOException;
 import java.lang.reflect.Type;
+import java.util.Map;
 
-/**
- * Created by jkodumal on 11/2/15.
- */
 class FeatureRequestor {
 
   private final String apiKey;
@@ -60,50 +58,90 @@ protected CloseableHttpClient createClient() {
     return client;
   }
 
-  <T> FeatureRep<T> makeRequest(String featureKey, boolean latest) throws IOException {
+  Map<String, FeatureRep<?>> makeAllRequest(boolean latest) throws IOException {
     Gson gson = new Gson();
     HttpCacheContext context = HttpCacheContext.create();
 
-    String resource = latest ? "/api/eval/latest-features/" : "/api/eval/features/";
+    String resource = latest ? "/api/eval/latest-features" : "/api/eval/features";
 
-    HttpGet request = config.getRequest(apiKey,resource + featureKey);
+    HttpGet request = config.getRequest(apiKey, resource);
 
     CloseableHttpResponse response = null;
     try {
       response = client.execute(request, context);
 
-      CacheResponseStatus responseStatus = context.getCacheResponseStatus();
-
-      switch (responseStatus) {
-        case CACHE_HIT:
-          logger.debug("A response was generated from the cache with " +
-              "no requests sent upstream");
-          break;
-        case CACHE_MODULE_RESPONSE:
-          logger.debug("The response was generated directly by the " +
-              "caching module");
-          break;
-        case CACHE_MISS:
-          logger.debug("The response came from an upstream server");
-          break;
-        case VALIDATED:
-          logger.debug("The response was generated from the cache " +
-              "after validating the entry with the origin server");
-          break;
+      logCacheResponse(context.getCacheResponseStatus());
+
+      handleResponseStatus(response.getStatusLine().getStatusCode(), null);
+
+      Type type = new TypeToken<Map<String, FeatureRep<?>>>() {}.getType();
+
+      Map<String, FeatureRep<?>> result = gson.fromJson(EntityUtils.toString(response.getEntity()), type);
+      return result;
+    }
+    finally {
+      try {
+        if (response != null) response.close();
+      } catch (IOException e) {
       }
+    }
+  }
 
-      int status = response.getStatusLine().getStatusCode();
+  void logCacheResponse(CacheResponseStatus status) {
+    switch (status) {
+      case CACHE_HIT:
+        logger.debug("A response was generated from the cache with " +
+            "no requests sent upstream");
+        break;
+      case CACHE_MODULE_RESPONSE:
+        logger.debug("The response was generated directly by the " +
+            "caching module");
+        break;
+      case CACHE_MISS:
+        logger.debug("The response came from an upstream server");
+        break;
+      case VALIDATED:
+        logger.debug("The response was generated from the cache " +
+            "after validating the entry with the origin server");
+        break;
+    }
+  }
 
-      if (status != HttpStatus.SC_OK) {
-        if (status == HttpStatus.SC_UNAUTHORIZED) {
-          logger.error("Invalid API key");
-        } else if (status == HttpStatus.SC_NOT_FOUND) {
+  void handleResponseStatus(int status, String featureKey) throws IOException {
+
+    if (status != HttpStatus.SC_OK) {
+      if (status == HttpStatus.SC_UNAUTHORIZED) {
+        logger.error("Invalid API key");
+      } else if (status == HttpStatus.SC_NOT_FOUND) {
+        if (featureKey != null) {
           logger.error("Unknown feature key: " + featureKey);
-        } else {
-          logger.error("Unexpected status code: " + status);
         }
-        throw new IOException("Failed to fetch flag");
+        else {
+          logger.error("Resource not found");
+        }
+      } else {
+        logger.error("Unexpected status code: " + status);
       }
+      throw new IOException("Failed to fetch flag");
+    }
+
+  }
+
+  <T> FeatureRep<T> makeRequest(String featureKey, boolean latest) throws IOException {
+    Gson gson = new Gson();
+    HttpCacheContext context = HttpCacheContext.create();
+
+    String resource = latest ? "/api/eval/latest-features/" : "/api/eval/features/";
+
+    HttpGet request = config.getRequest(apiKey,resource + featureKey);
+
+    CloseableHttpResponse response = null;
+    try {
+      response = client.execute(request, context);
+
+      logCacheResponse(context.getCacheResponseStatus());
+
+      handleResponseStatus(response.getStatusLine().getStatusCode(), featureKey);
 
       Type type = new TypeToken<FeatureRep<T>>() {}.getType();
 

src/main/java/com/launchdarkly/client/StreamProcessor.java

@@ -21,6 +21,7 @@ class StreamProcessor implements Closeable {
   private static final String PUT = "put";
   private static final String PATCH = "patch";
   private static final String DELETE = "delete";
+  private static final String INDIRECT_PUT = "indirect/put";
   private static final String INDIRECT_PATCH = "indirect/patch";
   private static final Logger logger = LoggerFactory.getLogger(StreamProcessor.class);
 
@@ -65,8 +66,16 @@ else if (event.getName().equals(DELETE)) {
           FeatureDeleteData data = gson.fromJson(event.readData(), FeatureDeleteData.class);
           store.delete(data.key(), data.version());
         }
+        else if (event.getName().equals(INDIRECT_PUT)) {
+          try {
+            Map<String, FeatureRep<?>> features = requestor.makeAllRequest(true);
+            store.init(features);
+          } catch (IOException e) {
+            logger.error("Encountered exception in LaunchDarkly client", e);
+          }
+        }
         else if (event.getName().equals(INDIRECT_PATCH)) {
-          String key = gson.fromJson(event.readData(), String.class);
+          String key = event.readData();
           try {
             FeatureRep<?> feature = requestor.makeRequest(key, true);
             store.upsert(key, feature);