package com.airbnb.plog.listeners;

import com.airbnb.plog.EndOfPipeline;
import com.airbnb.plog.filters.FilterProvider;
import com.airbnb.plog.sinks.ConsoleSink;
import com.airbnb.plog.sinks.KafkaSink;
import com.airbnb.plog.sinks.Sink;
import com.airbnb.plog.stats.SimpleStatisticsReporter;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigValue;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import kafka.javaapi.producer.Producer;
import kafka.producer.ProducerConfig;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Constructor;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Map;
import java.util.Properties;

@Slf4j
public abstract class Listener {
    @Getter
    private final Config config;

    @Getter
    private final Sink sink;

    @Getter
    private final SimpleStatisticsReporter stats;

    @Getter
    private final EndOfPipeline eopHandler;

    public Listener(int id, Config config)
            throws UnknownHostException {
        this.config = config;

        final String clientId = "plog_" + InetAddress.getLocalHost().getHostName() + "_" + id;

        this.stats = new SimpleStatisticsReporter(clientId);

        this.eopHandler = new EndOfPipeline(stats);

        final String topic = config.getString("topic");
        if ("STDOUT".equals(topic)) {
            log.info("Using STDOUT");
            this.sink = new ConsoleSink();
        } else {
            final Properties kafkaProperties = new Properties();
            for (Map.Entry<String, ConfigValue> kv : config.getConfig("kafka").entrySet())
                kafkaProperties.put(kv.getKey(), kv.getValue().unwrapped().toString());
            kafkaProperties.put("client.id", clientId);

            log.info("Using Kafka with properties {}", kafkaProperties);

            final Producer<byte[], byte[]> producer = new Producer<byte[], byte[]>(new ProducerConfig(kafkaProperties));
            this.sink = new KafkaSink(topic, producer, stats);
        }
    }

    public abstract ChannelFuture start(final EventLoopGroup group);

    void appendFilters(ChannelPipeline pipeline)
            throws Exception {
        for (Config filterConfig : config.getConfigList("filters")) {
            final String providerName = filterConfig.getString("provider");
            log.debug("Loading provider for {}", providerName);

            final Class<?> providerClass = Class.forName(providerName);
            final Constructor<?> providerConstructor = providerClass.getConstructor();
            final FilterProvider provider = (FilterProvider) providerConstructor.newInstance();

            pipeline.addLast(provider.getFilter(filterConfig));
        }
    }
}

@@ -1162,83 +1162,124 @@
 ing 
-clientId = %22plog_%22 + InetAddress.getLocalHost().getHostName() + %22_%22 + id;%0A%0A
+topic = config.getString(%22topic%22);%0A        if (%22STDOUT%22.equals(topic)) %7B%0A            log.info(%22Using STDOUT%22);%0A%0A    
     
@@ -1324,28 +1324,27 @@
 eporter(
-clientId);%0A%0A
+null);%0A    
         
@@ -1390,141 +1390,8 @@
 ts);
-%0A%0A        final String topic = config.getString(%22topic%22);%0A        if (%22STDOUT%22.equals(topic)) %7B%0A            log.info(%22Using STDOUT%22);
 %0A   
@@ -1695,61 +1695,282 @@
 ));%0A
-            kafkaProperties.put(%22client.id%22, clientId
+%0A            final String clientId = %22plog_%22 + InetAddress.getLocalHost().getHostName() + %22_%22 + id;%0A            kafkaProperties.put(%22client.id%22, clientId);%0A            this.stats = new SimpleStatisticsReporter(clientId);%0A            this.eopHandler = new EndOfPipeline(stats
 );%0A%0A

use null kafka topic for STDOUT