package reactor.kafka;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.Decoder;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaPublisher<K, V> implements Publisher<KeyValuePair<K, V>> {

  private final static int concurrencyLevel = 1;

  private final ExecutorService                                           executor;
  private final Map<Subscription, Subscriber<? super KeyValuePair<K, V>>> subscribers;

  public KafkaPublisher(Properties consumerProperties,
                        String topic,
                        Decoder<K> keyDecoder,
                        Decoder<V> valueDecoder) {
    this.subscribers = new ConcurrentHashMap<>();
    this.executor = Executors.newFixedThreadPool(concurrencyLevel);
    ConsumerConnector consumer = Consumer.createJavaConsumerConnector(
      new ConsumerConfig(consumerProperties));

    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
    topicCountMap.put(topic, concurrencyLevel);
    Map<String, List<KafkaStream<K, V>>> consumerMap =
      consumer.createMessageStreams(topicCountMap,
                                    keyDecoder,
                                    valueDecoder);
    List<KafkaStream<K, V>> streams = consumerMap.get(topic);


    for (final KafkaStream stream : streams) {
      executor.execute(() -> {
        ConsumerIterator<K, V> it = stream.iterator();
        while (it.hasNext() && !Thread.currentThread().isInterrupted()) {
          MessageAndMetadata<K, V> msg = it.next();
          KeyValuePair<K, V> kvp = new KeyValuePair<K, V>(msg.key(), msg.message());

          for(Map.Entry<Subscription, Subscriber<? super KeyValuePair<K, V>>> entry: subscribers.entrySet()) {
            entry.getValue().onNext(kvp);
          }

        }
      });
    }

  }

  @Override
  public void subscribe(Subscriber<? super KeyValuePair<K, V>> subscriber) {
    this.subscribers.put(new KafkaSubscription<>(subscriber,
                                                 subscribers::remove),
                         subscriber);

  }


}

@@ -402,16 +402,79 @@
 ription;
+%0Aimport reactor.fn.tuple.Tuple;%0Aimport reactor.fn.tuple.Tuple2;
 %0A%0Aimport
@@ -754,28 +754,22 @@
 blisher%3C
-KeyValuePair
+Tuple2
 %3CK, V%3E%3E 
@@ -889,22 +889,16 @@
         
-      
 executor
@@ -948,36 +948,30 @@
 ber%3C? super 
-KeyValuePair
+Tuple2
 %3CK, V%3E%3E%3E sub
@@ -2072,55 +2072,37 @@
     
-KeyValuePair%3CK, V%3E kvp = new KeyValuePair%3CK, V%3E
+Tuple2%3CK, V%3E tuple = Tuple.of
 (msg
@@ -2121,25 +2121,24 @@
 message());%0A
-%0A
           fo
@@ -2138,16 +2138,17 @@
      for
+ 
 (Map.Ent
@@ -2179,36 +2179,30 @@
 ber%3C? super 
-KeyValuePair
+Tuple2
 %3CK, V%3E%3E%3E ent
@@ -2203,16 +2203,17 @@
 %3E%3E entry
+ 
 : subscr
@@ -2272,11 +2272,13 @@
 ext(
-kvp
+tuple
 );%0A 
@@ -2380,20 +2380,14 @@
 per 
-KeyValuePair
+Tuple2
 %3CK, 
@@ -2578,13 +2578,11 @@
 r);%0A
-%0A
   %7D%0A%0A
-%0A
 %7D%0A

Migrate publisher to Tuple2