Only support blocking aware proxies when no executor is explicitly pr…

…ovided.

protocols/raft/src/main/java/io/atomix/protocols/raft/proxy/impl/BlockingAwareRaftProxy.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2017-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.atomix.protocols.raft.proxy.impl;
+
+import com.google.common.collect.Maps;
+import io.atomix.protocols.raft.RaftCommand;
+import io.atomix.protocols.raft.RaftQuery;
+import io.atomix.protocols.raft.proxy.RaftProxy;
+import io.atomix.protocols.raft.proxy.RaftProxyDelegate;
+import io.atomix.utils.concurrent.Futures;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.function.Consumer;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Raft proxy delegate that completes futures on a thread pool.
+ */
+public class BlockingAwareRaftProxy extends RaftProxyDelegate {
+  private final Executor orderedExecutor;
+  private final Executor threadPoolExecutor;
+  private final Map<Consumer, Consumer> listenerMap = Maps.newConcurrentMap();
+
+  public BlockingAwareRaftProxy(RaftProxy delegate, Executor orderedExecutor, Executor threadPoolExecutor) {
+    super(delegate);
+    this.orderedExecutor = checkNotNull(orderedExecutor, "orderedExecutor cannot be null");
+    this.threadPoolExecutor = checkNotNull(threadPoolExecutor, "threadPoolExecutor cannot be null");
+  }
+
+  @Override
+  public <T> void addEventListener(Consumer<T> listener) {
+    Consumer<T> wrappedListener = event -> orderedExecutor.execute(() -> listener.accept(event));
+    listenerMap.put(listener, wrappedListener);
+    super.addEventListener(wrappedListener);
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public <T> void removeEventListener(Consumer<T> listener) {
+    Consumer<T> wrappedListener = listenerMap.remove(listener);
+    if (wrappedListener != null) {
+      super.removeEventListener(wrappedListener);
+    }
+  }
+
+  @Override
+  public <T> CompletableFuture<T> submit(RaftCommand<T> command) {
+    return Futures.blockingAwareFuture(super.submit(command), orderedExecutor, threadPoolExecutor);
+  }
+
+  @Override
+  public <T> CompletableFuture<T> submit(RaftQuery<T> query) {
+    return Futures.blockingAwareFuture(super.submit(query), orderedExecutor, threadPoolExecutor);
+  }
+}

protocols/raft/src/main/java/io/atomix/protocols/raft/proxy/impl/ExecutingRaftProxy.java

@@ -15,35 +15,53 @@
  */
 package io.atomix.protocols.raft.proxy.impl;
 
+import com.google.common.collect.Maps;
 import io.atomix.protocols.raft.RaftCommand;
 import io.atomix.protocols.raft.RaftQuery;
 import io.atomix.protocols.raft.proxy.RaftProxy;
 import io.atomix.protocols.raft.proxy.RaftProxyDelegate;
 import io.atomix.utils.concurrent.Futures;
 
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
+import java.util.function.Consumer;
 
 /**
- * Raft proxy delegate that completes futures on a thread pool.
+ * Raft proxy that completes all events on an executor.
  */
 public class ExecutingRaftProxy extends RaftProxyDelegate {
-  private final Executor orderedExecutor;
-  private final Executor threadPoolExecutor;
+  private final Executor executor;
+  private final Map<Consumer, Consumer> listenerMap = Maps.newConcurrentMap();
 
-  public ExecutingRaftProxy(RaftProxy delegate, Executor orderedExecutor, Executor threadPoolExecutor) {
+  public ExecutingRaftProxy(RaftProxy delegate, Executor executor) {
     super(delegate);
-    this.orderedExecutor = orderedExecutor;
-    this.threadPoolExecutor = threadPoolExecutor;
+    this.executor = executor;
+  }
+
+  @Override
+  public <T> void addEventListener(Consumer<T> listener) {
+    Consumer<T> wrappedListener = event -> executor.execute(() -> listener.accept(event));
+    listenerMap.put(listener, wrappedListener);
+    super.addEventListener(wrappedListener);
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public <T> void removeEventListener(Consumer<T> listener) {
+    Consumer<T> wrappedListener = listenerMap.remove(listener);
+    if (wrappedListener != null) {
+      super.removeEventListener(wrappedListener);
+    }
   }
 
   @Override
   public <T> CompletableFuture<T> submit(RaftCommand<T> command) {
-    return Futures.blockingAwareFuture(super.submit(command), orderedExecutor, threadPoolExecutor);
+    return Futures.asyncFuture(super.submit(command), executor);
   }
 
   @Override
   public <T> CompletableFuture<T> submit(RaftQuery<T> query) {
-    return Futures.blockingAwareFuture(super.submit(query), orderedExecutor, threadPoolExecutor);
+    return Futures.asyncFuture(super.submit(query), executor);
   }
 }

protocols/raft/src/main/java/io/atomix/protocols/raft/proxy/impl/RaftProxyManager.java

@@ -157,7 +157,12 @@ public CompletableFuture<RaftProxy> openSession(
               communicationStrategy,
               serializer,
               proxyContext);
-          proxy = new ExecutingRaftProxy(proxy, executor, threadPoolExecutor);
+
+          if (executor != null) {
+            proxy = new ExecutingRaftProxy(proxy, executor);
+          } else {
+            proxy = new BlockingAwareRaftProxy(proxy, new ThreadPoolContext(threadPoolExecutor), threadPoolExecutor);
+          }
           future.complete(proxy);
         } else {
           future.completeExceptionally(response.error().createException());

utils/src/main/java/io/atomix/utils/concurrent/Futures.java

@@ -85,6 +85,28 @@ public static <T> CompletableFuture<T> orderedFuture() {
     return new OrderedFuture<>();
   }
 
+  /**
+   * Returns a wrapped future that will be completed on the given executor.
+   *
+   * @param future the future to be completed on the given executor
+   * @param executor the executor with which to complete the future
+   * @param <T> the future value type
+   * @return a wrapped future to be completed on the given executor
+   */
+  public static <T> CompletableFuture<T> asyncFuture(CompletableFuture<T> future, Executor executor) {
+    CompletableFuture<T> newFuture = new CompletableFuture<>();
+    future.whenComplete((result, error) -> {
+      executor.execute(() -> {
+        if (error == null) {
+          newFuture.complete(result);
+        } else {
+          newFuture.completeExceptionally(error);
+        }
+      });
+    });
+    return newFuture;
+  }
+
   /**
    * Returns a future that's completed using the given {@code orderedExecutor} if the future is not blocked or the
    * given {@code threadPoolExecutor} if the future is blocked.