package it.uniroma2.ember.elasticsearch;

import it.uniroma2.ember.utils.LampEMAConsumption;
import it.uniroma2.ember.utils.StreetLamp;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch2.RequestIndexer;
import org.codehaus.jackson.map.ObjectMapper;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;


public class EmberElasticsearchSinkFunction implements ElasticsearchSinkFunction<Object> {

    private String index = "";
    private String type = "";

    /**
     * This method can be used to create an IndexRequest
     *
     * @param element, the StreetLamp to store
     */
    private IndexRequest createIndexRequest(Object element) throws Exception {

        byte[] convertedElem = new ObjectMapper().writeValueAsBytes(element);
        if (element instanceof StreetLamp) {
            StreetLamp elem = (StreetLamp) element;
            // creating update request
            return Requests.indexRequest()
                    .index(index)
                    .type(type)
                    .id(String.valueOf(elem.getId()))
                    .timestamp(String.valueOf(elem.getSent()))
                    .source(convertedElem);
        } else {
            LampEMAConsumption elem = (LampEMAConsumption) element;
            // creating update request
            return Requests.indexRequest()
                    .index(index)
                    .type(type)
                    .id(String.valueOf(elem.getId()))
                    .source(convertedElem);
        }
    }

    /**
     * @param element,        a StreetLamp object
     * @param runtimeContext, the runtime context used by Flink Elasticsearch connector
     * @param indexer,        the IndexRequests processor
     */
    @Override
    public void process(Object element, RuntimeContext runtimeContext, RequestIndexer indexer) {
        try {
            indexer.add(createIndexRequest(element));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public EmberElasticsearchSinkFunction(String index, String type) {
        this.index = index;
        this.type = type;
    }
}

@@ -494,16 +494,43 @@
 uests;%0A%0A
+import java.time.Instant;%0A%0A
 %0Apublic 
@@ -1483,32 +1483,115 @@
  update request%0A
+            // no id in this case to maintain a history for the consumption values%0A
             retu
@@ -1696,34 +1696,41 @@
                .
-id
+timestamp
 (String.valueOf(
@@ -1725,33 +1725,51 @@
 ing.valueOf(
-elem.getI
+Instant.now().getEpochSecon
 d()))%0A      

Add support for timestamp based indexing of new lamps.