package zeit.recommend;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

/**
 * zeit.recommend.Recommender
 *
 * This class has no description.
 *
 * Copyright: (c) 2013 by Nicolas Drebenstedt.
 * License: BSD, see LICENSE for more details.
 */

public class Recommender {

  public static void main(String[] args) throws Exception {
    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout(
        "zonapi",
        new PythonSpout("zonapi.py", "path"),
        1);

     builder.setBolt(
        "item",
        new PythonBolt("item.py", "path"),
        1).shuffleGrouping("zonapi");

    builder.setSpout(
        "rabbitmq",
        new PythonSpout("rabbitmq.py", "timestamp", "path", "user"),
        1);

    builder.setBolt(
        "user",
        new PythonBolt("user.py", "user", "paths"),
        1).shuffleGrouping("rabbitmq");

    builder.setBolt(
        "recommendation",
        new PythonBolt("recommendation.py", "user", "events", "recommendations"),
        1).shuffleGrouping("user");

    builder.setBolt(
        "outlet",
        new PythonBolt("outlet.py"),
        1).shuffleGrouping("recommendation");

    Config conf = new Config();
    conf.setDebug(false);
    conf.setMaxTaskParallelism(1);

    // TODO: Read config data from file.
    // conf.put("zeit.recommend.elasticsearch.host", "localhost");
    conf.put("zeit.recommend.elasticsearch.host", "217.13.68.236");
    conf.put("zeit.recommend.elasticsearch.port", 9200);
    conf.put("zeit.recommend.rabbitmq.exchange", "zr_spout");
    conf.put("zeit.recommend.rabbitmq.host", "217.13.68.236");
    conf.put("zeit.recommend.rabbitmq.key", "logstash");
    conf.put("zeit.recommend.rabbitmq.port", 5672);
    conf.put("zeit.recommend.rabbitmq.throughput", 0.1);
    conf.put("zeit.recommend.svd.base", 5000);
    conf.put("zeit.recommend.svd.rank", 250);
    conf.put("zeit.recommend.zonapi.host", "217.13.68.229");
    conf.put("zeit.recommend.zonapi.port", 8983);

    LocalCluster cluster = new LocalCluster();
    cluster.submitTopology("zeit-recommend", conf, builder.createTopology());

    Thread.sleep(1 * 1 * 60 * 1000);
    cluster.shutdown();
  }
}

@@ -1349,75 +1349,8 @@
 le.%0A
-    // conf.put(%22zeit.recommend.elasticsearch.host%22, %22localhost%22);%0A
     
@@ -1964,16 +1964,61 @@
 , 8983);
+%0A    conf.put(%22zeit.recommend.runtime%22, 300);
 %0A%0A    Lo
@@ -2145,31 +2145,133 @@
     
-Thread.sleep(1 * 1 * 60
+Object runtime = conf.get(%22zeit.recommend.runtime%22);%0A    if ((Integer)(runtime)%3E 0) %7B%0A        Thread.sleep((Integer)(runtime)
  * 1
@@ -2272,24 +2272,28 @@
 e) * 1000);%0A
+    
     cluster.
@@ -2304,14 +2304,20 @@
 down();%0A
+    %7D%0A
   %7D%0A%7D%0A

Add runtime as config option