package reactor.kafka;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

public class KafkaPublisher<T> implements Publisher<T> {

  @Override
  public void subscribe(Subscriber<? super T> subscriber) {

  }
}

@@ -28,76 +28,617 @@
 ort 
-org.reactivestreams.Publisher;%0Aimport org.reactivestreams.Subscriber
+kafka.consumer.Consumer;%0Aimport kafka.consumer.ConsumerConfig;%0Aimport kafka.consumer.ConsumerIterator;%0Aimport kafka.consumer.KafkaStream;%0Aimport kafka.javaapi.consumer.ConsumerConnector;%0Aimport kafka.message.MessageAndMetadata;%0Aimport kafka.serializer.Decoder;%0Aimport org.reactivestreams.Publisher;%0Aimport org.reactivestreams.Subscriber;%0Aimport org.reactivestreams.Subscription;%0A%0Aimport java.util.HashMap;%0Aimport java.util.List;%0Aimport java.util.Map;%0Aimport java.util.Properties;%0Aimport java.util.concurrent.ConcurrentHashMap;%0Aimport java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Executors
 ;%0A%0Ap
@@ -664,17 +664,20 @@
 blisher%3C
-T
+K, V
 %3E implem
@@ -695,89 +695,1878 @@
 her%3C
-T%3E %7B%0A%0A  @Override%0A  public void subscribe(Subscriber%3C? super T%3E subscriber) %7B
+KeyValuePair%3CK, V%3E%3E %7B%0A%0A  private final static int concurrencyLevel = 1;%0A%0A  private final ExecutorService                                           executor;%0A  private final Map%3CSubscription, Subscriber%3C? super KeyValuePair%3CK, V%3E%3E%3E subscribers;%0A%0A  public KafkaPublisher(Properties consumerProperties,%0A                        String topic,%0A                        Decoder%3CK%3E keyDecoder,%0A                        Decoder%3CV%3E valueDecoder) %7B%0A    this.subscribers = new ConcurrentHashMap%3C%3E();%0A    this.executor = Executors.newFixedThreadPool(concurrencyLevel);%0A    ConsumerConnector consumer = Consumer.createJavaConsumerConnector(%0A      new ConsumerConfig(consumerProperties));%0A%0A    Map%3CString, Integer%3E topicCountMap = new HashMap%3CString, Integer%3E();%0A    topicCountMap.put(topic, concurrencyLevel);%0A    Map%3CString, List%3CKafkaStream%3CK, V%3E%3E%3E consumerMap =%0A      consumer.createMessageStreams(topicCountMap,%0A                                    keyDecoder,%0A                                    valueDecoder);%0A    List%3CKafkaStream%3CK, V%3E%3E streams = consumerMap.get(topic);%0A%0A%0A    for (final KafkaStream stream : streams) %7B%0A      executor.execute(() -%3E %7B%0A        ConsumerIterator%3CK, V%3E it = stream.iterator();%0A        while (it.hasNext() && !Thread.currentThread().isInterrupted()) %7B%0A          MessageAndMetadata%3CK, V%3E msg = it.next();%0A          KeyValuePair%3CK, V%3E kvp = new KeyValuePair%3CK, V%3E(msg.key(), msg.message());%0A%0A          for(Map.Entry%3CSubscription, Subscriber%3C? super KeyValuePair%3CK, V%3E%3E%3E entry: subscribers.entrySet()) %7B%0A            entry.getValue().onNext(kvp);%0A          %7D%0A%0A        %7D%0A      %7D);%0A    %7D%0A%0A  %7D%0A%0A  @Override%0A  public void subscribe(Subscriber%3C? super KeyValuePair%3CK, V%3E%3E subscriber) %7B%0A    this.subscribers.put(new KafkaSubscription%3C%3E(subscriber,%0A                                                 subscribers::remove),%0A                         subscriber);
 %0A%0A  %7D%0A
+%0A%0A
 %7D%0A

Implement kafka publisher on top of reactive streams