package reactor.kafka;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public class KafkaSubscriber<T> implements Subscriber<T> {

  @Override
  public void onSubscribe(Subscription subscription) {

  }

  @Override
  public void onNext(T t) {

  }

  @Override
  public void onError(Throwable throwable) {

  }

  @Override
  public void onComplete() {

  }
}

@@ -32,75 +32,306 @@
 org.
-reactivestreams.Subscriber;%0Aimport org.reactivestreams.Subscription
+apache.kafka.clients.producer.KafkaProducer;%0Aimport org.apache.kafka.clients.producer.ProducerRecord;%0Aimport org.reactivestreams.Subscriber;%0Aimport org.reactivestreams.Subscription;%0Aimport reactor.fn.Consumer;%0Aimport reactor.fn.Function;%0Aimport reactor.fn.tuple.Tuple2;%0A%0Aimport java.util.Properties
 ;%0A%0Ap
@@ -358,17 +358,20 @@
 scriber%3C
-T
+K, V
 %3E implem
@@ -391,126 +391,1220 @@
 er%3CT
-%3E %7B%0A%0A  @Override%0A  public void onSubscribe(Subscription subscription) %7B%0A%0A  %7D%0A%0A  @Override%0A  public void onNext(T t) %7B%0A
+uple2%3CK, V%3E%3E %7B%0A%0A  private final    KafkaProducer%3CK, V%3E  kafkaProducer;%0A  private final    String               topic;%0A  private final    Function%3CK, Integer%3E partitioner;%0A  private final    Consumer%3CThrowable%3E  errorConsumer;%0A  private volatile Subscription         subscription;%0A%0A  public KafkaSubscriber(Properties producerProperties,%0A                         String topic,%0A                         Function%3CK, Integer%3E partitioner,%0A                         Consumer%3CThrowable%3E errorConsumer) %7B%0A    this.kafkaProducer = new KafkaProducer%3C%3E(producerProperties);%0A    this.topic = topic;%0A    this.partitioner = partitioner;%0A    this.errorConsumer = errorConsumer;%0A  %7D%0A%0A  @Override%0A  public void onSubscribe(Subscription subscription) %7B%0A    this.subscription.request(Long.MAX_VALUE);%0A  %7D%0A%0A  @Override%0A  public void onNext(Tuple2%3CK, V%3E tuple) %7B%0A    ProducerRecord%3CK, V%3E record = new ProducerRecord%3C%3E(topic,%0A                                                       partitioner.apply(tuple.getT1()),%0A                                                       tuple.getT1(),%0A                                                       tuple.getT2());%0A    kafkaProducer.send(record);%0A    this.subscription.request(Long.MAX_VALUE);
 %0A  %7D
@@ -1658,24 +1658,60 @@
 hrowable) %7B%0A
+    errorConsumer.accept(throwable);
 %0A  %7D%0A%0A  @Ove
@@ -1745,15 +1745,46 @@
 ete() %7B%0A
+    this.kafkaProducer.close();
 %0A  %7D%0A%7D%0A

Implement kafka subscriber