NettyMessagingService implements ManagedMessagingService { @Override public CompletableFuture<Void> sendAsync( final Address address, final String type, final byte[] payload, final boolean keepAlive) { final long messageId = messageIdGenerator.incrementAndGet(); final ProtocolRequest message = new ProtocolRequest(messageId, returnAddress, type, payload); return executeOnPooledConnection( address, type, c -> c.sendAsync(message), 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 testSendAsyncToUnresolvable() { final Address unresolvable = Address.from("unknown.local", address1.port()); final String subject = nextSubject(); final CompletableFuture<Void> response = netty1.sendAsync(unresolvable, subject, "hello world".getBytes()); assertTrue(response.isCompletedExceptionally()); }
@Test public void testSendAsync() { final String subject = nextSubject(); final CountDownLatch latch1 = new CountDownLatch(1); CompletableFuture<Void> response = netty1.sendAsync(address2, subject, "hello world".getBytes()); response.whenComplete( (r, e) -> { assertNull(e); latch1.countDown(); }); Uninterruptibles.awaitUninterruptibly(latch1); final CountDownLatch latch2 = new CountDownLatch(1); response = netty1.sendAsync(invalidAddress, subject, "hello world".getBytes()); response.whenComplete( (r, e) -> { assertNotNull(e); assertTrue(e instanceof ConnectException); latch2.countDown(); }); Uninterruptibles.awaitUninterruptibly(latch2); }