// Path: src/main/java/net/redborder/cep/sources/Source.java
// public abstract class Source {
//     private static final Logger log = LogManager.getLogger(Source.class);
// 
//     // The instance of disruptor that will consume the events that are produced by this source
//     public EventProducer eventProducer;
// 
//     // The instance of the parsers manager that will serve parser instances in order to parse
//     // the events that come from this source
//     public ParsersManager parsersManager;
// 
//     // The list of properties associated with this instance on the config file
//     public Map<String, Object> properties;
// 
//     /**
//      * Create a new source.
//      * <p>This method will prepare the instance with some needed variables
//      * in order to be started later with the start method (implemented by children).
//      *
//      * @param parsersManager Instance of ParserManager that will serve parsers to this source instance
//      * @param eventHandler Instance of EventHandler that will receive the events generated by this source instance
//      * @param properties Map of properties associated with this source
//      */
// 
//     public Source(ParsersManager parsersManager, EventHandler eventHandler, Map<String, Object> properties) {
//         // Save the references for later use
//         this.parsersManager = parsersManager;
//         this.properties = properties;
// 
//         // Create the ring buffer for this topic and start it
//         Disruptor<MapEvent> disruptor = new Disruptor<>(new MapEventFactory(), ConfigData.getRingBufferSize(), Executors.newCachedThreadPool());
//         disruptor.handleEventsWith(eventHandler);
//         disruptor.start();
// 
//         // Create the event producer that will receive the events produced by
//         // this source instance
//         eventProducer = new EventProducer(disruptor.getRingBuffer());
//         prepare();
//     }
// 
//     /**
//      * Sends a new string message to the stream.
//      * <p>The param msg should be a string that will be parsed by the parser manager. The parser manager
//      * stores a reference of the parser that must be used with each stream, so when a string message
//      * is sent, the parser manager parses it to convert it into a map object.
//      * <p>This method should be called when the source wants to generate a message to the stream.
//      *
//      * @param streamName The stream that will receive the message
//      * @param msg The string message that will be sent
//      */
// 
//     public void send(String streamName, String msg) {
//         Map<String, Object> data = parsersManager.parse(streamName, msg);
//         eventProducer.putData(streamName, data);
//     }
// 
//     /**
//      * Sends a new map message to the producer.
//      *
//      * @param streamName The stream that will receive the message
//      * @param data The message that will be sent
//      */
// 
//     public void send(String streamName, Map<String, Object> data) {
//         eventProducer.putData(streamName, data);
//     }
// 
//     /**
//      * Gets a property from the source properties.
//      * <p>The source properties are a series of optional properties that are specified on the config file
//      * and associated with each stream.
//      *
//      * @param propertyName The property that will be returned
//      * @return The value associated with the property name specified
//      */
// 
//     public Object getProperty(String propertyName){
//         return properties.get(propertyName);
//     }
// 
//     /**
//      * This method is called by the SourceManager to add one or more new streams to the source.
//      * <p>The source implementation must process the streams and do whatever it needs to do to add
//      * that stream to the source. For example, you could create a new consumer thread for each stream
//      * received, open a new HTTP connection to some external third-party API, etc.
//      * <p>The streams that will be passed to the function addStreams of the implementations are
//      * specified on the config file by the user.
//      *
//      * @param streamName One or more streams that has been added in the source.
//      * @see SourcesManager
//      */
// 
//     public abstract void addStreams(String ... streamName);
// 
//     /**
//      * This method is called automatically by the Source constructor after finishing, so your
//      * implementation can use it to prepare for the following start. For example, you could use it
//      * to establish a connection, open some resource, etc.
//      */
// 
//     public abstract void prepare();
// 
//     /**
//      * This method will be called when the Source should be started, so after this call, the source
//      * must start to generate messages to the producers using the send methods provided by this class.
//      */
// 
//     public abstract void start();
// 
//     /**
//      * This method will be called when the Source must stop, in order to close connections and
//      * release resources that will no longer be useful.
//      */
// 
//     public abstract void shutdown();
// }
// 
// Path: src/main/java/net/redborder/cep/sources/parsers/Parser.java
// public interface Parser {
// 
//     /**
//      * This method receives a message string and produces a
//      * java map from the string given.
//      *
//      * @param message The string message that will be parsed
//      * @return A map that is the result of parsing the given string
//      */
// 
//     Map<String, Object> parse(String message);
// }


import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import net.redborder.cep.sources.Source;
import net.redborder.cep.sources.parsers.Parser;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.UnsupportedEncodingException;
import java.util.Map;
package net.redborder.cep.sources.kafka;



/**
 * A thread that will consume messages from a given partition from a topic.
 * This thread will forward the read messages to a source, that gets from a
 * given topic.
 *
 * @see Topic
 * @see Source
 */

public class Consumer implements Runnable {
    private final static Logger log = LogManager.getLogger(Consumer.class);

    // The KafkaStream from where this thread will consume messages
    // This object represents a partition on a topic.
    private KafkaStream stream;

    // The partition's topic
    private Topic topic;

    // The parser that will be used to parse the consumed messages
    private Parser parser;

