// Path: src/main/java/edu/sjsu/mithai/config/Configuration.java
// public class Configuration {
// 
//     protected String propertyFile;
//     protected Properties properties;
// 
//     public Configuration(String propertyFile) throws IOException {
//         this.propertyFile = propertyFile;
//         this.properties = new Properties();
//         properties.load(new FileReader(propertyFile));
//     }
// 
//     public String getProperty(String key) {
//         return properties.getProperty(key);
//     }
// 
//     public void setProperty(String key, String value) {
//         properties.setProperty(key, value);
//     }
// 
//     public Properties getProperties() {
//         return properties;
//     }
// 
//     public void setProperties(Properties properties) {
//         this.properties = properties;
//     }
// 
//     public void setPropertyFile(String propertyFile) {
//         this.propertyFile = propertyFile;
//     }
// 
//     public String getPropertyFile() {
//         return propertyFile;
//     }
// 
//     public void reload() throws IOException {
//         properties.clear();
//         properties.load(new FileReader(propertyFile));
//         System.out.println("=>" + properties);
//     }
// }
// 
// Path: src/main/java/edu/sjsu/mithai/export/http/HttpExporter.java
// public class HttpExporter implements IExporter<ExportMessage> {
// 
//     private CloseableHttpClient client;
// 
//     public HttpExporter() {
//     }
// 
//     @Override
//     public void setup() throws Exception {
//         this.client = HttpClients.createDefault();
//     }
// 
//     @Override
//     public void send(ExportMessage message) throws IOException {
// 
//         if (message instanceof HttpExportMessage) {
//             HttpExportMessage message1 = (HttpExportMessage) message;
//             System.out.println("Sending message over HTTP: " + message);
//             System.out.println("URI:" + message1.getUri());
//             HttpPost post = new HttpPost(message1.getUri());
//             post.addHeader("content-type", "application/json");
//             post.setEntity(new StringEntity(message1.getMessage()));
//             CloseableHttpResponse response = client.execute(post);
//             System.out.println(response.getStatusLine());
//             EntityUtils.consume(response.getEntity());
//         } else {
//             System.out.println("HTTP message of type ExportMessage: " + message);
//         }
//     }
// 
//     @Override
//     public void tearDown() throws IOException {
//         client.close();
//     }
// }
// 
// Path: src/main/java/edu/sjsu/mithai/util/StoppableRunnableTask.java
// public abstract class StoppableRunnableTask implements Runnable, Stoppable {
// 
//     protected boolean stop;
// 
//     public StoppableRunnableTask() {
//         this.stop = false;
//     }
// 
//     @Override
//     public void stop() {
//         stop = true;
//     }
// 
// }


import edu.sjsu.mithai.config.Configuration;
import edu.sjsu.mithai.export.http.HttpExporter;
import edu.sjsu.mithai.spark.Store;
import edu.sjsu.mithai.util.StoppableRunnableTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
    public HttpExporterTask(Configuration configuration) throws Exception {
        this.configuration = configuration;
        this.messageStore = Store.httpMessageStore();
        exporter = new HttpExporter();
        exporter.setup();
    }

    @Override
    public void run() {

        while (true) {
            try {
                HttpExportMessage message = messageStore.getMessageQueue().take();

                if (message.getMessage().equals(STOP_MESSAGE)) {
                    break;
                }
                exporter.send(message);
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }
    }

    @Override
    public void stop() {
        super.stop();
        try {
// Path: src/main/java/edu/sjsu/mithai/config/Configuration.java
// public class Configuration {
// 
//     protected String propertyFile;
//     protected Properties properties;
// 
//     public Configuration(String propertyFile) throws IOException {
//         this.propertyFile = propertyFile;
//         this.properties = new Properties();
//         properties.load(new FileReader(propertyFile));
//     }
// 
//     public String getProperty(String key) {
//         return properties.getProperty(key);
//     }
// 
//     public void setProperty(String key, String value) {
//         properties.setProperty(key, value);
//     }
// 
//     public Properties getProperties() {
//         return properties;
//     }
// 
//     public void setProperties(Properties properties) {
//         this.properties = properties;
//     }
// 
//     public void setPropertyFile(String propertyFile) {
//         this.propertyFile = propertyFile;
//     }
// 
//     public String getPropertyFile() {
//         return propertyFile;
//     }
// 
//     public void reload() throws IOException {
//         properties.clear();
//         properties.load(new FileReader(propertyFile));
//         System.out.println("=>" + properties);
//     }
// }
// 
// Path: src/main/java/edu/sjsu/mithai/export/http/HttpExporter.java
// public class HttpExporter implements IExporter<ExportMessage> {
// 
//     private CloseableHttpClient client;
// 
//     public HttpExporter() {
//     }
// 
//     @Override
//     public void setup() throws Exception {
//         this.client = HttpClients.createDefault();
//     }
// 
//     @Override
//     public void send(ExportMessage message) throws IOException {
// 
//         if (message instanceof HttpExportMessage) {
//             HttpExportMessage message1 = (HttpExportMessage) message;
//             System.out.println("Sending message over HTTP: " + message);
//             System.out.println("URI:" + message1.getUri());
//             HttpPost post = new HttpPost(message1.getUri());
//             post.addHeader("content-type", "application/json");
//             post.setEntity(new StringEntity(message1.getMessage()));
//             CloseableHttpResponse response = client.execute(post);
//             System.out.println(response.getStatusLine());
//             EntityUtils.consume(response.getEntity());
//         } else {
//             System.out.println("HTTP message of type ExportMessage: " + message);
//         }
//     }
// 
//     @Override
//     public void tearDown() throws IOException {
//         client.close();
//     }
// }
// 
// Path: src/main/java/edu/sjsu/mithai/util/StoppableRunnableTask.java
// public abstract class StoppableRunnableTask implements Runnable, Stoppable {
// 
//     protected boolean stop;
// 
//     public StoppableRunnableTask() {
//         this.stop = false;
//     }
// 
//     @Override
//     public void stop() {
//         stop = true;
//     }
// 
// }


// Path: src/main/java/edu/sjsu/mithai/export/HttpExporterTask.java
import edu.sjsu.mithai.config.Configuration;
import edu.sjsu.mithai.export.http.HttpExporter;
import edu.sjsu.mithai.spark.Store;
import edu.sjsu.mithai.util.StoppableRunnableTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;

    public HttpExporterTask(Configuration configuration) throws Exception {
        this.configuration = configuration;
        this.messageStore = Store.httpMessageStore();
        exporter = new HttpExporter();
        exporter.setup();
    }

    @Override
    public void run() {

        while (true) {
            try {
                HttpExportMessage message = messageStore.getMessageQueue().take();

                if (message.getMessage().equals(STOP_MESSAGE)) {
                    break;
                }
                exporter.send(message);
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }
    }

    @Override
    public void stop() {
        super.stop();
        try {
            exporter.tearDown();