package org.yaourtcorp.kafka.producer.example;

import java.util.Properties;

import javax.annotation.Resource;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;

public abstract class EventsProducer<T extends EventObject> {

	protected final Logger logger = LoggerFactory.getLogger(getClass());
	
	@Value("#{kafkaProperties}")
	private Properties kafkaProperties;
	
	@Value( "${kafka.topic}" )
	private String topicName;

	@Value( "${thead.sleep.in.ms}" )
	private int threadSleepInMs;
	
	
	@Resource(name="eventSerializerForKafka")
	private EventSerializerForKafka<T> eventSerializerForKafka;
	
	@Resource(name="eventsFactory")
	protected EventsFactory<T> eventsFactory;

	public void execute() throws Exception{

		ProducerConfig config = new ProducerConfig(kafkaProperties);

		Producer<String,  String> producer = null;

		try {
			producer =  new Producer<String, String>(config);
			while(true) {
				T event = eventsFactory.createNewEvent();
				KeyedMessage<String, String> data = new KeyedMessage<String, String>(topicName, eventSerializerForKafka.serialize(event));
				logger.info("Sending Message #:{}", event.getUuid());
				producer.send(data);
				Thread.sleep(threadSleepInMs);
			}
		} finally {
			if(producer != null){
				producer.close();
			}
		}
	}
}

@@ -337,17 +337,414 @@
 .Value;%0A
+/*%0A * Main class from all (aka the only one to have some interesting code, forget the others!).%3Cbr/%3E%0A * Use the kafka producer api to send %22events%22 (objects) to kafka. %3Cbr/%3E%0A * Use a %7B@code EventsFactory%7D to create new events to send.%0A * Use a %7B@code EventSerializerForKafka%7D to serialized events as String before throwing them into kafka.%0A * @see EventsFactory%0A * @see EventSerializerForKafka%0A */
 %0A
-
 public a
@@ -866,24 +866,72 @@
 Class());%0A%09%0A
+%09/* properties to connect to kafka/zookeeper */%0A
 %09@Value(%22#%7Bk
@@ -987,16 +987,47 @@
 ties;%0A%09%0A
+%09/* name of the kafka topic */%0A
 %09@Value(
@@ -1074,16 +1074,56 @@
 cName;%0A%0A
+%09/* time to wait between to messages */%0A
 %09@Value(
@@ -1407,16 +1407,48 @@
 ption%7B%0A%0A
+%09%09// configure a kafka producer%0A
 %09%09Produc
@@ -1502,17 +1502,16 @@
 rties);%0A
-%0A
 %09%09Produc

Clean comments