### Question:
NettyUnicastService implements ManagedUnicastService { @Override public void unicast(Address address, String subject, byte[] payload) { final InetAddress resolvedAddress = address.address(); if (resolvedAddress == null) { LOGGER.debug("Failed sending unicast message (destination address {} cannot be resolved)", address); return; } Message message = new Message(this.address, subject, payload); byte[] bytes = SERIALIZER.encode(message); 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(Address address, MessagingConfig config); @Override void unicast(Address address, String subject, byte[] payload); @Override synchronized void addListener(String subject, BiConsumer<Address, byte[]> listener, Executor executor); @Override synchronized void removeListener(String subject, BiConsumer<Address, byte[]> listener); @Override CompletableFuture<UnicastService> start(); @Override boolean isRunning(); @Override CompletableFuture<Void> stop();  }### Answer:
@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); }