NettyMessagingService implements ManagedMessagingService { @Override public CompletableFuture<byte[]> sendAndReceive( final Address address, final String type, final byte[] payload, final boolean keepAlive) { return sendAndReceive(address, type, payload, keepAlive, null, MoreExecutors.directExecutor()); } NettyMessagingService(
      final String cluster, final Address address, final MessagingConfig config);   NettyMessagingService(
      final String cluster,
      final Address address,
      final MessagingConfig config,
      final ProtocolVersion protocolVersion); @Override Address address(); @Override CompletableFuture<Void> sendAsync(
      final Address address, final String type, final byte[] payload, final boolean keepAlive); @Override CompletableFuture<byte[]> sendAndReceive(
      final Address address, final String type, final byte[] payload, final boolean keepAlive); @Override CompletableFuture<byte[]> sendAndReceive(
      final Address address,
      final String type,
      final byte[] payload,
      final boolean keepAlive,
      final Executor executor); @Override CompletableFuture<byte[]> sendAndReceive(
      final Address address,
      final String type,
      final byte[] payload,
      final boolean keepAlive,
      final Duration timeout); @Override CompletableFuture<byte[]> sendAndReceive(
      final Address address,
      final String type,
      final byte[] payload,
      final boolean keepAlive,
      final Duration timeout,
      final Executor executor); @Override void registerHandler(
      final String type, final BiConsumer<Address, byte[]> handler, final Executor executor); @Override void registerHandler(
      final String type,
      final BiFunction<Address, byte[], byte[]> handler,
      final Executor executor); @Override void registerHandler(
      final String type, final BiFunction<Address, byte[], CompletableFuture<byte[]>> handler); @Override void unregisterHandler(final String type); @Override CompletableFuture<MessagingService> start(); @Override boolean isRunning(); @Override CompletableFuture<Void> stop();  }
@Test public void testSendAndReceive() { final String subject = nextSubject(); final AtomicBoolean handlerInvoked = new AtomicBoolean(false); final AtomicReference<byte[]> request = new AtomicReference<>(); final AtomicReference<Address> sender = new AtomicReference<>(); final BiFunction<Address, byte[], byte[]> handler = (ep, data) -> { handlerInvoked.set(true); sender.set(ep); request.set(data); return "hello there".getBytes(); }; netty2.registerHandler(subject, handler, MoreExecutors.directExecutor()); final CompletableFuture<byte[]> response = netty1.sendAndReceive(address2, subject, "hello world".getBytes()); assertTrue(Arrays.equals("hello there".getBytes(), response.join())); assertTrue(handlerInvoked.get()); assertTrue(Arrays.equals(request.get(), "hello world".getBytes())); assertEquals(address1.address(), sender.get().address()); }