// Path: src/main/java/rxbroadcast/Broadcast.java
// public interface Broadcast {
//     /**
//      * Broadcasts the given value when the returned {@code Observable} is subscribed to.
//      * @param value the value to broadcast
//      * @return an Observable stream representing the message sent status
//      */
//     @Contract(pure = true)
//     Observable<@NotNull Void> send(@NotNull Object value);
// 
//     /**
//      * Returns a (hot) {@code Observable} stream of broadcasted values.
//      * @param clazz the class type to filter the broadcasted values by
//      * @param <T> the output stream type
//      * @return an Observable stream of broadcasted values
//      */
//     @Contract(pure = true)
//     <T> Observable<@NotNull T> valuesOfType(@NotNull Class<T> clazz);
// }
// 
// Path: src/main/java/rxbroadcast/NoOrder.java
// public final class NoOrder<T> implements BroadcastOrder<T, T> {
//     /**
//      * {@inheritDoc}
//      */
//     @Contract(pure = true)
//     @Override
//     public T prepare(final T value) {
//         return value;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void receive(final Sender sender, final Consumer<T> consumer, final T message) {
//         consumer.accept(message);
//     }
// }
// 
// Path: src/test/java/rxbroadcast/TestValue.java
// public final class TestValue implements Serializable {
//     private static final long serialVersionUID = 114L;
// 
//     @SuppressWarnings("WeakerAccess")
//     public int value;
// 
//     public TestValue(final int value) {
//         this.value = value;
//     }
// 
//     @Deprecated
//     @SuppressWarnings("unused")
//     public TestValue() {
// 
//     }
// 
//     @Override
//     public final boolean equals(final Object o) {
//         if (this == o) {
//             return true;
//         }
//         if (o == null || getClass() != o.getClass()) {
//             return false;
//         }
//         final TestValue testValue = (TestValue) o;
//         return value == testValue.value;
//     }
// 
//     @Override
//     public final int hashCode() {
//         return Objects.hash(value);
//     }
// 
//     @Override
//     public final String toString() {
//         return String.format("TestValue{%d}", value);
//     }
// }
// 
// Path: src/main/java/rxbroadcast/UdpBroadcast.java
// public final class UdpBroadcast<T> implements Broadcast {
//     private final DatagramSocket socket;
// 
//     private final Observable<Object> values;
// 
//     private final ConcurrentHashMap<Class<?>, Observable<?>> streams;
// 
//     private final Serializer<T> serializer;
// 
//     private final InetSocketAddress destination;
// 
//     private final Single<BroadcastOrder<T, Object>> broadcastOrder;
// 
//     @SuppressWarnings("WeakerAccess")
//     public UdpBroadcast(
//         final DatagramSocket socket,
//         final InetSocketAddress destination,
//         final Serializer<T> serializer,
//         final Function<Sender, BroadcastOrder<T, Object>> createBroadcastOrder
//     ) {
//         this.socket = socket;
//         final Scheduler singleThreadScheduler = Schedulers.from(
//             Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory()));
//         final Single<Sender> host = Single.fromCallable(new WhoAmI(
//             socket.getLocalPort(), destination.getAddress() instanceof Inet6Address));
//         this.broadcastOrder = host.subscribeOn(Schedulers.io()).map(createBroadcastOrder::apply).cache();
//         this.broadcastOrder.subscribe();
//         this.values = broadcastOrder.flatMapObservable((order1) ->
//             Observable.unsafeCreate(new UdpBroadcastOnSubscribe<>(socket, serializer, order1))
//                 .subscribeOn(singleThreadScheduler))
//             .share();
//         this.serializer = serializer;
//         this.streams = new ConcurrentHashMap<>();
//         this.destination = destination;
//     }
// 
//     public UdpBroadcast(
//         final DatagramSocket socket,
//         final InetSocketAddress destination,
//         final Serializer<T> serializer,
//         final BroadcastOrder<T, Object> order
//     ) {
//         this(socket, destination, serializer, (host) -> order);
//     }
// 
//     @SuppressWarnings("unchecked")
//     public UdpBroadcast(
//         final DatagramSocket socket,
//         final InetSocketAddress destination,
//         final BroadcastOrder<T, Object> order
//     ) {
//         this(socket, destination, new KryoSerializer<>(), (host) -> order);
//     }
// 
//     @SuppressWarnings("unchecked")
//     public UdpBroadcast(
//         final DatagramSocket socket,
//         final InetSocketAddress destination,
//         final Function<Sender, BroadcastOrder<T, Object>> createBroadcastOrder
//     ) {
//         this(socket, destination, new KryoSerializer<>(), createBroadcastOrder);
//     }
// 
//     @Override
//     public Observable<Void> send(@NotNull final Object value) {
//         return broadcastOrder.flatMapObservable((order) -> {
//             try {
//                 final byte[] data = serializer.encode(order.prepare(value));
//                 final DatagramPacket packet = new DatagramPacket(
//                     data, data.length, destination.getAddress(), destination.getPort());
//                 socket.send(packet);
//                 return null;
//             } catch (final Exception e) {
//                 return Observable.error(e);
//             }
//         });
//     }
// 
//     @Override
//     @SuppressWarnings("unchecked")
//     public <V> Observable<@NotNull V> valuesOfType(@NotNull final Class<V> clazz) {
//         return (Observable<V>) streams.computeIfAbsent(clazz, k -> values.ofType(k).share());
//     }
// }


