package com.altamiracorp.lumify.model;

import com.altamiracorp.lumify.core.config.Configuration;
import com.altamiracorp.lumify.core.model.workQueue.WorkQueueRepository;
import kafka.javaapi.producer.Producer;
import kafka.javaapi.producer.ProducerData;
import kafka.producer.ProducerConfig;
import org.json.JSONArray;
import org.json.JSONObject;

import java.util.Map;
import java.util.Properties;

public class KafkaWorkQueueRepository extends WorkQueueRepository {
    private Producer<String, JSONObject> kafkaProducer;

    @Override
    public void init(Map config) {
        super.init(config);

        String zookeeperServerNames = (String) config.get(Configuration.ZK_SERVERS);
        Properties props = new Properties();
        props.put("zk.connect", zookeeperServerNames + "/kafka"); // TODO what happens if zookeeperServerNames has multiple names
        props.put("serializer.class", KafkaJsonEncoder.class.getName());
        ProducerConfig producerConfig = new ProducerConfig(props);
        kafkaProducer = new Producer<String, JSONObject>(producerConfig);
    }

    @Override
    protected void pushOnQueue(String queueName, JSONObject json, String... extra) {
        if (extra != null && extra.length > 0) {
            JSONArray extraArray = new JSONArray();
            for (String e : extra) {
                extraArray.put(e);
            }
            json.put(KafkaJsonEncoder.EXTRA, extraArray);
        }

        json.put("sourceBolt", getClass().getName());

        ProducerData<String, JSONObject> data = new ProducerData<String, JSONObject>(queueName, json);
        kafkaProducer.send(data);
    }
}

@@ -342,16 +342,45 @@
 bject;%0A%0A
+import java.util.Collection;%0A
 import j
@@ -642,24 +642,17 @@
 String z
-ookeeper
+k
 ServerNa
@@ -661,17 +661,27 @@
 s = 
-(
+getZkConnect
 String
-) 
+(
 conf
@@ -712,16 +712,17 @@
 SERVERS)
+)
 ;%0A      
@@ -793,24 +793,17 @@
 nect%22, z
-ookeeper
+k
 ServerNa
@@ -809,85 +809,10 @@
 ames
- + %22/kafka%22); // TODO what happens if zookeeperServerNames has multiple names
+);
 %0A   
@@ -1025,24 +1025,717 @@
 ig);%0A    %7D%0A%0A
+    private String getZkConnectString(Object zkServers) %7B%0A        String zkServerNames;%0A        if (zkServers instanceof Collection) %7B%0A            Collection c = (Collection) zkServers;%0A            StringBuilder temp = new StringBuilder();%0A            boolean first = true;%0A            for (Object zkServer : c) %7B%0A                if (!first) %7B%0A                    temp.append(%22,%22);%0A                %7D%0A                temp.append(zkServer.toString());%0A                temp.append(%22/kafka%22);%0A                first = false;%0A            %7D%0A            zkServerNames = temp.toString();%0A        %7D else %7B%0A            zkServerNames = zkServers + %22/kafka%22;%0A        %7D%0A        return zkServerNames;%0A    %7D%0A%0A
     @Overrid

fix KafkaWorkQueueRepository to work with multiple zk servers