package org.ctrip.ops.sysdev.inputs;

import java.lang.reflect.Constructor;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;
import org.ctrip.ops.sysdev.decoder.IDecode;
import org.ctrip.ops.sysdev.decoder.JsonDecoder;
import org.ctrip.ops.sysdev.decoder.PlainDecoder;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class Kafka extends BaseInput {
	private static final Logger logger = Logger
			.getLogger(Kafka.class.getName());

	private class Consumer implements Runnable {
		private KafkaStream<byte[], byte[]> m_stream;
		private ArrayBlockingQueue messageQueue;
		private IDecode decoder;

		public Consumer(KafkaStream<byte[], byte[]> a_stream,
				ArrayBlockingQueue fairQueue, IDecode decoder) {
			m_stream = a_stream;
			this.messageQueue = fairQueue;
			this.decoder = decoder;
		}

		public void run() {
			ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
			while (it.hasNext()) {
				String m = new String(it.next().message());
				try {
					Map<String, Object> event = decoder.decode(m);
					this.messageQueue.put(event);
				} catch (InterruptedException e) {
					logger.warn("put message to queue failed");
					logger.trace(e.getMessage());
				} catch (Exception e) {
					logger.error("json decode failed:" + m);
					logger.trace(e.getMessage());
				}
			}
		}
	}

	private ConsumerConnector consumer;
	private ExecutorService executor;
	private IDecode decoder;
	private Map<String, Integer> topic;
	private int threads;

	public Kafka(Map<String, Object> config, ArrayBlockingQueue messageQueue) {
		super(config, messageQueue);
	}

	@SuppressWarnings("unchecked")
	@Override
	protected void prepare() {
		if (this.config.containsKey("threads")) {
			this.threads = (int) this.config.get("threads");
		} else {
			this.threads = 1;
		}

		this.topic = (Map<String, Integer>) this.config.get("topic");

		Properties props = new Properties();

		HashMap<String, String> consumerSettings = (HashMap<String, String>) this.config
				.get("consumer_settings");
		Iterator<Entry<String, String>> consumerSetting = consumerSettings
				.entrySet().iterator();

		while (consumerSetting.hasNext()) {
			Map.Entry<String, String> entry = consumerSetting.next();
			String k = entry.getKey();
			String v = entry.getValue();
			props.put(k, v);
		}
		consumer = kafka.consumer.Consumer
				.createJavaConsumerConnector(new ConsumerConfig(props));

		String codec = (String) this.config.get("codec");
		if (codec != null && codec.equalsIgnoreCase("plain")) {
			this.decoder = new PlainDecoder();
		} else {
			this.decoder = new JsonDecoder();
		}

	}

	public void emit() {
		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = null;

		consumerMap = consumer.createMessageStreams(this.topic);

		Iterator<Entry<String, Integer>> topicIT = this.topic.entrySet()
				.iterator();

		while (topicIT.hasNext()) {
			Map.Entry<String, Integer> entry = topicIT.next();
			String topic = entry.getKey();
			List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);

			executor = Executors.newFixedThreadPool(this.threads);

			for (final KafkaStream<byte[], byte[]> stream : streams) {
				executor.submit(new Consumer(stream, messageQueue, this.decoder));
			}
		}

	}
}

@@ -35,46 +35,8 @@
 s;%0A%0A
-import java.lang.reflect.Constructor;%0A
 impo
@@ -1790,30 +1790,8 @@
 pic;
-%0A%09private int threads;
 %0A%0A%09p
@@ -1975,141 +1975,8 @@
 ) %7B%0A
-%09%09if (this.config.containsKey(%22threads%22)) %7B%0A%09%09%09this.threads = (int) this.config.get(%22threads%22);%0A%09%09%7D else %7B%0A%09%09%09this.threads = 1;%0A%09%09%7D%0A%0A
 %09%09th
@@ -2772,17 +2772,16 @@
 ();%0A%09%09%7D%0A
-%0A
 %09%7D%0A%0A%09pub
@@ -3131,16 +3131,55 @@
 tKey();%0A
+%09%09%09Integer threads = entry.getValue();%0A
 %09%09%09List%3C
@@ -3242,16 +3242,16 @@
 opic);%0A%0A
+
 %09%09%09execu
@@ -3285,21 +3285,16 @@
 eadPool(
-this.
 threads)

remove threads settings