package io.cozmic.usher.plugins.kafka;

import de.odysseus.el.util.SimpleContext;
import io.cozmic.usher.core.OutPipeline;
import io.cozmic.usher.core.OutputPlugin;
import io.cozmic.usher.message.PipelinePack;
import io.cozmic.usher.plugins.core.UsherInitializationFailedException;
import io.cozmic.usher.streams.*;
import io.vertx.core.*;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.json.JsonObject;
import io.vertx.core.streams.WriteStream;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.el.ExpressionFactory;
import javax.el.ValueExpression;
import java.util.Objects;
import java.util.Properties;

/**
 * Created by chuck on 10/1/15.
 */
public class KafkaOutput implements OutputPlugin {
    private JsonObject configObj;
    private static ExpressionFactory factory = ExpressionFactory.newInstance();
    private SimpleContext context = new SimpleContext();
    private ValueExpression topicExpression;
    private Vertx vertx;
    private KafkaProducer<byte[], byte[]> producer;
    private String topic;

    @Override
    public void run(AsyncResultHandler<DuplexStream<Buffer, Buffer>> duplexStreamAsyncResultHandler) {
        final KafkaProducerStream kafkaProducerStream = new KafkaProducerStream(vertx, topicExpression);
        final DuplexStream<Buffer, Buffer> duplexStream = new DuplexStream<>(NullReadStream.getInstance(), kafkaProducerStream);
        duplexStreamAsyncResultHandler.handle(Future.succeededFuture(duplexStream));
    }

    @Override
    public void stop(OutPipeline outPipeline) {

    }

    @Override
    public void init(JsonObject configObj, Vertx vertx) throws UsherInitializationFailedException {

        this.configObj = configObj;
        this.vertx = vertx;
        Properties kafkaProducerProps = new Properties();
        topic = configObj.getString("topic", "default");
        topicExpression = factory.createValueExpression(context, topic, String.class);
        kafkaProducerProps.put("bootstrap.servers", configObj.getString("bootstrap.servers", "kafka.dev:9092"));
        kafkaProducerProps.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
        kafkaProducerProps.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

        producer = new KafkaProducer<>(kafkaProducerProps);


    }

    private class KafkaProducerStream implements ClosableWriteStream<Buffer> {
        private final Vertx vertx;
        private ValueExpression topicExpression;

        public KafkaProducerStream(Vertx vertx, ValueExpression topicExpression) {

            this.vertx = vertx;
            this.topicExpression = topicExpression;
        }

        @Override
        public KafkaProducerStream exceptionHandler(Handler<Throwable> handler) {
            return this;
        }

        @Override
        public WriteStream<Buffer> write(Buffer data) {
            return this;
        }

        @Override
        public WriteStream<Buffer> setWriteQueueMaxSize(int maxSize) {
            return this;
        }

        @Override
        public boolean writeQueueFull() {
            return false;
        }

        @Override
        public WriteStream<Buffer> drainHandler(Handler<Void> handler) {
            return this;
        }


        /**
         * Send message to Kafka with async callback on the vertx eventloop.
         *
         * @param data
         * @param context
         * @return
         */
        @Override
        public AsyncWriteStream<Buffer> write(Buffer data, Future<Void> future, PipelinePack context) {
            Objects.requireNonNull(data, "Must provide data to write to Kafka");
            SimpleContext runtimeContext = new SimpleContext();
            factory.createValueExpression(runtimeContext, "${pack}", PipelinePack.class).setValue(runtimeContext, context);
            final String dynamicTopic = (String) topicExpression.getValue(runtimeContext);
            ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(dynamicTopic, data.getBytes());
            try {
                producer.send(record, (metadata, exception) -> vertx.runOnContext(v -> {
                    if (exception != null) {
                        future.fail(exception);
                        return;
                    }
                    future.complete();
                }));
            } catch (Exception ex) {
                future.fail(ex);
            }

            return this;
        }

        @Override
        public void close() {
            //no op
        }
    }
}

@@ -310,33 +310,296 @@
 ams.
-*;%0Aimport io.vertx.core.*
+AsyncWriteStream;%0Aimport io.cozmic.usher.streams.ClosableWriteStream;%0Aimport io.cozmic.usher.streams.DuplexStream;%0Aimport io.cozmic.usher.streams.NullReadStream;%0Aimport io.vertx.core.AsyncResultHandler;%0Aimport io.vertx.core.Future;%0Aimport io.vertx.core.Handler;%0Aimport io.vertx.core.Vertx
 ;%0Aim
@@ -4364,24 +4364,79 @@
 etBytes());%0A
+            vertx.executeBlocking(blockFuture -%3E %7B%0A    
             
@@ -4441,16 +4441,20 @@
   try %7B%0A
+    
         
@@ -4512,34 +4512,14 @@
  -%3E 
-vertx.runOnContext(v -%3E %7B%0A
+%7B%0A    
     
@@ -4575,33 +4575,42 @@
                 
-f
+    blockF
 uture.fail(excep
@@ -4644,16 +4644,20 @@
     
+   
+ 
 return;%0A
     
@@ -4652,16 +4652,20 @@
 return;%0A
+    
         
@@ -4694,25 +4694,34 @@
             
-f
+    blockF
 uture.comple
@@ -4746,13 +4746,20 @@
     
+  
+  
 %7D)
-)
 ;%0A
+    
     
@@ -4811,22 +4811,268 @@
     
-future.fail(ex
+    blockFuture.fail(ex);%0A                %7D%0A            %7D, false, asyncResult -%3E %7B%0A                if (asyncResult.failed()) %7B%0A                    future.fail(asyncResult.cause());%0A                    return;%0A                %7D%0A%0A                future.complete(
 );%0A 
@@ -5075,32 +5075,35 @@
 );%0A            %7D
+);%0A
 %0A%0A            re

Update kafkaoutput to use vertx threading helpers