package com.viadeo.axonframework.eventhandling.terminal.kafka;

import com.google.common.base.Optional;
import kafka.javaapi.producer.Producer;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Properties;
import java.util.UUID;

import static com.google.common.base.Preconditions.checkNotNull;

public class ProducerFactory {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProducerFactory.class);
    private static Optional<String> brokerId = Optional.absent();

    private final Properties baseProperties;

    public ProducerFactory(final ProducerConfig producerConfig) {
        this.baseProperties = checkNotNull(producerConfig).props().props();
    }

    public <K, V> Producer<K, V> create() {
        final Properties properties = getPropertiesFor(getBrokerId());
        return new Producer<>(new ProducerConfig(properties));
    }

    protected Properties getPropertiesFor(final String brokerId) {
        final Properties properties = new Properties();

        properties.put("serializer.class", EventMessageSerializer.class.getName());
        properties.put("key.serializer.class", StringEncoder.class.getName());

        for (final Object key : baseProperties.keySet()) {
            properties.put(key, baseProperties.get(key));
        }

        properties.put("broker.id", brokerId);

        LOGGER.debug("Get properties for '{}' : {}", brokerId, properties);

        return properties;
    }

    protected String getBrokerId() {
        if ( ! brokerId.isPresent()) {
            try {
                brokerId = Optional.fromNullable(InetAddress.getLocalHost().getCanonicalHostName());
            } catch (UnknownHostException e) {
                brokerId = Optional.of(UUID.randomUUID().toString());
                LOGGER.warn("Unable to define 'broker.id' property. Generating default : {}", brokerId, e);
            }
        }
        return brokerId.get();
    }

}

@@ -615,22 +615,22 @@
 String%3E 
-broker
+client
 Id = Opt
@@ -943,22 +943,22 @@
 sFor(get
-Broker
+Client
 Id());%0A 
@@ -1081,22 +1081,22 @@
  String 
-broker
+client
 Id) %7B%0A  
@@ -1470,26 +1470,26 @@
 ut(%22
-broker.id%22, broker
+client.id%22, client
 Id);
@@ -1539,30 +1539,30 @@
 '%7B%7D' : %7B%7D%22, 
-broker
+client
 Id, properti
@@ -1629,14 +1629,14 @@
  get
-Broker
+Client
 Id()
@@ -1653,22 +1653,22 @@
  if ( ! 
-broker
+client
 Id.isPre
@@ -1703,38 +1703,38 @@
                 
-broker
+client
 Id = Optional.fr
@@ -1859,22 +1859,22 @@
         
-broker
+client
 Id = Opt
@@ -1960,22 +1960,22 @@
 define '
-broker
+client
 .id' pro
@@ -2007,22 +2007,22 @@
  : %7B%7D%22, 
-broker
+client
 Id, e);%0A
@@ -2064,14 +2064,14 @@
 urn 
-broker
+client
 Id.g

rename property to 'client.id' instead of 'broker.id'