Add event sampling. Name threads.

Author: Dan Richelson

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

@@ -1,5 +1,6 @@
 package com.launchdarkly.client;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.gson.Gson;
 import org.apache.http.HttpStatus;
 import org.apache.http.client.methods.CloseableHttpResponse;
@@ -14,22 +15,34 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Random;
 import java.util.concurrent.*;
 
 class EventProcessor implements Closeable {
-  private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory());
+  ThreadFactory threadFactory = new ThreadFactoryBuilder()
+      .setDaemon(true)
+      .setNameFormat("LaunchDarkly-EventProcessor-%d")
+      .build();
+  private final ScheduledExecutorService scheduler =
+      Executors.newSingleThreadScheduledExecutor(threadFactory);
+  private final Random random = new Random();
   private final BlockingQueue<Event> queue;
   private final String apiKey;
+  private final LDConfig config;
   private final Consumer consumer;
 
   EventProcessor(String apiKey, LDConfig config) {
     this.apiKey = apiKey;
     this.queue = new ArrayBlockingQueue<>(config.capacity);
     this.consumer = new Consumer(config);
+    this.config = config;
     this.scheduler.scheduleAtFixedRate(consumer, 0, config.flushInterval, TimeUnit.SECONDS);
   }
 
   boolean sendEvent(Event e) {
+    if (config.samplingInterval > 0 && random.nextInt(config.samplingInterval) != 0) {
+      return true;
+    }
     return queue.offer(e);
   }
 
@@ -43,18 +56,8 @@ public void flush() {
     this.consumer.flush();
   }
 
-  static class DaemonThreadFactory implements ThreadFactory {
-    public Thread newThread(Runnable r) {
-      Thread thread = new Thread(r);
-      thread.setDaemon(true);
-      return thread;
-    }
-  }
-
   class Consumer implements Runnable {
     private final Logger logger = LoggerFactory.getLogger(Consumer.class);
-
-
     private final CloseableHttpClient client;
     private final LDConfig config;
 
@@ -78,6 +81,7 @@ public void flush() {
     }
 
     private void postEvents(List<Event> events) {
+      logger.debug("Posting " + events.size() + " event(s)..");
       CloseableHttpResponse response = null;
       Gson gson = new Gson();
       String json = gson.toJson(events);
@@ -95,13 +99,11 @@ private void postEvents(List<Event> events) {
         if (status >= 300) {
           if (status == HttpStatus.SC_UNAUTHORIZED) {
             logger.error("Invalid API key");
-          }
-          else {
+          } else {
             logger.error("Unexpected status code: " + status);
           }
-        }
-        else {
-          logger.debug("Successfully processed events");
+        } else {
+          logger.debug("Successfully posted " + events.size() + " event(s).");
         }
       } catch (IOException e) {
         logger.error("Unhandled exception in LaunchDarkly client attempting to connect to URI: " + config.eventsURI, e);
@@ -112,7 +114,6 @@ private void postEvents(List<Event> events) {
           logger.error("Unhandled exception in LaunchDarkly client", e);
         }
       }
-
     }
   }
 }

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

@@ -23,6 +23,7 @@ public final class LDConfig {
   private static final int DEFAULT_FLUSH_INTERVAL = 5;
   private static final long DEFAULT_POLLING_INTERVAL_MILLIS = 1000L;
   private static final long DEFAULT_START_WAIT_MILLIS = 0L;
+  private static final int DEFAULT_SAMPLING_INTERVAL = 0;
   private static final Logger logger = LoggerFactory.getLogger(LDConfig.class);
 
   protected static final LDConfig DEFAULT = new Builder().build();
@@ -42,6 +43,7 @@ public final class LDConfig {
   final boolean offline;
   final long pollingIntervalMillis;
   final long startWaitMillis;
+  final int samplingInterval;
 
   protected LDConfig(Builder builder) {
     this.baseURI = builder.baseURI;
@@ -63,6 +65,7 @@ protected LDConfig(Builder builder) {
       this.pollingIntervalMillis = builder.pollingIntervalMillis;
     }
     this.startWaitMillis = builder.startWaitMillis;
+    this.samplingInterval = builder.samplingInterval;
   }
 
   /**
@@ -94,6 +97,7 @@ public static class Builder {
     private long pollingIntervalMillis = DEFAULT_POLLING_INTERVAL_MILLIS;
     private FeatureStore featureStore = new InMemoryFeatureStore();
     public long startWaitMillis = DEFAULT_START_WAIT_MILLIS;
+    public int samplingInterval = DEFAULT_SAMPLING_INTERVAL;
 
     /**
      * Creates a builder with all configuration parameters set to the default
@@ -335,6 +339,21 @@ public Builder startWaitMillis(long startWaitMillis) {
       return this;
     }
 
+   /**
+     * Enable event sampling. When set to the default of zero, sampling is disabled and all events
+     * are sent back to LaunchDarkly. When set to greater than zero, there is a 1 in
+     * <code>samplingInterval</code> chance events will be will be sent.
+     *
+     * <p>Example: if you want 5% sampling rate, set <code>samplingInterval</code> to 20.
+     *
+     * @param samplingInterval the sampling interval.
+     * @return the builder
+     */
+    public Builder samplingInterval(int samplingInterval) {
+      this.samplingInterval = samplingInterval;
+      return this;
+    }
+
     HttpHost proxyHost() {
       if (this.proxyHost == null && this.proxyPort == -1 && this.proxyScheme == null) {
         return null;

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

@@ -1,13 +1,11 @@
 package com.launchdarkly.client;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 public class PollingProcessor implements UpdateProcessor {
@@ -40,7 +38,10 @@ public Future<Void> start() {
     logger.info("Starting LaunchDarkly polling client with interval: "
         + config.pollingIntervalMillis + " milliseconds");
     final VeryBasicFuture initFuture = new VeryBasicFuture();
-    scheduler = Executors.newScheduledThreadPool(1);
+    ThreadFactory threadFactory = new ThreadFactoryBuilder()
+        .setNameFormat("LaunchDarkly-PollingProcessor-%d")
+        .build();
+    scheduler = Executors.newScheduledThreadPool(1, threadFactory);
 
     scheduler.scheduleAtFixedRate(new Runnable() {
       @Override