NettyUnicastService implements ManagedUnicastService { @Override public void unicast(final Address address, final String subject, final byte[] payload) { if (!started.get()) { LOGGER.debug("Failed sending unicast message, unicast service was not started."); return; } final InetAddress resolvedAddress = address.address(); if (resolvedAddress == null) { LOGGER.debug( "Failed sending unicast message (destination address {} cannot be resolved)", address); return; } final Message message = new Message(this.address, subject, payload); final byte[] bytes = SERIALIZER.encode(message); final ByteBuf buf = channel.alloc().buffer(4 + bytes.length); buf.writeInt(bytes.length).writeBytes(bytes); channel.writeAndFlush( new DatagramPacket(buf, new InetSocketAddress(resolvedAddress, address.port()))); } NettyUnicastService(final Address address, final MessagingConfig config); @Override void unicast(final Address address, final String subject, final byte[] payload); @Override synchronized void addListener(
      final String subject, final BiConsumer<Address, byte[]> listener, final Executor executor); @Override synchronized void removeListener(
      final String subject, final BiConsumer<Address, byte[]> listener); @Override CompletableFuture<UnicastService> start(); @Override boolean isRunning(); @Override CompletableFuture<Void> stop();  }
@Test public void testUnicast() throws Exception { service1.addListener( "test", (address, payload) -> { assertEquals(address2, address); assertArrayEquals("Hello world!".getBytes(), payload); resume(); }); service2.unicast(address1, "test", "Hello world!".getBytes()); await(5000); }