UdpTransport extends AbstractUdpTransport { public CompletableFuture<? extends Transport> open(BufferSupplier buffers, TransportConsumer consumer) { Objects.requireNonNull(buffers); Objects.requireNonNull(consumer); this.buffers = buffers; this.consumer = consumer; CompletableFuture<UdpTransport> future = new CompletableFuture<>(); try { this.socketChannel = DatagramChannel.open(); register(SelectionKey.OP_READ | SelectionKey.OP_WRITE); this.socketChannel.bind(localAddress); this.socketChannel.connect(remoteAddress); if (dispatcher != null) { dispatcher.addTransport(this); } if (buffers instanceof Service) { ((Service) buffers).open().get(); } future.complete(this); consumer.connected(); } catch (IOException | InterruptedException ex) { future.completeExceptionally(ex); try { socketChannel.close(); } catch (IOException e) { } } catch (ExecutionException ex) { future.completeExceptionally(ex.getCause()); try { socketChannel.close(); } catch (IOException e) { } } return future; } UdpTransport(Dispatcher dispatcher, SocketAddress localAddress,
      SocketAddress remoteAddress);  UdpTransport(Selector selector, SocketAddress localAddress, SocketAddress remoteAddress); CompletableFuture<? extends Transport> open(BufferSupplier buffers,
      TransportConsumer consumer);  }
@Test public void unicast() throws IOException, InterruptedException, ExecutionException, TimeoutException { final TestReceiver serverReceiver = new TestReceiver(); serverTransport = new UdpTransport(iOReactor.getSelector(), serverAddress, clientAddress); BufferedTransportConsumer serverBuffers = new BufferedTransportConsumer(Executors.defaultThreadFactory(), serverReceiver); serverTransport.open( serverBuffers, serverBuffers); clientTransport = new UdpTransport(iOReactor.getSelector(), clientAddress, serverAddress); TransportConsumer clientReceiver = new TransportConsumer() { @Override public void accept(ByteBuffer t) { } @Override public void connected() { } @Override public void disconnected() { } }; BufferedTransportConsumer clientBuffers = new BufferedTransportConsumer(Executors.defaultThreadFactory(), clientReceiver); clientTransport.open( clientBuffers, clientBuffers).get(1000L, TimeUnit.MILLISECONDS); ByteBuffer buf = ByteBuffer.allocate(8096).order(ByteOrder.nativeOrder()); int totalBytesSent = 0; for (int i = 0; i < messageCount; ++i) { buf.clear(); buf.put(messages[i], 0, messages[i].length); int bytesSent = clientTransport.write(buf); assertEquals(messages[i].length, bytesSent); totalBytesSent += bytesSent; } try { Thread.sleep(1000); } catch (InterruptedException e) { } assertEquals(totalBytesSent, serverReceiver.getBytesReceived()); }
@Test public void multicast() throws IOException, InterruptedException, ExecutionException, TimeoutException { NetworkInterface networkInterface = getMulticastInterface(); if (networkInterface == null) { fail("No available multicast interface"); } InetAddress multicastAddress = InetAddress.getByName("224.2.2.3"); int port = 5555; final TestReceiver serverReceiver = new TestReceiver(); serverTransport = new UdpMulticastTransport(iOReactor.getSelector(), multicastAddress, port, networkInterface); BufferedTransportConsumer serverBuffers = new BufferedTransportConsumer(Executors.defaultThreadFactory(), serverReceiver); serverTransport.open( serverBuffers, serverBuffers); clientTransport = new UdpMulticastTransport(iOReactor.getSelector(), multicastAddress, port, networkInterface); TransportConsumer clientReceiver = new TransportConsumer() { @Override public void accept(ByteBuffer t) { } @Override public void connected() { } @Override public void disconnected() { } }; BufferedTransportConsumer clientBuffers = new BufferedTransportConsumer(Executors.defaultThreadFactory(), clientReceiver); clientTransport.open( clientBuffers, clientBuffers).get(1000L, TimeUnit.MILLISECONDS); ByteBuffer buf = ByteBuffer.allocate(8096).order(ByteOrder.nativeOrder()); int totalBytesSent = 0; for (int i = 0; i < messageCount; ++i) { buf.clear(); buf.put(messages[i], 0, messages[i].length); int bytesSent = clientTransport.write(buf); assertEquals(messages[i].length, bytesSent); totalBytesSent += bytesSent; } try { Thread.sleep(500); } catch (InterruptedException e) { } assertEquals(totalBytesSent, serverReceiver.getBytesReceived()); }