package com.neverwinterdp.kafka.producer;

import java.nio.charset.Charset;
import java.util.UUID;

import org.apache.kafka.clients.producer.Callback;

import com.neverwinterdp.util.JSONSerializer;

/**
 * @author Tuan Nguyen
 * @email tuan08@gmail.com
 */
abstract public class AbstractKafkaWriter implements KafkaWriter {
  final static public  Charset UTF8 = Charset.forName("UTF-8") ;
  
  private String name;

  public String getName() { return name; }

  public void setName(String name) { this.name = name; }

  public AbstractKafkaWriter(String name) {
    this.name = name;
  }

  @Override
  public void send(String topic, String data, long timeout) throws Exception {
    send(topic, nextKey(-1), data, timeout);
  }

  @Override
  public void send(String topic, String key, String data, long timeout) throws Exception {
    send(topic, -1, key, data, null, timeout);
  }

  public void send(String topic, int partition, String data, long timeout) throws Exception {
    send(topic, partition, nextKey(-1), data, null, timeout);
  }
  
  @Override
  public void send(String topic, int partition, String key, String data, long timeout) throws Exception {
    send(topic, partition, key, data, null, timeout);
  }

  @Override
  public <T> void send(String topic, T obj, long timeout) throws Exception {
    send(topic, -1, nextKey(-1), JSONSerializer.INSTANCE.toString(obj), null, timeout);
  }
  
  public void send(String topic, int partition, String key, String data, Callback callback, long timeout) throws Exception {
    byte[] keyBytes = key.getBytes(UTF8);
    byte[] messageBytes = data.getBytes(UTF8);
    send(topic, partition, keyBytes, messageBytes, callback, timeout);
  }
  
  
  private String nextKey(int partition) {
    if(partition >= 0) {
      return "p:" + partition + ":" + UUID.randomUUID().toString();
    } else {
      return UUID.randomUUID().toString();
    }
  }
  
  public void commit() throws Exception {
  }
}
@@ -91,16 +91,63 @@
 il.UUID;
+%0Aimport java.util.concurrent.atomic.AtomicLong;
 %0A%0Aimport
@@ -428,24 +428,143 @@
 TF-8%22) ;%0A  %0A
+  private String writerUUID     = UUID.randomUUID().toString();%0A  private AtomicLong  idTracker = new AtomicLong();%0A  %0A
   private St
@@ -574,16 +574,21 @@
 g name;%0A
+  %0A  
 %0A  publi
@@ -1977,90 +1977,126 @@
 %22 + 
-UUID.randomUUID().toString();%0A    %7D else %7B%0A      return UUID.randomUUID().toString
+writerUUID + %22:%22 + idTracker.incrementAndGet();%0A    %7D else %7B%0A      return writerUUID + %22:%22 + idTracker.incrementAndGet
 ();%0A

fix message key generation