package rx.broadcast.integration.pp;

import org.junit.Test;
import rx.Observable;
import rx.broadcast.Broadcast;
import rx.broadcast.SingleSourceFifoOrder;
import rx.broadcast.UdpBroadcast;
import rx.observers.TestSubscriber;

import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.concurrent.TimeUnit;

public class PingPongUdpSingleSourceFifoOrder {
    private static final int MESSAGE_COUNT = 100;

    private static final long TIMEOUT = 30;

    /**
     * Receive a PING and respond with a PONG.
     * @throws SocketException if the socket could not be opened, or the socket could not bind to the given port.
     * @throws UnknownHostException if no IP address for the host machine could be found.
     */
    @Test
    public final void recv() throws SocketException, UnknownHostException {
        final int port = Integer.valueOf(System.getProperty("port"));
        final DatagramSocket socket = new DatagramSocket(port);
        final InetAddress destination = InetAddress.getByName(System.getProperty("destination"));
        final Broadcast broadcast = new UdpBroadcast<>(socket, destination, port, new SingleSourceFifoOrder<>());

        final TestSubscriber<Ping> subscriber = new TestSubscriber<>();

        broadcast.valuesOfType(Ping.class)
            .doOnNext(System.out::println)
            .concatMap(ping ->
                broadcast.send(new Pong(ping.value))
                    // Once we've sent the response, we can emit the PING value to the subscriber.
                    // The cast here is a hack to allow us to concatenate a PING onto the stream.
                    // Where this is an `Observable<Void>` we know we won't get anything that needs to be casted.
                    .cast(Ping.class)
                    .concatWith(Observable.just(ping))
                    .doOnCompleted(() -> System.out.println("Sent PONG")))
            .take(MESSAGE_COUNT)
            .subscribe(subscriber);

        subscriber.awaitTerminalEventAndUnsubscribeOnTimeout(TIMEOUT, TimeUnit.SECONDS);
        subscriber.assertNoErrors();
        subscriber.assertValueCount(MESSAGE_COUNT);
    }

    /**
     * Send a set PING messages to the receiver, expecting PONG messages in response.
     * @param args the command line arguments passed to the program
     * @throws SocketException if the socket could not be opened, or the socket could not bind to the given port.
     * @throws UnknownHostException if no IP address for the host machine could be found.
     */
    public static void main(final String[] args) throws InterruptedException, SocketException, UnknownHostException {
        final int port = Integer.valueOf(System.getProperty("port"));
        final DatagramSocket socket = new DatagramSocket(port);
        final InetAddress destination = InetAddress.getByName(System.getProperty("destination"));
        final Broadcast broadcast = new UdpBroadcast<>(socket, destination, port, new SingleSourceFifoOrder<>());

        Observable.range(1, MESSAGE_COUNT)
            .map(Ping::new)
            .doOnNext(System.out::println)
            .concatMap(value ->
                broadcast.send(value)
                    .doOnCompleted(() -> System.out.printf("Sent %s%n", value))
                    .cast(Pong.class)
                    .concatWith(broadcast.valuesOfType(Pong.class).first()))
            .timeout(TIMEOUT, TimeUnit.SECONDS)
            .toBlocking()
            .subscribe(pong ->
                System.out.printf("Received %s%n", pong));
    }
}

@@ -392,16 +392,125 @@
 eUnit;%0A%0A
+@SuppressWarnings(%7B%22checkstyle:MagicNumber%22, %22checkstyle:LineLength%22, %22checkstyle:AvoidInlineConditionals%22%7D)%0A
 public c
@@ -1027,51 +1027,292 @@
 t = 
-Integer.valueOf(System.getProperty(%22port%22))
+System.getProperty(%22port%22) != null%0A            ? Integer.valueOf(System.getProperty(%22port%22))%0A            : 54321;%0A        final int destinationPort = System.getProperty(%22destinationPort%22) != null%0A            ? Integer.valueOf(System.getProperty(%22destinationPort%22))%0A            : 12345
 ;%0A  
@@ -1404,32 +1404,88 @@
 ss destination =
+ System.getProperty(%22destination%22) != null%0A            ?
  InetAddress.get
@@ -1517,32 +1517,81 @@
 (%22destination%22))
+%0A            : InetAddress.getByName(%22localhost%22)
 ;%0A        final 
@@ -1644,33 +1644,44 @@
 t, destination, 
-p
+destinationP
 ort, new SingleS
@@ -1693,33 +1693,32 @@
 FifoOrder%3C%3E());%0A
-%0A
         final Te
@@ -3199,51 +3199,292 @@
 t = 
-Integer.valueOf(System.getProperty(%22port%22))
+System.getProperty(%22port%22) != null%0A            ? Integer.valueOf(System.getProperty(%22port%22))%0A            : 54321;%0A        final int destinationPort = System.getProperty(%22destinationPort%22) != null%0A            ? Integer.valueOf(System.getProperty(%22destinationPort%22))%0A            : 12345
 ;%0A  
@@ -3584,16 +3584,72 @@
 nation =
+ System.getProperty(%22destination%22) != null%0A            ?
  InetAdd
@@ -3697,16 +3697,65 @@
 ation%22))
+%0A            : InetAddress.getByName(%22localhost%22)
 ;%0A      
@@ -3824,17 +3824,28 @@
 nation, 
-p
+destinationP
 ort, new

Add default values to system properties read in SSF ping pong test