package com.gregory.testing.communication;

import com.gregory.testing.message.Message;
import com.gregory.testing.message.TimestampedMessage;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import static kafka.consumer.Consumer.createJavaConsumerConnector;

public final class KafkaConsumer implements OutputChannel {

    private final String zookeeper;
    private final String topic;
    private final KafkaStream<byte[], byte[]> stream;
    private final BlockingQueue<TimestampedMessage> queue = new LinkedBlockingQueue<>();

    public KafkaConsumer(String zookeeper, String topic) throws InterruptedException {
        this.zookeeper = zookeeper;
        this.topic = topic;

        Properties properties = new Properties();
        properties.setProperty("group.id", "kafka-consumer");
        properties.setProperty("zookeeper.connect", zookeeper);

        ConsumerConnector connector = createJavaConsumerConnector(new ConsumerConfig(properties));
        Map<String, Integer> topicCountMap = new HashMap<>();
        topicCountMap.put(topic, 1);
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = connector.createMessageStreams(topicCountMap);
        stream = consumerMap.get(topic).get(0);
        new Thread(new Task()).start();
        Thread.sleep(2500);
        this.queue.clear();
    }

    @Override
    public TimestampedMessage getMessage() {
        try {
            return queue.poll(1, TimeUnit.HOURS);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    private class Task implements Runnable {
        @Override
        public void run() {
            ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
            while (iterator.hasNext()) {
                byte[] data = iterator.next().message();
                long timestamp = System.currentTimeMillis();
                queue.add(new TimestampedMessage(timestamp, new Message(data)));
            }
        }
    }

}

@@ -731,62 +731,8 @@
 ic;%0A
-    private final KafkaStream%3Cbyte%5B%5D, byte%5B%5D%3E stream;%0A
     
@@ -969,16 +969,470 @@
 topic;%0A%0A
+        new Thread(new Task()).start();%0A        Thread.sleep(2500);%0A        this.queue.clear();%0A    %7D%0A%0A    @Override%0A    public TimestampedMessage getMessage() %7B%0A        try %7B%0A            return queue.poll(1, TimeUnit.HOURS);%0A        %7D catch (InterruptedException e) %7B%0A            throw new RuntimeException(e);%0A        %7D%0A    %7D%0A%0A    private class Task implements Runnable %7B%0A        @Override%0A        public void run() %7B%0A            while(true) %7B%0A        
         
@@ -1473,32 +1473,40 @@
 ties();%0A        
+    
+    
 properties.setPr
@@ -1543,32 +1543,40 @@
 umer%22);%0A        
+    
+    
 properties.setPr
@@ -1616,32 +1616,40 @@
 eper);%0A%0A        
+   
+     
 ConsumerConnecto
@@ -1719,24 +1719,32 @@
 operties));%0A
+        
         Map%3C
@@ -1801,16 +1801,24 @@
         
+  
+      
 topicCou
@@ -1838,16 +1838,24 @@
 ic, 1);%0A
+        
         
@@ -1974,468 +1974,89 @@
     
-stream = consumerMap.get(topic).get(0);%0A        new Thread(new Task()).start();%0A        Thread.sleep(2500);%0A        this.queue.clear();%0A    %7D%0A%0A    @Override%0A    public TimestampedMessage getMessage() %7B%0A        try %7B%0A            return queue.poll(1, TimeUnit.HOURS);%0A        %7D catch (InterruptedException e) %7B%0A            throw new RuntimeException(e);%0A        %7D%0A    %7D%0A%0A    private class Task implements Runnable %7B%0A        @Override%0A        public void run() %7B%0A
+        KafkaStream%3Cbyte%5B%5D, byte%5B%5D%3E stream = consumerMap.get(topic).get(0);%0A%0A    
     
@@ -2122,24 +2122,28 @@
 iterator();%0A
+    
             
@@ -2187,16 +2187,20 @@
         
+    
 byte%5B%5D d
@@ -2228,24 +2228,28 @@
 .message();%0A
+    
             
@@ -2313,16 +2313,20 @@
         
+    
 queue.ad
@@ -2378,24 +2378,42 @@
 ge(data)));%0A
+                %7D%0A
             

Fix kafka message consumer