package rx.broadcast;

import org.jetbrains.annotations.NotNull;
import rx.Observable;
import rx.Scheduler;
import rx.Single;
import rx.schedulers.Schedulers;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.function.Function;

public final class UdpBroadcast<A> implements Broadcast {
    private final DatagramSocket socket;

    private final Observable<Object> values;

    private final ConcurrentHashMap<Class<?>, Observable<?>> streams;

    private final Serializer<A> serializer;

    private final InetAddress destinationAddress;

    private final int destinationPort;

    private final Single<BroadcastOrder<A, Object>> broadcastOrder;

    @SuppressWarnings("WeakerAccess")
    public UdpBroadcast(
        final DatagramSocket socket,
        final InetAddress destinationAddress,
        final int destinationPort,
        final Serializer<A> serializer,
        final Function<Sender, BroadcastOrder<A, Object>> createBroadcastOrder
    ) {
        this.socket = socket;
        final Scheduler singleThreadScheduler = Schedulers.from(
            Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory()));
        final Single<Sender> host = Single.fromCallable(new WhoAmI(destinationPort));
        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.destinationAddress = destinationAddress;
        this.destinationPort = destinationPort;
    }

    public UdpBroadcast(
        final DatagramSocket socket,
        final InetAddress destinationAddress,
        final int destinationPort,
        final Serializer<A> serializer,
        final BroadcastOrder<A, Object> order
    ) {
        this(socket, destinationAddress, destinationPort, serializer, (host) -> order);
    }

    @SuppressWarnings("unchecked")
    public UdpBroadcast(
        final DatagramSocket socket,
        final InetAddress destinationAddress,
        final int destinationPort,
        final BroadcastOrder<A, Object> order
    ) {
        this(socket, destinationAddress, destinationPort, new KryoSerializer<>(), (host) -> order);
    }

    @SuppressWarnings("unchecked")
    public UdpBroadcast(
        final DatagramSocket socket,
        final InetAddress destinationAddress,
        final int destinationPort,
        final Function<Sender, BroadcastOrder<A, Object>> createBroadcastOrder
    ) {
        this(socket, destinationAddress, destinationPort, 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, destinationAddress, destinationPort);
                socket.send(packet);
                return Observable.empty();
            } catch (final Throwable e) {
                return Observable.error(e);
            }
        });
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T> Observable<@NotNull T> valuesOfType(@NotNull final Class<T> clazz) {
        return (Observable<T>) streams.computeIfAbsent(clazz, k -> values.ofType(k).share());
    }
}

@@ -401,17 +401,17 @@
 oadcast%3C
-A
+T
 %3E implem
@@ -607,33 +607,33 @@
 inal Serializer%3C
-A
+T
 %3E serializer;%0A%0A 
@@ -754,33 +754,33 @@
 %3CBroadcastOrder%3C
-A
+T
 , Object%3E%3E broad
@@ -989,33 +989,33 @@
 inal Serializer%3C
-A
+T
 %3E serializer,%0A  
@@ -1050,33 +1050,33 @@
  BroadcastOrder%3C
-A
+T
 , Object%3E%3E creat
@@ -2113,17 +2113,17 @@
 ializer%3C
-A
+T
 %3E serial
@@ -2149,33 +2149,33 @@
  BroadcastOrder%3C
-A
+T
 , Object%3E order%0A
@@ -2476,33 +2476,33 @@
  BroadcastOrder%3C
-A
+T
 , Object%3E order%0A
@@ -2840,17 +2840,17 @@
 stOrder%3C
-A
+T
 , Object
@@ -3630,17 +3630,17 @@
 public %3C
-T
+V
 %3E Observ
@@ -3653,17 +3653,17 @@
 NotNull 
-T
+V
 %3E values
@@ -3690,17 +3690,17 @@
 l Class%3C
-T
+V
 %3E clazz)
@@ -3729,17 +3729,17 @@
 ervable%3C
-T
+V
 %3E) strea

Rename UdpBroadcast generic types for clarity