@Test public void testSendAndReceive() { AtomicBoolean handlerInvoked = new AtomicBoolean(false); AtomicReference<byte[]> request = new AtomicReference<>(); AtomicReference<Endpoint> sender = new AtomicReference<>(); BiFunction<Endpoint, byte[], byte[]> handler = (ep, data) -> { handlerInvoked.set(true); sender.set(ep); request.set(data); return "hello there".getBytes(); }; netty2.registerHandler("test-subject", handler, MoreExecutors.directExecutor()); CompletableFuture<byte[]> response = netty1.sendAndReceive(ep2, "test-subject", "hello world".getBytes()); assertTrue(Arrays.equals("hello there".getBytes(), response.join())); assertTrue(handlerInvoked.get()); assertTrue(Arrays.equals(request.get(), "hello world".getBytes())); assertEquals(ep1, sender.get()); }
@Override public CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload) { checkPermission(CLUSTER_WRITE); return sendAndReceive(ep, type, payload, MoreExecutors.directExecutor()); }
NettyMessagingManager implements MessagingService { @Override public CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload) { checkPermission(CLUSTER_WRITE); return sendAndReceive(ep, type, payload, MoreExecutors.directExecutor()); } }
NettyMessagingManager implements MessagingService { @Override public CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload) { checkPermission(CLUSTER_WRITE); return sendAndReceive(ep, type, payload, MoreExecutors.directExecutor()); }  }
NettyMessagingManager implements MessagingService { @Override public CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload) { checkPermission(CLUSTER_WRITE); return sendAndReceive(ep, type, payload, MoreExecutors.directExecutor()); }  @Activate void activate(); @Deactivate void deactivate(); @Override CompletableFuture<Void> sendAsync(Endpoint ep, String type, byte[] payload); @Override CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload); @Override CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload, Executor executor); @Override void registerHandler(String type, BiConsumer<Endpoint, byte[]> handler, Executor executor); @Override void registerHandler(String type, BiFunction<Endpoint, byte[], byte[]> handler, Executor executor); @Override void registerHandler(String type, BiFunction<Endpoint, byte[], CompletableFuture<byte[]>> handler); @Override void unregisterHandler(String type); }
NettyMessagingManager implements MessagingService { @Override public CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload) { checkPermission(CLUSTER_WRITE); return sendAndReceive(ep, type, payload, MoreExecutors.directExecutor()); }  @Activate void activate(); @Deactivate void deactivate(); @Override CompletableFuture<Void> sendAsync(Endpoint ep, String type, byte[] payload); @Override CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload); @Override CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload, Executor executor); @Override void registerHandler(String type, BiConsumer<Endpoint, byte[]> handler, Executor executor); @Override void registerHandler(String type, BiFunction<Endpoint, byte[], byte[]> handler, Executor executor); @Override void registerHandler(String type, BiFunction<Endpoint, byte[], CompletableFuture<byte[]>> handler); @Override void unregisterHandler(String type);  }