NettyBroadcastService implements ManagedBroadcastService { @Override public void broadcast(final String subject, final byte[] payload) { if (enabled) { final Message message = new Message(subject, payload); final byte[] bytes = SERIALIZER.encode(message); final ByteBuf buf = serverChannel.alloc().buffer(4 + bytes.length); buf.writeInt(bytes.length).writeBytes(bytes); serverChannel.writeAndFlush(new DatagramPacket(buf, groupAddress)); } } NettyBroadcastService(
      final Address localAddress, final Address groupAddress, final boolean enabled); static Builder builder(); @Override void broadcast(final String subject, final byte[] payload); @Override synchronized void addListener(final String subject, final Consumer<byte[]> listener); @Override synchronized void removeListener(final String subject, final Consumer<byte[]> listener); @Override CompletableFuture<BroadcastService> start(); @Override boolean isRunning(); @Override CompletableFuture<Void> stop();  }
@Test public void testBroadcast() throws Exception { netty1.addListener( "test", bytes -> { threadAssertEquals(0, bytes.length); resume(); }); netty2.broadcast("test", new byte[0]); await(5000); }