package storm.ml.bolt;

import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;

public class EvaluationBolt extends BaseBasicBolt {
    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        List<Double> result = new ArrayList<Double>();
        result.add(0.0);
        result.add(0.0);

        collector.emit(new Values(tuple.getValue(0), result));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("id", "result"));
    }
}

@@ -310,16 +310,42 @@
 ayList;%0A
+import java.util.Scanner;%0A
 import j
@@ -493,24 +493,176 @@
 ollector) %7B%0A
+        List%3CDouble%3E weights = new ArrayList%3CDouble%3E();%0A        weights.add(1.0);%0A        weights.add(2.0);%0A%0A        String input = tuple.getString(1);%0A
         List
@@ -704,24 +704,25 @@
 ble%3E();%0A
+%0A
         
 result.a
@@ -717,47 +717,197 @@
     
-result.add(0.0);%0A        result.add(0.0
+Scanner scanner = new Scanner(input.substring(1, input.length()-1));%0A        scanner.useDelimiter(%22, %22);%0A%0A        while (scanner.hasNextDouble())%0A            result.add(scanner.nextDouble()
 );%0A%0A

parse input