package nl.us2.cloudpelican.stormprocessor;

/**
 * Created by robin on 07/06/15.
 */

import backtype.storm.Config;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import org.apache.commons.codec.binary.Base64;
import org.apache.storm.http.HttpResponse;
import org.apache.storm.http.client.HttpClient;
import org.apache.storm.http.client.methods.HttpGet;
import org.apache.storm.http.client.methods.HttpPut;
import org.apache.storm.http.impl.client.HttpClientBuilder;
import org.apache.storm.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.starter.util.TupleHelpers;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.regex.Matcher;

import static backtype.storm.utils.Utils.DEFAULT_STREAM_ID;

/**
 *
 * @author robin
 */
public class ResultSupervisorWriterBolt extends BaseRichBolt {

    OutputCollector _collector;
    HashMap<String, ArrayList<String>> resultAggregator;
    private HashMap<String, String> settings;

    private static final Logger LOG = LoggerFactory.getLogger(ResultSupervisorWriterBolt.class);

    public ResultSupervisorWriterBolt(HashMap<String, String> settings) {
        super();
        this.settings = settings;
    }

    public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
        _collector = collector;
        resultAggregator = new HashMap<String, ArrayList<String>>();
    }

    public void execute(Tuple tuple) {
        if (TupleHelpers.isTickTuple(tuple)) {
            executeTick();
        } else {
            executeTuple(tuple);
        }
        _collector.ack(tuple);
    }

    public void executeTick() {
        _flush();
    }

    protected void _flush() {
        for (Map.Entry<String, ArrayList<String>> kv : resultAggregator.entrySet()) {
            try {
                HttpClient client = HttpClientBuilder.create().build();

                String url = settings.get("supervisor_host") + "filter/" + kv.getKey() + "/result";
                LOG.info(url);
                HttpPut put = new HttpPut(url);
                String token = new String(Base64.encodeBase64((settings.get("supervisor_username") + ":" + settings.get("supervisor_password")).getBytes()));
                LOG.debug(token);
                put.setHeader("Authorization", "Basic " + token);
                HttpResponse resp = client.execute(put);
                int status = resp.getStatusLine().getStatusCode();
                if (status >= 400) {
                    throw new Exception("Invalid status " + status);
                }
            } catch (Exception e) {
                LOG.error("Failed to write data to supervisor", e);
            }
        }
        resultAggregator.clear();
    }

    public Map<String, Object> getComponentConfiguration() {
        Config conf = new Config();
        int tickFrequencyInSeconds = 1;
        conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, tickFrequencyInSeconds);
        return conf;
    }


    public void executeTuple(Tuple tuple) {
        String filterId = tuple.getStringByField("filter_id");
        String msg = tuple.getStringByField("msg");

        // Append in-memory
        if (!resultAggregator.containsKey(filterId)) {
            resultAggregator.put(filterId, new ArrayList<String>());
        }
        resultAggregator.get(filterId).add(msg);

        // @todo Flush if we have a lot of messages in memory

        // No ack, is handled in outer
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }
}
@@ -595,16 +595,57 @@
 Base64;%0A
+import org.apache.storm.http.HttpEntity;%0A
 import o
@@ -833,16 +833,66 @@
 ttpPut;%0A
+import org.apache.storm.http.entity.StringEntity;%0A
 import o
@@ -2585,24 +2585,310 @@
 tpPut(url);%0A
+                StringBuilder sb = new StringBuilder();%0A                for (String line : kv.getValue()) %7B%0A                    sb.append(line).append(%22%5Cn%22);%0A                %7D%0A                StringEntity entity = new StringEntity(sb.toString());%0A                put.setEntity(entity);%0A
             

Send data