bda-datastore/src/main/java/gr/ntua/ece/cslab/selis/bda/datastore/StorageBackend.java
public String insert(Message message) throws Exception {
        // Convert message to appropriate format taking into account the schema
        /*JSONObject json = new JSONObject(); // to store blob
        List<String> fields = ELconnector.describe("").getSchema().getColumnNames();
        HashMap<String, String> msg = new HashMap<>();
        for (KeyValue element : message.getEntries()) {
            if (!fields.contains(element.getKey()))
                json.put(element.getKey(), element.getValue());
            else
                msg.put(element.getKey(), element.getValue());
        }
        for (String column : fields)
            if (!msg.containsKey(column))
                msg.put(column, "null");
        msg.put("payload", json.toJSONString());
        msg.put("event_timestamp", String.valueOf(LocalDateTime.now()));

        if (msg.containsKey("payload") && msg.containsKey("message_type") && msg.containsKey("scn_slug") && msg.size() == 4)
            throw new Exception("Message does not contain any foreign keys.");
        else if (json.isEmpty() || !msg.containsKey("message_type") || msg.size() < 4)
            throw new Exception("Message contains strange event format. Append aborted.");*/
        return ELconnector.put(message);
    }
/** Insert a new message in the EventLog.
     *  This method takes as input a Message and saves each key that matches with an EventLog column name in the
     *  relevant column of the eventLog table, while all the non-matching keys are saved as a blob in json format
     *  in the 'message' column of the eventLog table. **/
Insert a new message in the EventLog.
This method takes as input a Message and saves each key that matches with an EventLog column name in the
relevant column of the eventLog table, while all the non-matching keys are saved as a blob in json format
in the 'message' column of the eventLog table.
Insert a new message in the EventLog . This method takes as input a Message and saves each key that matches with an EventLog column name in the relevant column of the eventLog table while all the non - matching keys are saved as a blob in json format in the ' message ' column of the eventLog table .
public String insert(Message message) throws Exception {
        /*JSONObject json = new JSONObject(); // to store blob
        List<String> fields = ELconnector.describe("").getSchema().getColumnNames();
        HashMap<String, String> msg = new HashMap<>();
        for (KeyValue element : message.getEntries()) {
            if (!fields.contains(element.getKey()))
                json.put(element.getKey(), element.getValue());
            else
                msg.put(element.getKey(), element.getValue());
        }
        for (String column : fields)
            if (!msg.containsKey(column))
                msg.put(column, "null");
        msg.put("payload", json.toJSONString());
        msg.put("event_timestamp", String.valueOf(LocalDateTime.now()));
        if (msg.containsKey("payload") && msg.containsKey("message_type") && msg.containsKey("scn_slug") && msg.size() == 4)
            throw new Exception("Message does not contain any foreign keys.");
        else if (json.isEmpty() || !msg.containsKey("message_type") || msg.size() < 4)
            throw new Exception("Message contains strange event format. Append aborted.");*/
        return ELconnector.put(message);
    }
public String insert ( Message message ) throws Exception { /*JSONObject json = new JSONObject(); // to store blob
        List<String> fields = ELconnector.describe("").getSchema().getColumnNames();
        HashMap<String, String> msg = new HashMap<>();
        for (KeyValue element : message.getEntries()) {
            if (!fields.contains(element.getKey()))
                json.put(element.getKey(), element.getValue());
            else
                msg.put(element.getKey(), element.getValue());
        }
        for (String column : fields)
            if (!msg.containsKey(column))
                msg.put(column, "null");
        msg.put("payload", json.toJSONString());
        msg.put("event_timestamp", String.valueOf(LocalDateTime.now()));

        if (msg.containsKey("payload") && msg.containsKey("message_type") && msg.containsKey("scn_slug") && msg.size() == 4)
            throw new Exception("Message does not contain any foreign keys.");
        else if (json.isEmpty() || !msg.containsKey("message_type") || msg.size() < 4)
            throw new Exception("Message contains strange event format. Append aborted.");*/ return ELconnector . put ( message ) ; }
Insert a new message in the EventLog.
Insert a new message in the EventLog .
// Convert message to appropriate format taking into account the schema