server: allow multiple callbacks to be provided (#72)

* server: allow multiple callbacks to be provided

This will make it reasonable to provide open source callbacks that can
be used alongside user defined callbacks.

Signed-off-by: Snow Pettersen <snowp@squareup.com>

Author: snowp

server/src/main/java/io/envoyproxy/controlplane/server/DiscoveryServer.java

@@ -2,6 +2,7 @@
 
 import static com.google.common.base.Strings.isNullOrEmpty;
 
+import com.google.common.base.Preconditions;
 import com.google.protobuf.Any;
 import envoy.api.v2.ClusterDiscoveryServiceGrpc.ClusterDiscoveryServiceImplBase;
 import envoy.api.v2.Discovery.DiscoveryRequest;
@@ -18,6 +19,7 @@
 import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -28,22 +30,33 @@
 
 public class DiscoveryServer {
 
-  private static final DiscoveryServerCallbacks DEFAULT_CALLBACKS = new DiscoveryServerCallbacks() { };
   private static final Logger LOGGER = LoggerFactory.getLogger(DiscoveryServer.class);
 
   static final String ANY_TYPE_URL = "";
 
-  private final DiscoveryServerCallbacks callbacks;
+  private final List<DiscoveryServerCallbacks> callbacks;
   private final ConfigWatcher configWatcher;
   private final AtomicLong streamCount = new AtomicLong();
 
+  public DiscoveryServer(ConfigWatcher configWatcher) {
+    this(Collections.emptyList(), configWatcher);
+  }
+
   public DiscoveryServer(DiscoveryServerCallbacks callbacks, ConfigWatcher configWatcher) {
-    this.callbacks = callbacks;
-    this.configWatcher = configWatcher;
+    this(Collections.singletonList(callbacks), configWatcher);
   }
 
-  public DiscoveryServer(ConfigWatcher configWatcher) {
-    this(DEFAULT_CALLBACKS, configWatcher);
+  /**
+   * Creates the server.
+   * @param callbacks server callbacks
+   * @param configWatcher source of configuration updates
+   */
+  public DiscoveryServer(List<DiscoveryServerCallbacks> callbacks, ConfigWatcher configWatcher) {
+    Preconditions.checkNotNull(callbacks, "callbacks cannot be null");
+    Preconditions.checkNotNull(configWatcher, "configWatcher cannot be null");
+
+    this.callbacks = callbacks;
+    this.configWatcher = configWatcher;
   }
 
   /**
@@ -137,7 +150,7 @@ private StreamObserver<DiscoveryRequest> createRequestHandler(
 
     LOGGER.info("[{}] open stream from {}", streamId, defaultTypeUrl);
 
-    callbacks.onStreamOpen(streamId, defaultTypeUrl);
+    callbacks.forEach(cb -> cb.onStreamOpen(streamId, defaultTypeUrl));
 
     return new StreamObserver<DiscoveryRequest>() {
 
@@ -173,7 +186,7 @@ public void onNext(DiscoveryRequest request) {
             nonce,
             request.getVersionInfo());
 
-        callbacks.onStreamRequest(streamId, request);
+        callbacks.forEach(cb -> cb.onStreamRequest(streamId, request));
 
         for (String typeUrl : Resources.TYPE_URLS) {
           DiscoveryResponse response = latestResponse.get(typeUrl);
@@ -213,7 +226,7 @@ public void onError(Throwable t) {
         }
 
         try {
-          callbacks.onStreamCloseWithError(streamId, defaultTypeUrl, t);
+          callbacks.forEach(cb -> cb.onStreamCloseWithError(streamId, defaultTypeUrl, t));
           responseObserver.onError(Status.fromThrowable(t).asException());
         } finally {
           cancel();
@@ -225,7 +238,7 @@ public void onCompleted() {
         LOGGER.info("[{}] stream closed", streamId);
 
         try {
-          callbacks.onStreamClose(streamId, defaultTypeUrl);
+          callbacks.forEach(cb -> cb.onStreamClose(streamId, defaultTypeUrl));
           responseObserver.onCompleted();
         } finally {
           cancel();
@@ -248,7 +261,7 @@ private void send(Response response, String typeUrl) {
 
         LOGGER.info("[{}] response {} with nonce {} version {}", streamId, typeUrl, nonce, response.version());
 
-        callbacks.onStreamResponse(streamId, response.request(), discoveryResponse);
+        callbacks.forEach(cb -> cb.onStreamResponse(streamId, response.request(), discoveryResponse));
 
         // Store the latest response *before* we send the response. This ensures that by the time the request
         // is processed the map is guaranteed to be updated. Doing it afterwards leads to a race conditions