package de.zalando.aruha.nakadi.repository.kafka;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import javax.annotation.PostConstruct;

import org.apache.curator.framework.CuratorFramework;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;

import de.zalando.aruha.nakadi.repository.zookeeper.ZooKeeperHolder;

public class KafkaLocationManager {

    private static final Logger LOG = LoggerFactory.getLogger(KafkaLocationManager.class);

    private final String _BROKERS_IDS_PATH = "/brokers/ids";

    @Autowired
    private ZooKeeperHolder zkFactory;

    private List<Broker> kafkaBrokerList;

    private Properties kafkaProperties;

    class Broker {
        final String host;
        final Integer port;

        public Broker(final String host, final Integer port) {
            this.host = host;
            this.port = port;
        }

        public String toString() {
            return this.host + ":" + this.port;
        }
    }

    private Broker getBroker(final byte[] data) throws JSONException, UnsupportedEncodingException {
        final JSONObject json = new JSONObject(new String(data, "UTF-8"));
        final String host = json.getString("host");
        final Integer port = json.getInt("port");
        return new Broker(host, port);
    }

    private List<Broker> fetchBrokers() {
        final List<Broker> brokers = new ArrayList<Broker>();
        try {
            final CuratorFramework curator = zkFactory.get();
            for (final String brokerId : curator.getChildren().forPath(_BROKERS_IDS_PATH)) {
                try {
                    final byte[] brokerData = curator.getData().forPath(_BROKERS_IDS_PATH + "/" + brokerId);
                    brokers.add(getBroker(brokerData));
                } catch (Exception e) {
                    LOG.info(String.format("Failed to fetch connection string for broker %s", brokerId), e);
                }
            }
        } catch (Exception e) {
            LOG.error("Failed to fetch list of brokers from ZooKeeper", e);
        }

        return brokers;
    }

    private static String buildBootstrapServers(List<Broker> brokers) {
        final StringBuilder builder = new StringBuilder();
        brokers.stream().forEach(broker -> builder.append(broker).append(","));
        return builder.deleteCharAt(builder.length() - 1).toString();
    }

    private Properties buildKafkaProperties() {
        final Properties props = new Properties();
        props.put("bootstrap.servers", buildBootstrapServers(kafkaBrokerList));
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return props;
    }

    @PostConstruct
    private void init() {
        kafkaBrokerList = fetchBrokers();
        kafkaProperties = buildKafkaProperties();
    }

    @Scheduled(fixedDelay = 30000)
    private void updateBrokers() {
        if (kafkaProperties != null) {
            List<Broker> brokers = fetchBrokers();
            if (!brokers.isEmpty()) {
                kafkaBrokerList = brokers;
                kafkaProperties.setProperty("bootstrap.servers", buildBootstrapServers(brokers));
            }
        }
    }

    public List<Broker> getKafkaBrokers() {
        return kafkaBrokerList;
    }

    public Properties getKafkaProperties() {
        return kafkaProperties;
    }
}

@@ -908,16 +908,23 @@
 s;%0A%0A    
+static 
 class Br
@@ -995,21 +995,22 @@
        p
-ublic
+rivate
  Broker(
@@ -1133,128 +1133,34 @@
     
-public String toString() %7B%0A            return this.host + %22:%22 + this.port;%0A        %7D%0A    %7D%0A%0A    private Broker getBroker
+static Broker fromByteJson
 (fin
@@ -1232,24 +1232,28 @@
 n %7B%0A        
+  
+  
 final JSONOb
@@ -1307,32 +1307,36 @@
 F-8%22));%0A        
+    
 final String hos
@@ -1363,32 +1363,36 @@
 host%22);%0A        
+  
+  
 final Integer po
@@ -1421,24 +1421,28 @@
 %22);%0A        
+    
 return new B
@@ -1460,16 +1460,120 @@
  port);%0A
+        %7D%0A%0A        public String toString() %7B%0A            return this.host + %22:%22 + this.port;%0A        %7D%0A
     %7D%0A%0A 
@@ -2011,17 +2011,27 @@
 add(
-get
 Broker
+.fromByteJson
 (bro

Move getBroker into Broker class and rename it into fromByteJson