package stroom.kafka;

import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import stroom.util.spring.StroomScope;

import java.util.Properties;
import java.util.function.Consumer;

/**
 * A singleton responsible for sending records to Kafka.
 * <p>
 * Requires configuration:
 * <<code>
 * kafka.bootstrap.servers=<host>:<port>
 * </code>
 */
@Component
@Scope(StroomScope.SINGLETON)
public class StroomKafkaProducer {
    private static final Logger LOGGER = LoggerFactory.getLogger(StroomKafkaProducer.class);

    private final String bootstrapServers;
    private volatile Producer<String, String> producer = null;

    public StroomKafkaProducer(
            @Value("#{propertyConfigurer.getProperty('kafka.bootstrap.servers')}") final String bootstrapServers) {

        this.bootstrapServers = bootstrapServers;

        try {
            intiProducer();
        } catch (Exception e) {
            //Log and swallow the exception to prevent creation of the bean
            LOGGER.error("Error initialising kafka producer to " + bootstrapServers, e);
        }
    }

    public void send(final ProducerRecord<String, String> record, final Consumer<Exception> exceptionConsumer) {
        //kafka may not have been up on startup so ensure we have a producer instance now
        try {
            intiProducer();

            Preconditions.checkNotNull(producer).send(record, (recordMetadata, exception) -> {

                if (exception != null) {
                    exceptionConsumer.accept(exception);
                }
                LOGGER.trace("Record sent to Kafka");
            });

            producer.flush();
        } catch (Exception e) {
            LOGGER.error("Error initialising kafka producer to " + bootstrapServers, e);
            exceptionConsumer.accept(e);
        }
    }

    private void intiProducer() {
        if (producer == null) {
            synchronized (this) {
                if (producer == null) {
                    if (Strings.isNullOrEmpty(bootstrapServers)) {
                        LOGGER.error("Kafka is not properly configured: 'kafka.bootstrap.servers' is required.");
                    } else {
                        Properties props = new Properties();
                        props.put("bootstrap.servers", bootstrapServers);
                        props.put("acks", "all");
                        props.put("retries", 0);
                        props.put("batch.size", 16384);
                        props.put("linger.ms", 1);
                        props.put("buffer.memory", 33554432);
                        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
                        // TODO We will probably want to send Avro'd bytes (or something similar) at some point, so the serializer will need to change.
                        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

                        producer = new KafkaProducer<>(props);
                        LOGGER.info("Ready to send records to Kafka.");
                    }
                }
            }
        }
    }

    public void close() {
        producer.close();
    }
}

@@ -520,16 +520,58 @@
 omScope;
+%0Aimport stroom.util.spring.StroomShutdown;
 %0A%0Aimport
@@ -1004,16 +1004,87 @@
 Servers;
+%0A%0A    //instance of a kafka producer that will be shared by all threads
 %0A    pri
@@ -1342,258 +1342,8 @@
 ers;
-%0A%0A        try %7B%0A            intiProducer();%0A        %7D catch (Exception e) %7B%0A            //Log and swallow the exception to prevent creation of the bean%0A            LOGGER.error(%22Error initialising kafka producer to %22 + bootstrapServers, e);%0A        %7D
 %0A   
@@ -2080,24 +2080,89 @@
    %7D%0A    %7D%0A%0A
+    /**%0A     * Lazy initialisation of the kafka producer%0A     */%0A
     private 
@@ -2499,16 +2499,109 @@
  else %7B%0A
+                        LOGGER.info(%22Initialising kafka producer for %7B%7D%22, bootstrapServers);%0A
         
@@ -3368,24 +3368,58 @@
 ializer%22);%0A%0A
+                        try %7B%0A    
             
@@ -3465,24 +3465,28 @@
 r%3C%3E(props);%0A
+    
             
@@ -3569,58 +3569,296 @@
     
-%7D%0A                %7D%0A            %7D%0A        %7D%0A    %7D%0A
+    %7D catch (Exception e) %7B%0A                            LOGGER.error(%22Error initialising kafka producer for %7B%7D%22, bootstrapServers, e);%0A                            throw e;%0A                        %7D%0A                    %7D%0A                %7D%0A            %7D%0A        %7D%0A    %7D%0A%0A    @StroomShutdown
 %0A   
@@ -3870,21 +3870,24 @@
 ic void 
-close
+shutdown
 () %7B%0A   
@@ -3895,25 +3895,208 @@
     
-producer.close();
+if (producer != null) %7B%0A            try %7B%0A                producer.close();%0A            %7D catch (Exception e) %7B%0A                LOGGER.error(%22Error closing kafka producer%22, e);%0A            %7D%0A        %7D
 %0A   

Make KafkaProducer lazily initialise the producer instance