// Path: src/main/java/edu/sjsu/mithai/main/Mithai.java
// public class Mithai implements Observer {
// 
//     protected static Configuration configuration;
//     protected SensorStore sensorStore;
// 
// 
//     public static void main(String[] args) throws Exception {
//         Mithai mithai = new Mithai();
//         if(args.length<1)
//             mithai.start(null);
//         else
//             mithai.start(args[0]);
//     }
// 
//     public static Configuration getConfiguration() {
//         return configuration;
//     }
// 
//     protected void start(String arg) throws Exception {
// 
//         Logger.getLogger("org").setLevel(Level.ERROR);
//         Logger.getLogger("akka").setLevel(Level.ERROR);
// 
//         ConfigFileObservable.getInstance().addObserver(this);
// 
//         Runtime.getRuntime().addShutdownHook(new ShutDownHook());
// 
//         //TODO file path will be provided by user
//         if (arg == null || arg.equals("")) {
//             File configFile = new File(Mithai.class.getProtectionDomain().getCodeSource().getLocation().toURI().getPath());
//             configuration = new Configuration(configFile.getParent() + "/application.properties");
//         } else
//             configuration = new Configuration(arg);
// 
//         sensorStore = new SensorStore();
// 
//         loadDevices();
// 
//         setupHandlers();
// 
//         boolean receiverTask = false;
//         //Start tasks here
// //        TaskManager.getInstance().submitTask(new ConfigMonitorTask(configuration));
// 
//         String type = configuration.getProperty(MithaiProperties.TASK_LIST);
//         String[] tasks = type.split(",");
// 
//         for (String task : tasks) {
//             task = task.trim();
//             if (task.equals("MQTTDataReceiverTask")) {
//                 TaskManager.getInstance().submitTask(new MQTTDataReceiverTask(configuration));
//                 receiverTask = true;
//             } else if (task.equals("MQTTMetaDataRecieverTask")) {
//                 TaskManager.getInstance().submitTask(new MQTTMetaDataRecieverTask(configuration));
//                 receiverTask = true;
//             }
//             if (task.equals("MQTTPublisherTask")) {
//                 TaskManager.getInstance().submitTask(new MQTTPublisherTask(configuration));
//             }
//             if (task.equals("DataGenerationTask")) {
//                 TaskManager.getInstance().submitTask(new DataGenerationTask(configuration, sensorStore));
//             } else if (task.equals("MetadataGenerationTask")) {
//                 TaskManager.getInstance().submitTask(new MetadataGenerationTask(configuration));
//             }
// 
//         }
//         TaskManager.getInstance().submitTask(new HttpExporterTask(configuration));
// 
//         if (!configuration.getProperty(EXPORTER_TYPE).equals("HTTP")) {
//             TaskManager.getInstance().submitTask(new ExporterTask(configuration, Store.messageStore()));
//         }
// 
//         // Start Streaming context
// 
//         Thread.sleep(Long.parseLong(configuration.getProperty(MithaiProperties.STARTUP_THRESHOLD)) * 1000);
//         if (receiverTask) {
//             SparkStreamingObject.streamingContext().start();
//         }
// //        // Stop all tasks and wait 60 seconds to finish them
// //        TaskManager.getInstance().stopAll();
//     }
// 
//     protected synchronized void loadDevices() {
//         sensorStore.getDevices().clear();
// 
//         for (int i = 1; i<= Integer.parseInt(configuration.getProperty(NUMBER_OF_SENSORS)); i++) {
//             sensorStore.addDevice(new TemperatureSensor("sensor" + i));
//         }
//     }
// 
//     protected synchronized void setupHandlers() {
//         TaskManager.getInstance().addHandler(new MithaiHandler());
//     }
// 
//     @Override
//     public void update(Observable observable, Object o) {
// 
//         if (observable instanceof ConfigFileObservable) {
//             loadDevices();
// 
//             // Kick out old data generation task and start new one
//             TaskManager.getInstance().stop(DataGenerationTask.class);
// 
//             try {
//                 TaskManager.getInstance().submitTask(new DataGenerationTask(configuration, sensorStore));
//             } catch (IOException e) {
//                 e.printStackTrace();
//             }
//         }
//     }
// 
//     private static class ShutDownHook extends Thread {
// 
//         @Override
//         public void run() {
//             System.out.println("###Shutdown triggered.. Stopping all tasks..");
//             try {
//                 TaskManager.getInstance().stopAll();
//             } catch (InterruptedException e) {
//                 e.printStackTrace();
//             }
//         }
//     }
// }


import edu.sjsu.mithai.main.Mithai;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.*;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.log4j.Logger;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.Base64;
package edu.sjsu.mithai.data;



public class AvroSerializationHelper implements SerializationHelper<GenericRecord>{

    private Schema schema;
    private Logger logger = Logger.getLogger(getClass());

