/*
* Copyright 2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package wordcount;
import java.util.HashMap;
import java.util.Map;

import net.kuujo.vertigo.Cluster;
import net.kuujo.vertigo.LocalCluster;
import net.kuujo.vertigo.Networks;
import net.kuujo.vertigo.component.feeder.BasicFeeder;
import net.kuujo.vertigo.component.worker.Worker;
import net.kuujo.vertigo.context.NetworkContext;
import net.kuujo.vertigo.definition.ComponentDefinition;
import net.kuujo.vertigo.definition.NetworkDefinition;
import net.kuujo.vertigo.grouping.FieldsGrouping;
import net.kuujo.vertigo.java.VertigoVerticle;
import net.kuujo.vertigo.messaging.JsonMessage;

import org.vertx.java.core.AsyncResult;
import org.vertx.java.core.Handler;
import org.vertx.java.core.json.JsonObject;
import org.vertx.java.platform.Verticle;

/**
 * A word count network example.
 *
 * This example demonstrates the use of groupings - in particular the FieldsGrouping -
 * to control the dispersion of messages between multiple verticle instances.
 *
 * @author Jordan Halterman
 */
public class WordCountNetwork extends Verticle {

  public static class WordCountWorker extends VertigoVerticle {

    private Worker worker;

    private Map<String, Integer> counts = new HashMap<String, Integer>();

    @Override
    public void start() {
      worker = vertigo.createWorker();
      worker.messageHandler(new Handler<JsonMessage>() {
        @Override
        public void handle(JsonMessage message) {
          String word = message.body().getString("word");
          Integer count = counts.get(word);
          if (count == null) {
            count = 0;
          }
          count++;
          worker.emit(new JsonObject().putString("word", word).putNumber("count", count));
          worker.ack(message);
        }
      }).start();
    }
  }

  public static class WordFeeder extends VertigoVerticle {

    @Override
    public void start() {
      vertigo.createBasicFeeder().start(new Handler<AsyncResult<BasicFeeder>>() {
        @Override
        public void handle(AsyncResult<BasicFeeder> result) {
          if (result.failed()) {
            container.logger().error(result.cause());
          }
          else {
            BasicFeeder feeder = result.result();
            feeder.feed(new JsonObject().putString("word", "foo"));
            feeder.feed(new JsonObject().putString("word", "bar"));
            feeder.feed(new JsonObject().putString("word", "foobar"));
            feeder.feed(new JsonObject().putString("word", "barbaz"));
            feeder.feed(new JsonObject().putString("word", "bar"));
            feeder.feed(new JsonObject().putString("word", "foobar"));
            feeder.feed(new JsonObject().putString("word", "barbaz"));
            feeder.feed(new JsonObject().putString("word", "bar"));
            feeder.feed(new JsonObject().putString("word", "foobar"));
            feeder.feed(new JsonObject().putString("word", "foobar"));
            feeder.feed(new JsonObject().putString("word", "foo"));
            feeder.feed(new JsonObject().putString("word", "bar"));
            feeder.feed(new JsonObject().putString("word", "bar"));
            feeder.feed(new JsonObject().putString("word", "bar"));
            feeder.feed(new JsonObject().putString("word", "bar"));
          }
        }
      });
    }

  }

  @Override
  public void start() {
    NetworkDefinition network = Networks.createNetwork("word_count");
    ComponentDefinition feeder = network.fromVerticle("word_feeder", WordFeeder.class.getName());

    // Group the word counter worker by the "word" field. This ensures that the
    // same word field always go to the same verticle instance.
    feeder.toVerticle("word_counter", WordCountWorker.class.getName(), 4).groupBy(new FieldsGrouping("word"));

    final Cluster cluster = new LocalCluster(vertx, container);
    cluster.deploy(network, new Handler<AsyncResult<NetworkContext>>() {
      @Override
      public void handle(AsyncResult<NetworkContext> result) {
        if (result.failed()) {
          container.logger().error(result.cause());
        }
        else {
          final NetworkContext context = result.result();
          vertx.setTimer(5000, new Handler<Long>() {
            @Override
            public void handle(Long timerID) {
              cluster.shutdown(context);
            }
          });
        }
      }
    });
  }

}

@@ -609,16 +609,17 @@
 dcount;%0A
+%0A
 import j
@@ -688,54 +688,23 @@
 igo.
-Cluster;%0Aimport net.kuujo.vertigo.LocalCluster
+VertigoVerticle
 ;%0Aim
@@ -726,24 +726,31 @@
 vertigo.
+network.
 Network
-s
 ;%0Aimport
@@ -772,36 +772,25 @@
 igo.
-component.feeder.BasicFeeder
+network.Component
 ;%0Aim
@@ -818,29 +818,28 @@
 o.co
-mponent.worker.Worker
+ntext.NetworkContext
 ;%0Aim
@@ -861,38 +861,39 @@
 vertigo.
-context.NetworkContext
+grouping.FieldsGrouping
 ;%0Aimport
@@ -915,93 +915,86 @@
 igo.
-definition.ComponentDefinition;%0Aimport net.kuujo.vertigo.definition.NetworkDefinition
+component.feeder.BasicFeeder;%0Aimport net.kuujo.vertigo.component.worker.Worker
 ;%0Aim
@@ -1016,39 +1016,31 @@
 vertigo.
-grouping.FieldsGrouping
+cluster.Cluster
 ;%0Aimport
@@ -1062,28 +1062,28 @@
 igo.
-java.VertigoVerticle
+cluster.LocalCluster
 ;%0Aim
@@ -3858,26 +3858,16 @@
  Network
-Definition
  network
@@ -3873,23 +3873,12 @@
 k = 
-Networks.create
+new 
 Netw
@@ -3913,18 +3913,8 @@
 nent
-Definition
  fee
@@ -3931,20 +3931,20 @@
 ork.
-fromVerticle
+addComponent
 (%22wo
@@ -3988,174 +3988,52 @@
 ));%0A
-%0A
     
-// Group the word counter worker by the %22word%22 field. This ensures that the%0A    // same word field always go to the same verticle instance.%0A    feeder.toVerticle
+Component counter = network.addComponent
 (%22wo
@@ -4080,16 +4080,53 @@
 me(), 4)
+;%0A    counter.addInput(%22word_feeder%22)
 .groupBy

Update word counter example.