package rx.broadcast;

import org.junit.Test;
import rx.observers.TestSubscriber;

import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.util.Collections;

public class UdpBroadcastTest {
    @SuppressWarnings({"checkstyle:magicnumber"})
    @Test
    public final void valuesOfTypeDoesReceiveBroadcastValue() throws SocketException {
        final TestSubscriber<Object> subscriber = new TestSubscriber<>();
        final DatagramSocket sa = new DatagramSocket();
        final DatagramSocket sb = new DatagramSocket();
        final Broadcast broadcast1 = new UdpBroadcast(sa, InetAddress.getLoopbackAddress(), sb.getLocalPort());
        final Broadcast broadcast2 = new UdpBroadcast(sb, InetAddress.getLoopbackAddress(), sa.getLocalPort());

        broadcast2.valuesOfType(TestValue.class).first().subscribe(subscriber);
        broadcast1.send(new TestValue(42)).toBlocking().subscribe();

        subscriber.awaitTerminalEvent();

        subscriber.assertNoErrors();
        subscriber.assertValueCount(1);
        subscriber.assertReceivedOnNext(Collections.singletonList(new TestValue(42)));

        sa.close();
        sb.close();
    }
}

@@ -39,16 +39,38 @@
 t.Test;%0A
+import rx.Observable;%0A
 import r
@@ -97,16 +97,16 @@
 criber;%0A
-
 %0Aimport 
@@ -192,16 +192,41 @@
 eption;%0A
+import java.util.Arrays;%0A
 import j
@@ -1107,16 +1107,16 @@
 unt(1);%0A
-
         
@@ -1191,24 +1191,1230 @@
 lue(42)));%0A%0A
+        sa.close();%0A        sb.close();%0A    %7D%0A%0A    @SuppressWarnings(%7B%22checkstyle:magicnumber%22%7D)%0A    @Test%0A    public final void eachSubscriptionDoesSendTheBroadcast() throws SocketException %7B%0A        final TestSubscriber%3CObject%3E subscriber = new TestSubscriber%3C%3E();%0A        final DatagramSocket sa = new DatagramSocket();%0A        final DatagramSocket sb = new DatagramSocket();%0A        final Broadcast broadcast1 = new UdpBroadcast(sa, InetAddress.getLoopbackAddress(), sb.getLocalPort());%0A        final Broadcast broadcast2 = new UdpBroadcast(sb, InetAddress.getLoopbackAddress(), sa.getLocalPort());%0A%0A        broadcast2.valuesOfType(TestValue.class).take(4).subscribe(subscriber);%0A%0A        final Observable%3CVoid%3E testValue = broadcast1.send(new TestValue(42));%0A        testValue.toBlocking().subscribe();%0A        testValue.toBlocking().subscribe();%0A        testValue.toBlocking().subscribe();%0A        testValue.toBlocking().subscribe();%0A%0A        subscriber.awaitTerminalEvent();%0A%0A        subscriber.assertNoErrors();%0A        subscriber.assertValueCount(4);%0A        subscriber.assertReceivedOnNext(Arrays.asList(%0A            new TestValue(42), new TestValue(42), new TestValue(42), new TestValue(42)));%0A%0A
         sa.c

Add test for multiple subscriptions