Madan Jampani
Committed by Gerrit Code Review

Use the Executor interface when specifying where to handle incoming messages

This is done so that one can simply specify a direct executor.
Change-Id: I1c3ea977dd7c2d604588d587fd67f7012355eedf
...@@ -17,6 +17,7 @@ package org.onosproject.store.cluster.messaging; ...@@ -17,6 +17,7 @@ package org.onosproject.store.cluster.messaging;
17 17
18 import java.util.Set; 18 import java.util.Set;
19 import java.util.concurrent.CompletableFuture; 19 import java.util.concurrent.CompletableFuture;
20 +import java.util.concurrent.Executor;
20 import java.util.concurrent.ExecutorService; 21 import java.util.concurrent.ExecutorService;
21 import java.util.function.Consumer; 22 import java.util.function.Consumer;
22 import java.util.function.Function; 23 import java.util.function.Function;
...@@ -173,7 +174,7 @@ public interface ClusterCommunicationService { ...@@ -173,7 +174,7 @@ public interface ClusterCommunicationService {
173 Function<byte[], M> decoder, 174 Function<byte[], M> decoder,
174 Function<M, R> handler, 175 Function<M, R> handler,
175 Function<R, byte[]> encoder, 176 Function<R, byte[]> encoder,
176 - ExecutorService executor); 177 + Executor executor);
177 178
178 /** 179 /**
179 * Adds a new subscriber for the specified message subject. 180 * Adds a new subscriber for the specified message subject.
...@@ -187,7 +188,7 @@ public interface ClusterCommunicationService { ...@@ -187,7 +188,7 @@ public interface ClusterCommunicationService {
187 <M> void addSubscriber(MessageSubject subject, 188 <M> void addSubscriber(MessageSubject subject,
188 Function<byte[], M> decoder, 189 Function<byte[], M> decoder,
189 Consumer<M> handler, 190 Consumer<M> handler,
190 - ExecutorService executor); 191 + Executor executor);
191 192
192 /** 193 /**
193 * Removes a subscriber for the specified message subject. 194 * Removes a subscriber for the specified message subject.
......
...@@ -43,6 +43,7 @@ import com.google.common.util.concurrent.SettableFuture; ...@@ -43,6 +43,7 @@ import com.google.common.util.concurrent.SettableFuture;
43 import java.io.IOException; 43 import java.io.IOException;
44 import java.util.Set; 44 import java.util.Set;
45 import java.util.concurrent.CompletableFuture; 45 import java.util.concurrent.CompletableFuture;
46 +import java.util.concurrent.Executor;
46 import java.util.concurrent.ExecutorService; 47 import java.util.concurrent.ExecutorService;
47 import java.util.function.Consumer; 48 import java.util.function.Consumer;
48 import java.util.function.Function; 49 import java.util.function.Function;
...@@ -270,7 +271,7 @@ public class ClusterCommunicationManager ...@@ -270,7 +271,7 @@ public class ClusterCommunicationManager
270 Function<byte[], M> decoder, 271 Function<byte[], M> decoder,
271 Function<M, R> handler, 272 Function<M, R> handler,
272 Function<R, byte[]> encoder, 273 Function<R, byte[]> encoder,
273 - ExecutorService executor) { 274 + Executor executor) {
274 messagingService.registerHandler(subject.value(), 275 messagingService.registerHandler(subject.value(),
275 new InternalMessageResponder<>(decoder, encoder, handler), 276 new InternalMessageResponder<>(decoder, encoder, handler),
276 executor); 277 executor);
...@@ -280,7 +281,7 @@ public class ClusterCommunicationManager ...@@ -280,7 +281,7 @@ public class ClusterCommunicationManager
280 public <M> void addSubscriber(MessageSubject subject, 281 public <M> void addSubscriber(MessageSubject subject,
281 Function<byte[], M> decoder, 282 Function<byte[], M> decoder,
282 Consumer<M> handler, 283 Consumer<M> handler,
283 - ExecutorService executor) { 284 + Executor executor) {
284 messagingService.registerHandler(subject.value(), 285 messagingService.registerHandler(subject.value(),
285 new InternalMessageConsumer<>(decoder, handler), 286 new InternalMessageConsumer<>(decoder, handler),
286 executor); 287 executor);
......
...@@ -54,6 +54,7 @@ import java.util.Objects; ...@@ -54,6 +54,7 @@ import java.util.Objects;
54 import java.util.Set; 54 import java.util.Set;
55 import java.util.concurrent.CompletableFuture; 55 import java.util.concurrent.CompletableFuture;
56 import java.util.concurrent.CountDownLatch; 56 import java.util.concurrent.CountDownLatch;
57 +import java.util.concurrent.Executor;
57 import java.util.concurrent.ExecutorService; 58 import java.util.concurrent.ExecutorService;
58 import java.util.concurrent.TimeUnit; 59 import java.util.concurrent.TimeUnit;
59 import java.util.concurrent.atomic.AtomicLong; 60 import java.util.concurrent.atomic.AtomicLong;
...@@ -780,13 +781,13 @@ public class EventuallyConsistentMapImplTest { ...@@ -780,13 +781,13 @@ public class EventuallyConsistentMapImplTest {
780 @Override 781 @Override
781 public <M, R> void addSubscriber(MessageSubject subject, 782 public <M, R> void addSubscriber(MessageSubject subject,
782 Function<byte[], M> decoder, Function<M, R> handler, 783 Function<byte[], M> decoder, Function<M, R> handler,
783 - Function<R, byte[]> encoder, ExecutorService executor) { 784 + Function<R, byte[]> encoder, Executor executor) {
784 } 785 }
785 786
786 @Override 787 @Override
787 public <M> void addSubscriber(MessageSubject subject, 788 public <M> void addSubscriber(MessageSubject subject,
788 Function<byte[], M> decoder, Consumer<M> handler, 789 Function<byte[], M> decoder, Consumer<M> handler,
789 - ExecutorService executor) { 790 + Executor executor) {
790 } 791 }
791 792
792 @Override 793 @Override
......
...@@ -17,7 +17,7 @@ package org.onlab.netty; ...@@ -17,7 +17,7 @@ package org.onlab.netty;
17 17
18 import java.io.IOException; 18 import java.io.IOException;
19 import java.util.concurrent.CompletableFuture; 19 import java.util.concurrent.CompletableFuture;
20 -import java.util.concurrent.ExecutorService; 20 +import java.util.concurrent.Executor;
21 21
22 /** 22 /**
23 * Interface for low level messaging primitives. 23 * Interface for low level messaging primitives.
...@@ -48,7 +48,7 @@ public interface MessagingService { ...@@ -48,7 +48,7 @@ public interface MessagingService {
48 * @param handler message handler 48 * @param handler message handler
49 * @param executor executor to use for running message handler logic. 49 * @param executor executor to use for running message handler logic.
50 */ 50 */
51 - public void registerHandler(String type, MessageHandler handler, ExecutorService executor); 51 + public void registerHandler(String type, MessageHandler handler, Executor executor);
52 52
53 /** 53 /**
54 * Registers a new message handler for message type. 54 * Registers a new message handler for message type.
......
...@@ -42,7 +42,7 @@ import java.net.UnknownHostException; ...@@ -42,7 +42,7 @@ import java.net.UnknownHostException;
42 import java.util.concurrent.CompletableFuture; 42 import java.util.concurrent.CompletableFuture;
43 import java.util.concurrent.ConcurrentHashMap; 43 import java.util.concurrent.ConcurrentHashMap;
44 import java.util.concurrent.ConcurrentMap; 44 import java.util.concurrent.ConcurrentMap;
45 -import java.util.concurrent.ExecutorService; 45 +import java.util.concurrent.Executor;
46 import java.util.concurrent.TimeUnit; 46 import java.util.concurrent.TimeUnit;
47 import java.util.concurrent.TimeoutException; 47 import java.util.concurrent.TimeoutException;
48 import java.util.concurrent.atomic.AtomicLong; 48 import java.util.concurrent.atomic.AtomicLong;
...@@ -202,11 +202,11 @@ public class NettyMessagingService implements MessagingService { ...@@ -202,11 +202,11 @@ public class NettyMessagingService implements MessagingService {
202 } 202 }
203 203
204 @Override 204 @Override
205 - public void registerHandler(String type, MessageHandler handler, ExecutorService executor) { 205 + public void registerHandler(String type, MessageHandler handler, Executor executor) {
206 handlers.put(type, new MessageHandler() { 206 handlers.put(type, new MessageHandler() {
207 @Override 207 @Override
208 public void handle(Message message) throws IOException { 208 public void handle(Message message) throws IOException {
209 - executor.submit(() -> { 209 + executor.execute(() -> {
210 try { 210 try {
211 handler.handle(message); 211 handler.handle(message);
212 } catch (Exception e) { 212 } catch (Exception e) {
......