@@ -0,0 +1,3610 @@
+package rx.broadcast.integration.pp;%0A%0Aimport org.junit.Test;%0Aimport rx.Observable;%0Aimport rx.broadcast.Broadcast;%0Aimport rx.broadcast.SingleSourceFifoOrder;%0Aimport rx.broadcast.UdpBroadcast;%0Aimport rx.observers.TestSubscriber;%0A%0Aimport java.net.DatagramSocket;%0Aimport java.net.InetAddress;%0Aimport java.net.SocketException;%0Aimport java.net.UnknownHostException;%0Aimport java.util.concurrent.TimeUnit;%0A%0Apublic class PingPongUdpSingleSourceFifoOrder %7B%0A    private static final int MESSAGE_COUNT = 100;%0A%0A    private static final long TIMEOUT = 30;%0A%0A    /**%0A     * Receive a PING and respond with a PONG.%0A     * @throws SocketException if the socket could not be opened, or the socket could not bind to the given port.%0A     * @throws UnknownHostException if no IP address for the host machine could be found.%0A     */%0A    @Test%0A    public final void recv() throws SocketException, UnknownHostException %7B%0A        final int port = Integer.valueOf(System.getProperty(%22port%22));%0A        final DatagramSocket socket = new DatagramSocket(port);%0A        final InetAddress destination = InetAddress.getByName(System.getProperty(%22destination%22));%0A        final Broadcast broadcast = new UdpBroadcast%3C%3E(socket, destination, port, new SingleSourceFifoOrder%3C%3E());%0A%0A        final TestSubscriber%3CPing%3E subscriber = new TestSubscriber%3C%3E();%0A%0A        broadcast.valuesOfType(Ping.class)%0A            .doOnNext(System.out::println)%0A            .concatMap(ping -%3E%0A                broadcast.send(new Pong(ping.value))%0A                    // Once we've sent the response, we can emit the PING value to the subscriber.%0A                    // The cast here is a hack to allow us to concatenate a PING onto the stream.%0A                    // Where this is an %60Observable%3CVoid%3E%60 we know we won't get anything that needs to be casted.%0A                    .cast(Ping.class)%0A                    .concatWith(Observable.just(ping))%0A                    .doOnCompleted(() -%3E System.out.println(%22Sent PONG%22)))%0A            .take(MESSAGE_COUNT)%0A            .subscribe(subscriber);%0A%0A        subscriber.awaitTerminalEventAndUnsubscribeOnTimeout(TIMEOUT, TimeUnit.SECONDS);%0A        subscriber.assertNoErrors();%0A        subscriber.assertValueCount(MESSAGE_COUNT);%0A    %7D%0A%0A    /**%0A     * Send a set PING messages to the receiver, expecting PONG messages in response.%0A     * @param args the command line arguments passed to the program%0A     * @throws SocketException if the socket could not be opened, or the socket could not bind to the given port.%0A     * @throws UnknownHostException if no IP address for the host machine could be found.%0A     */%0A    public static void main(final String%5B%5D args) throws InterruptedException, SocketException, UnknownHostException %7B%0A        final int port = Integer.valueOf(System.getProperty(%22port%22));%0A        final DatagramSocket socket = new DatagramSocket(port);%0A        final InetAddress destination = InetAddress.getByName(System.getProperty(%22destination%22));%0A        final Broadcast broadcast = new UdpBroadcast%3C%3E(socket, destination, port, new SingleSourceFifoOrder%3C%3E());%0A%0A        Observable.range(1, MESSAGE_COUNT)%0A            .map(Ping::new)%0A            .doOnNext(System.out::println)%0A            .concatMap(value -%3E%0A                broadcast.send(value)%0A                    .doOnCompleted(() -%3E System.out.printf(%22Sent %25s%25n%22, value))%0A                    .cast(Pong.class)%0A                    .concatWith(broadcast.valuesOfType(Pong.class).first()))%0A            .timeout(TIMEOUT, TimeUnit.SECONDS)%0A            .toBlocking()%0A            .subscribe(pong -%3E%0A                System.out.printf(%22Received %25s%25n%22, pong));%0A    %7D%0A%7D%0A

Add ping pong test for SSF UdpBroadcast