import rxbroadcast.Broadcast;
import rxbroadcast.NoOrder;
import rxbroadcast.TestValue;
import rxbroadcast.UdpBroadcast;
import org.junit.Test;
import rx.Observable;
import rx.observers.TestSubscriber;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.concurrent.TimeUnit;
package rxbroadcast.integration;




@SuppressWarnings({"checkstyle:MagicNumber", "checkstyle:AvoidInlineConditionals"})
public final class NoOrderUdpBroadcastTest {
    private static final int MESSAGE_COUNT = 100;

// Path: src/main/java/rxbroadcast/Broadcast.java
// public interface Broadcast {
//     /**
//      * Broadcasts the given value when the returned {@code Observable} is subscribed to.
//      * @param value the value to broadcast
//      * @return an Observable stream representing the message sent status
//      */
//     @Contract(pure = true)
//     Observable<@NotNull Void> send(@NotNull Object value);
// 
//     /**
//      * Returns a (hot) {@code Observable} stream of broadcasted values.
//      * @param clazz the class type to filter the broadcasted values by
//      * @param <T> the output stream type
//      * @return an Observable stream of broadcasted values
//      */
//     @Contract(pure = true)
//     <T> Observable<@NotNull T> valuesOfType(@NotNull Class<T> clazz);
// }
// 
// Path: src/main/java/rxbroadcast/NoOrder.java
// public final class NoOrder<T> implements BroadcastOrder<T, T> {
//     /**
//      * {@inheritDoc}
//      */
//     @Contract(pure = true)
//     @Override
//     public T prepare(final T value) {
//         return value;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public void receive(final Sender sender, final Consumer<T> consumer, final T message) {
//         consumer.accept(message);
//     }
// }
// 
// Path: src/test/java/rxbroadcast/TestValue.java
// public final class TestValue implements Serializable {
//     private static final long serialVersionUID = 114L;
// 
//     @SuppressWarnings("WeakerAccess")
//     public int value;
// 
//     public TestValue(final int value) {
//         this.value = value;
//     }
// 
//     @Deprecated
//     @SuppressWarnings("unused")
//     public TestValue() {
// 
//     }
// 
//     @Override
//     public final boolean equals(final Object o) {
//         if (this == o) {
//             return true;
//         }
//         if (o == null || getClass() != o.getClass()) {
//             return false;
//         }
//         final TestValue testValue = (TestValue) o;
//         return value == testValue.value;
//     }
// 
//     @Override
//     public final int hashCode() {
//         return Objects.hash(value);
//     }
// 
//     @Override
//     public final String toString() {
//         return String.format("TestValue{%d}", value);
//     }
// }
// 
// Path: src/main/java/rxbroadcast/UdpBroadcast.java
// public final class UdpBroadcast<T> implements Broadcast {
//     private final DatagramSocket socket;
// 
//     private final Observable<Object> values;
// 
//     private final ConcurrentHashMap<Class<?>, Observable<?>> streams;
// 
//     private final Serializer<T> serializer;
// 
//     private final InetSocketAddress destination;
// 
//     private final Single<BroadcastOrder<T, Object>> broadcastOrder;
// 
//     @SuppressWarnings("WeakerAccess")
//     public UdpBroadcast(
//         final DatagramSocket socket,
//         final InetSocketAddress destination,
//         final Serializer<T> serializer,
//         final Function<Sender, BroadcastOrder<T, Object>> createBroadcastOrder
//     ) {
//         this.socket = socket;
//         final Scheduler singleThreadScheduler = Schedulers.from(
//             Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory()));
//         final Single<Sender> host = Single.fromCallable(new WhoAmI(
//             socket.getLocalPort(), destination.getAddress() instanceof Inet6Address));
//         this.broadcastOrder = host.subscribeOn(Schedulers.io()).map(createBroadcastOrder::apply).cache();
//         this.broadcastOrder.subscribe();
//         this.values = broadcastOrder.flatMapObservable((order1) ->
//             Observable.unsafeCreate(new UdpBroadcastOnSubscribe<>(socket, serializer, order1))
//                 .subscribeOn(singleThreadScheduler))
//             .share();
//         this.serializer = serializer;
//         this.streams = new ConcurrentHashMap<>();
//         this.destination = destination;
//     }
// 
//     public UdpBroadcast(
//         final DatagramSocket socket,
//         final InetSocketAddress destination,
//         final Serializer<T> serializer,
//         final BroadcastOrder<T, Object> order
//     ) {
//         this(socket, destination, serializer, (host) -> order);
//     }
// 
//     @SuppressWarnings("unchecked")
//     public UdpBroadcast(
//         final DatagramSocket socket,
//         final InetSocketAddress destination,
//         final BroadcastOrder<T, Object> order
//     ) {
//         this(socket, destination, new KryoSerializer<>(), (host) -> order);
//     }
// 
//     @SuppressWarnings("unchecked")
//     public UdpBroadcast(
//         final DatagramSocket socket,
//         final InetSocketAddress destination,
//         final Function<Sender, BroadcastOrder<T, Object>> createBroadcastOrder
//     ) {
//         this(socket, destination, new KryoSerializer<>(), createBroadcastOrder);
//     }
// 
//     @Override
//     public Observable<Void> send(@NotNull final Object value) {
//         return broadcastOrder.flatMapObservable((order) -> {
//             try {
//                 final byte[] data = serializer.encode(order.prepare(value));
//                 final DatagramPacket packet = new DatagramPacket(
//                     data, data.length, destination.getAddress(), destination.getPort());
//                 socket.send(packet);
//                 return null;
//             } catch (final Exception e) {
//                 return Observable.error(e);
//             }
//         });
//     }
// 
//     @Override
//     @SuppressWarnings("unchecked")
//     public <V> Observable<@NotNull V> valuesOfType(@NotNull final Class<V> clazz) {
//         return (Observable<V>) streams.computeIfAbsent(clazz, k -> values.ofType(k).share());
//     }
// }


// Path: src/test/java/rxbroadcast/integration/NoOrderUdpBroadcastTest.java
import rxbroadcast.Broadcast;
import rxbroadcast.NoOrder;
import rxbroadcast.TestValue;
import rxbroadcast.UdpBroadcast;
import org.junit.Test;
import rx.Observable;
import rx.observers.TestSubscriber;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.concurrent.TimeUnit;

package rxbroadcast.integration;




@SuppressWarnings({"checkstyle:MagicNumber", "checkstyle:AvoidInlineConditionals"})
public final class NoOrderUdpBroadcastTest {
    private static final int MESSAGE_COUNT = 100;

    private static final long TIMEOUT = 30;