    public void loadSchema(String schemaFile) throws IOException, URISyntaxException {
        Schema.Parser parser = new Schema.Parser();
//        URL url = getClass().getClassLoader().getResource(schemaFile);
// Path: src/main/java/edu/sjsu/mithai/main/Mithai.java
// public class Mithai implements Observer {
// 
//     protected static Configuration configuration;
//     protected SensorStore sensorStore;
// 
// 
//     public static void main(String[] args) throws Exception {
//         Mithai mithai = new Mithai();
//         if(args.length<1)
//             mithai.start(null);
//         else
//             mithai.start(args[0]);
//     }
// 
//     public static Configuration getConfiguration() {
//         return configuration;
//     }
// 
//     protected void start(String arg) throws Exception {
// 
//         Logger.getLogger("org").setLevel(Level.ERROR);
//         Logger.getLogger("akka").setLevel(Level.ERROR);
// 
//         ConfigFileObservable.getInstance().addObserver(this);
// 
//         Runtime.getRuntime().addShutdownHook(new ShutDownHook());
// 
//         //TODO file path will be provided by user
//         if (arg == null || arg.equals("")) {
//             File configFile = new File(Mithai.class.getProtectionDomain().getCodeSource().getLocation().toURI().getPath());
//             configuration = new Configuration(configFile.getParent() + "/application.properties");
//         } else
//             configuration = new Configuration(arg);
// 
//         sensorStore = new SensorStore();
// 
//         loadDevices();
// 
//         setupHandlers();
// 
//         boolean receiverTask = false;
//         //Start tasks here
// //        TaskManager.getInstance().submitTask(new ConfigMonitorTask(configuration));
// 
//         String type = configuration.getProperty(MithaiProperties.TASK_LIST);
//         String[] tasks = type.split(",");
// 
//         for (String task : tasks) {
//             task = task.trim();
//             if (task.equals("MQTTDataReceiverTask")) {
//                 TaskManager.getInstance().submitTask(new MQTTDataReceiverTask(configuration));
//                 receiverTask = true;
//             } else if (task.equals("MQTTMetaDataRecieverTask")) {
//                 TaskManager.getInstance().submitTask(new MQTTMetaDataRecieverTask(configuration));
//                 receiverTask = true;
//             }
//             if (task.equals("MQTTPublisherTask")) {
//                 TaskManager.getInstance().submitTask(new MQTTPublisherTask(configuration));
//             }
//             if (task.equals("DataGenerationTask")) {
//                 TaskManager.getInstance().submitTask(new DataGenerationTask(configuration, sensorStore));
//             } else if (task.equals("MetadataGenerationTask")) {
//                 TaskManager.getInstance().submitTask(new MetadataGenerationTask(configuration));
//             }
// 
//         }
//         TaskManager.getInstance().submitTask(new HttpExporterTask(configuration));
// 
//         if (!configuration.getProperty(EXPORTER_TYPE).equals("HTTP")) {
//             TaskManager.getInstance().submitTask(new ExporterTask(configuration, Store.messageStore()));
//         }
// 
//         // Start Streaming context
// 
//         Thread.sleep(Long.parseLong(configuration.getProperty(MithaiProperties.STARTUP_THRESHOLD)) * 1000);
//         if (receiverTask) {
//             SparkStreamingObject.streamingContext().start();
//         }
// //        // Stop all tasks and wait 60 seconds to finish them
// //        TaskManager.getInstance().stopAll();
//     }
// 
//     protected synchronized void loadDevices() {
//         sensorStore.getDevices().clear();
// 
//         for (int i = 1; i<= Integer.parseInt(configuration.getProperty(NUMBER_OF_SENSORS)); i++) {
//             sensorStore.addDevice(new TemperatureSensor("sensor" + i));
//         }
//     }
// 
//     protected synchronized void setupHandlers() {
//         TaskManager.getInstance().addHandler(new MithaiHandler());
//     }
// 
//     @Override
//     public void update(Observable observable, Object o) {
// 
//         if (observable instanceof ConfigFileObservable) {
//             loadDevices();
// 
//             // Kick out old data generation task and start new one
//             TaskManager.getInstance().stop(DataGenerationTask.class);
// 
//             try {
//                 TaskManager.getInstance().submitTask(new DataGenerationTask(configuration, sensorStore));
//             } catch (IOException e) {
//                 e.printStackTrace();
//             }
//         }
//     }
// 
//     private static class ShutDownHook extends Thread {
// 
//         @Override
//         public void run() {
//             System.out.println("###Shutdown triggered.. Stopping all tasks..");
//             try {
//                 TaskManager.getInstance().stopAll();
//             } catch (InterruptedException e) {
//                 e.printStackTrace();
//             }
//         }
//     }
// }


// Path: src/main/scala/edu/sjsu/mithai/data/AvroSerializationHelper.java
import edu.sjsu.mithai.main.Mithai;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.*;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.log4j.Logger;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.Base64;

package edu.sjsu.mithai.data;



public class AvroSerializationHelper implements SerializationHelper<GenericRecord>{

    private Schema schema;
    private Logger logger = Logger.getLogger(getClass());

    public void loadSchema(String schemaFile) throws IOException, URISyntaxException {
        Schema.Parser parser = new Schema.Parser();
//        URL url = getClass().getClassLoader().getResource(schemaFile);
        File jarfile = new File(Mithai.class.getProtectionDomain().getCodeSource().getLocation().toURI().getPath());