    // The source that will receive the consumed messages
// Path: src/main/java/net/redborder/cep/sources/Source.java
// public abstract class Source {
//     private static final Logger log = LogManager.getLogger(Source.class);
// 
//     // The instance of disruptor that will consume the events that are produced by this source
//     public EventProducer eventProducer;
// 
//     // The instance of the parsers manager that will serve parser instances in order to parse
//     // the events that come from this source
//     public ParsersManager parsersManager;
// 
//     // The list of properties associated with this instance on the config file
//     public Map<String, Object> properties;
// 
//     /**
//      * Create a new source.
//      * <p>This method will prepare the instance with some needed variables
//      * in order to be started later with the start method (implemented by children).
//      *
//      * @param parsersManager Instance of ParserManager that will serve parsers to this source instance
//      * @param eventHandler Instance of EventHandler that will receive the events generated by this source instance
//      * @param properties Map of properties associated with this source
//      */
// 
//     public Source(ParsersManager parsersManager, EventHandler eventHandler, Map<String, Object> properties) {
//         // Save the references for later use
//         this.parsersManager = parsersManager;
//         this.properties = properties;
// 
//         // Create the ring buffer for this topic and start it
//         Disruptor<MapEvent> disruptor = new Disruptor<>(new MapEventFactory(), ConfigData.getRingBufferSize(), Executors.newCachedThreadPool());
//         disruptor.handleEventsWith(eventHandler);
//         disruptor.start();
// 
//         // Create the event producer that will receive the events produced by
//         // this source instance
//         eventProducer = new EventProducer(disruptor.getRingBuffer());
//         prepare();
//     }
// 
//     /**
//      * Sends a new string message to the stream.
//      * <p>The param msg should be a string that will be parsed by the parser manager. The parser manager
//      * stores a reference of the parser that must be used with each stream, so when a string message
//      * is sent, the parser manager parses it to convert it into a map object.
//      * <p>This method should be called when the source wants to generate a message to the stream.
//      *
//      * @param streamName The stream that will receive the message
//      * @param msg The string message that will be sent
//      */
// 
//     public void send(String streamName, String msg) {
//         Map<String, Object> data = parsersManager.parse(streamName, msg);
//         eventProducer.putData(streamName, data);
//     }
// 
//     /**
//      * Sends a new map message to the producer.
//      *
//      * @param streamName The stream that will receive the message
//      * @param data The message that will be sent
//      */
// 
//     public void send(String streamName, Map<String, Object> data) {
//         eventProducer.putData(streamName, data);
//     }
// 
//     /**
//      * Gets a property from the source properties.
//      * <p>The source properties are a series of optional properties that are specified on the config file
//      * and associated with each stream.
//      *
//      * @param propertyName The property that will be returned
//      * @return The value associated with the property name specified
//      */
// 
//     public Object getProperty(String propertyName){
//         return properties.get(propertyName);
//     }
// 
//     /**
//      * This method is called by the SourceManager to add one or more new streams to the source.
//      * <p>The source implementation must process the streams and do whatever it needs to do to add
//      * that stream to the source. For example, you could create a new consumer thread for each stream
//      * received, open a new HTTP connection to some external third-party API, etc.
//      * <p>The streams that will be passed to the function addStreams of the implementations are
//      * specified on the config file by the user.
//      *
//      * @param streamName One or more streams that has been added in the source.
//      * @see SourcesManager
//      */
// 
//     public abstract void addStreams(String ... streamName);
// 
//     /**
//      * This method is called automatically by the Source constructor after finishing, so your
//      * implementation can use it to prepare for the following start. For example, you could use it
//      * to establish a connection, open some resource, etc.
//      */
// 
//     public abstract void prepare();
// 
//     /**
//      * This method will be called when the Source should be started, so after this call, the source
//      * must start to generate messages to the producers using the send methods provided by this class.
//      */
// 
//     public abstract void start();
// 
//     /**
//      * This method will be called when the Source must stop, in order to close connections and
//      * release resources that will no longer be useful.
//      */
// 
//     public abstract void shutdown();
// }
// 
// Path: src/main/java/net/redborder/cep/sources/parsers/Parser.java
// public interface Parser {
// 
//     /**
//      * This method receives a message string and produces a
//      * java map from the string given.
//      *
//      * @param message The string message that will be parsed
//      * @return A map that is the result of parsing the given string
//      */
// 
//     Map<String, Object> parse(String message);
// }


// Path: src/main/java/net/redborder/cep/sources/kafka/Consumer.java
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import net.redborder.cep.sources.Source;
import net.redborder.cep.sources.parsers.Parser;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.UnsupportedEncodingException;
import java.util.Map;

package net.redborder.cep.sources.kafka;



/**
 * A thread that will consume messages from a given partition from a topic.
 * This thread will forward the read messages to a source, that gets from a
 * given topic.
 *
 * @see Topic
 * @see Source
 */

public class Consumer implements Runnable {
    private final static Logger log = LogManager.getLogger(Consumer.class);

    // The KafkaStream from where this thread will consume messages
    // This object represents a partition on a topic.
    private KafkaStream stream;

    // The partition's topic
    private Topic topic;

    // The parser that will be used to parse the consumed messages
    private Parser parser;

    // The source that will receive the consumed messages
    private Source source;