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", "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.4);
    conf.put("zeit.recommend.svd.base", 1000);
    conf.put("zeit.recommend.svd.rank", 120);
    conf.put("zeit.recommend.zonapi.host", "217.13.68.229");
    conf.put("zeit.recommend.zonapi.port", 8983);
    conf.put("zeit.recommend.runtime", 860);

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

    Object runtime = conf.get("zeit.recommend.runtime");
    if ((Integer)(runtime) > 0) {
        Thread.sleep((Integer)(runtime) * 1000);
        cluster.shutdown();
    }
  }
}

@@ -447,27 +447,24 @@
 lder();%0A%0A   
- //
  builder.set
@@ -469,27 +469,24 @@
 etSpout(%0A   
- //
      %22zonapi
@@ -487,27 +487,24 @@
 zonapi%22,%0A   
- //
      new Pyt
@@ -533,27 +533,24 @@
 %22path%22),%0A   
- //
      1);%0A%0A  
@@ -550,19 +550,16 @@
 1);%0A%0A   
- //
   builde
@@ -568,27 +568,24 @@
 setBolt(%0A   
- //
      %22item%22,
@@ -584,27 +584,24 @@
  %22item%22,%0A   
- //
      new Pyt
@@ -627,27 +627,24 @@
 %22path%22),%0A   
- //
      1).shuf
@@ -916,35 +916,32 @@
 rabbitmq%22);%0A%0A   
- //
  builder.setBolt
@@ -946,23 +946,20 @@
 lt(%0A    
-//
     
- 
 %22recomme
@@ -967,27 +967,24 @@
 dation%22,%0A   
- //
      new Pyt
@@ -1049,27 +1049,24 @@
 tions%22),%0A   
- //
      1).shuf
@@ -1090,19 +1090,16 @@
 %22);%0A%0A   
- //
  builder
@@ -1107,27 +1107,24 @@
 setBolt(%0A   
- //
      %22outlet
@@ -1125,27 +1125,24 @@
 outlet%22,%0A   
- //
      new Pyt
@@ -1166,19 +1166,16 @@
 y%22),%0A   
- //
      1).
@@ -1761,9 +1761,9 @@
 , 0.
-4
+1
 );%0A 
@@ -1801,18 +1801,18 @@
 .base%22, 
-10
+32
 00);%0A   
@@ -1852,10 +1852,10 @@
 k%22, 
-1
 2
+0
 0);%0A
@@ -2008,10 +2008,10 @@
 e%22, 
+1
 8
-6
 0);%0A

Tweak settings, re-enable all bolts