Remove samples (#22288)

Add code, that was used for documentation to the appropriate projects
or akka-docs.

akka-docs/rst/java/code/docs/cluster/ClusterDocTest.java

@@ -21,7 +21,7 @@ public class ClusterDocTest extends AbstractJavaTest {
   @BeforeClass
   public static void setup() {
     system = ActorSystem.create("ClusterDocTest", 
-        ConfigFactory.parseString(ClusterDocSpec.config()));
+        ConfigFactory.parseString(scala.docs.cluster.ClusterDocSpec.config()));
   }
 
   @AfterClass

akka-docs/rst/java/code/docs/cluster/FactorialBackend.java

@@ -0,0 +1,47 @@
+package docs.cluster;
+
+import java.math.BigInteger;
+import java.util.concurrent.Callable;
+import scala.concurrent.Future;
+import akka.actor.UntypedActor;
+import akka.dispatch.Mapper;
+import static akka.dispatch.Futures.future;
+import static akka.pattern.Patterns.pipe;
+
+//#backend
+public class FactorialBackend extends UntypedActor {
+
+  @Override
+  public void onReceive(Object message) {
+    if (message instanceof Integer) {
+      final Integer n = (Integer) message;
+      Future<BigInteger> f = future(new Callable<BigInteger>() {
+        public BigInteger call() {
+          return factorial(n);
+        }
+      }, getContext().dispatcher());
+
+      Future<FactorialResult> result = f.map(
+          new Mapper<BigInteger, FactorialResult>() {
+            public FactorialResult apply(BigInteger factorial) {
+              return new FactorialResult(n, factorial);
+            }
+          }, getContext().dispatcher());
+
+      pipe(result, getContext().dispatcher()).to(getSender());
+
+    } else {
+      unhandled(message);
+    }
+  }
+
+  BigInteger factorial(int n) {
+    BigInteger acc = BigInteger.ONE;
+    for (int i = 1; i <= n; ++i) {
+      acc = acc.multiply(BigInteger.valueOf(i));
+    }
+    return acc;
+  }
+}
+//#backend
+

akka-docs/rst/java/code/docs/cluster/FactorialFrontend.java

@@ -0,0 +1,107 @@
+package docs.cluster;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import akka.actor.Props;
+import akka.cluster.metrics.AdaptiveLoadBalancingGroup;
+import akka.cluster.metrics.AdaptiveLoadBalancingPool;
+import akka.cluster.metrics.HeapMetricsSelector;
+import akka.cluster.metrics.SystemLoadAverageMetricsSelector;
+import akka.cluster.routing.ClusterRouterGroup;
+import akka.cluster.routing.ClusterRouterGroupSettings;
+import akka.cluster.routing.ClusterRouterPool;
+import akka.cluster.routing.ClusterRouterPoolSettings;
+import akka.routing.ConsistentHashingGroup;
+import akka.routing.ConsistentHashingPool;
+import scala.concurrent.duration.Duration;
+import akka.actor.ActorRef;
+import akka.actor.ReceiveTimeout;
+import akka.actor.UntypedActor;
+import akka.event.Logging;
+import akka.event.LoggingAdapter;
+import akka.routing.FromConfig;
+
+//#frontend
+public class FactorialFrontend extends UntypedActor {
+  final int upToN;
+  final boolean repeat;
+
+  LoggingAdapter log = Logging.getLogger(getContext().system(), this);
+
+  ActorRef backend = getContext().actorOf(FromConfig.getInstance().props(),
+      "factorialBackendRouter");
+
+  public FactorialFrontend(int upToN, boolean repeat) {
+    this.upToN = upToN;
+    this.repeat = repeat;
+  }
+
+  @Override
+  public void preStart() {
+    sendJobs();
+    getContext().setReceiveTimeout(Duration.create(10, TimeUnit.SECONDS));
+  }
+
+  @Override
+  public void onReceive(Object message) {
+    if (message instanceof FactorialResult) {
+      FactorialResult result = (FactorialResult) message;
+      if (result.n == upToN) {
+        log.debug("{}! = {}", result.n, result.factorial);
+        if (repeat)
+          sendJobs();
+        else
+          getContext().stop(getSelf());
+      }
+
+    } else if (message instanceof ReceiveTimeout) {
+      log.info("Timeout");
+      sendJobs();
+
+    } else {
+      unhandled(message);
+    }
+  }
+
+  void sendJobs() {
+    log.info("Starting batch of factorials up to [{}]", upToN);
+    for (int n = 1; n <= upToN; n++) {
+      backend.tell(n, getSelf());
+    }
+  }
+
+}
+//#frontend
+
+//not used, only for documentation
+abstract class FactorialFrontend2 extends UntypedActor {
+  //#router-lookup-in-code
+  int totalInstances = 100;
+  Iterable<String> routeesPaths = Arrays.asList("/user/factorialBackend", "");
+  boolean allowLocalRoutees = true;
+  String useRole = "backend";
+  ActorRef backend = getContext().actorOf(
+    new ClusterRouterGroup(new AdaptiveLoadBalancingGroup(
+      HeapMetricsSelector.getInstance(), Collections.<String> emptyList()),
+        new ClusterRouterGroupSettings(totalInstances, routeesPaths,
+          allowLocalRoutees, useRole)).props(), "factorialBackendRouter2");
+  //#router-lookup-in-code
+}
+
+//not used, only for documentation
+abstract class FactorialFrontend3 extends UntypedActor {
+  //#router-deploy-in-code
+  int totalInstances = 100;
+  int maxInstancesPerNode = 3;
+  boolean allowLocalRoutees = false;
+  String useRole = "backend";
+  ActorRef backend = getContext().actorOf(
+    new ClusterRouterPool(new AdaptiveLoadBalancingPool(
+      SystemLoadAverageMetricsSelector.getInstance(), 0),
+        new ClusterRouterPoolSettings(totalInstances, maxInstancesPerNode,
+          allowLocalRoutees, useRole)).props(Props
+            .create(FactorialBackend.class)), "factorialBackendRouter3");
+  //#router-deploy-in-code
+}

akka-docs/rst/java/code/docs/cluster/FactorialFrontendMain.java

@@ -0,0 +1,72 @@
+package docs.cluster;
+
+
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.TimeUnit;
+import scala.concurrent.Await;
+import scala.concurrent.duration.Duration;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import akka.cluster.Cluster;
+
+public class FactorialFrontendMain {
+
+  public static void main(String[] args) {
+    final int upToN = 200;
+
+    final Config config = ConfigFactory.parseString(
+        "akka.cluster.roles = [frontend]").withFallback(
+        ConfigFactory.load("factorial"));
+
+    final ActorSystem system = ActorSystem.create("ClusterSystem", config);
+    system.log().info(
+            "Factorials will start when 2 backend members in the cluster.");
+    //#registerOnUp
+    Cluster.get(system).registerOnMemberUp(new Runnable() {
+      @Override
+      public void run() {
+        system.actorOf(Props.create(FactorialFrontend.class, upToN, true),
+            "factorialFrontend");
+      }
+    });
+    //#registerOnUp
+
+    //#registerOnRemoved
+    Cluster.get(system).registerOnMemberRemoved(new Runnable() {
+      @Override
+      public void run() {
+        // exit JVM when ActorSystem has been terminated
+        final Runnable exit = new Runnable() {
+          @Override public void run() {
+            System.exit(0);
+          }
+        };
+        system.registerOnTermination(exit);
+
+        // shut down ActorSystem
+        system.terminate();
+
+        // In case ActorSystem shutdown takes longer than 10 seconds,
+        // exit the JVM forcefully anyway.
+        // We must spawn a separate thread to not block current thread,
+        // since that would have blocked the shutdown of the ActorSystem.
+        new Thread() {
+          @Override public void run(){
+            try {
+              Await.ready(system.whenTerminated(), Duration.create(10, TimeUnit.SECONDS));
+            } catch (Exception e) {
+              System.exit(-1);
+            }
+
+          }
+        }.start();
+      }
+    });
+    //#registerOnRemoved
+
+  }
+
+}

akka-docs/rst/java/code/docs/cluster/FactorialResult.java

@@ -0,0 +1,15 @@
+package docs.cluster;
+
+import java.math.BigInteger;
+import java.io.Serializable;
+
+public class FactorialResult implements Serializable {
+	private static final long serialVersionUID = 1L;
+	public final int n;
+	public final BigInteger factorial;
+
+	FactorialResult(int n, BigInteger factorial) {
+		this.n = n;
+		this.factorial = factorial;
+	}
+}

akka-docs/rst/java/code/docs/cluster/MetricsListener.java

@@ -0,0 +1,71 @@
+package docs.cluster;
+
+//#metrics-listener
+import akka.actor.UntypedActor;
+import akka.cluster.Cluster;
+import akka.cluster.ClusterEvent.CurrentClusterState;
+import akka.cluster.metrics.ClusterMetricsChanged;
+import akka.cluster.metrics.NodeMetrics;
+import akka.cluster.metrics.StandardMetrics;
+import akka.cluster.metrics.StandardMetrics.HeapMemory;
+import akka.cluster.metrics.StandardMetrics.Cpu;
+import akka.cluster.metrics.ClusterMetricsExtension;
+import akka.event.Logging;
+import akka.event.LoggingAdapter;
+
+public class MetricsListener extends UntypedActor {
+  LoggingAdapter log = Logging.getLogger(getContext().system(), this);
+
+  Cluster cluster = Cluster.get(getContext().system());
+  
+  ClusterMetricsExtension extension = ClusterMetricsExtension.get(getContext().system());
+
+  
+  // Subscribe unto ClusterMetricsEvent events.
+  @Override
+  public void preStart() {
+	  extension.subscribe(getSelf());
+  }
+
+  // Unsubscribe from ClusterMetricsEvent events.
+  @Override
+  public void postStop() {
+	  extension.unsubscribe(getSelf());
+  }
+
+
+  @Override
+  public void onReceive(Object message) {
+    if (message instanceof ClusterMetricsChanged) {
+    	ClusterMetricsChanged clusterMetrics = (ClusterMetricsChanged) message;
+      for (NodeMetrics nodeMetrics : clusterMetrics.getNodeMetrics()) {
+        if (nodeMetrics.address().equals(cluster.selfAddress())) {
+          logHeap(nodeMetrics);
+          logCpu(nodeMetrics);
+        }
+      }
+
+    } else if (message instanceof CurrentClusterState) {
+      // Ignore.
+    } else {
+      unhandled(message);
+    }
+  }
+
+  void logHeap(NodeMetrics nodeMetrics) {
+    HeapMemory heap = StandardMetrics.extractHeapMemory(nodeMetrics);
+    if (heap != null) {
+      log.info("Used heap: {} MB", ((double) heap.used()) / 1024 / 1024);
+    }
+  }
+
+  void logCpu(NodeMetrics nodeMetrics) {
+    Cpu cpu = StandardMetrics.extractCpu(nodeMetrics);
+    if (cpu != null && cpu.systemLoadAverage().isDefined()) {
+      log.info("Load: {} ({} processors)", cpu.systemLoadAverage().get(), 
+        cpu.processors());
+    }
+  }
+
+}
+//#metrics-listener

akka-docs/rst/java/code/docs/cluster/SimpleClusterListener.java

@@ -0,0 +1,54 @@
+package docs.cluster;
+
+import akka.actor.UntypedActor;
+import akka.cluster.Cluster;
+import akka.cluster.ClusterEvent;
+import akka.cluster.ClusterEvent.MemberEvent;
+import akka.cluster.ClusterEvent.MemberUp;
+import akka.cluster.ClusterEvent.MemberRemoved;
+import akka.cluster.ClusterEvent.UnreachableMember;
+import akka.event.Logging;
+import akka.event.LoggingAdapter;
+
+public class SimpleClusterListener extends UntypedActor {
+  LoggingAdapter log = Logging.getLogger(getContext().system(), this);
+  Cluster cluster = Cluster.get(getContext().system());
+
+  //subscribe to cluster changes
+  @Override
+  public void preStart() {
+    //#subscribe
+    cluster.subscribe(getSelf(), ClusterEvent.initialStateAsEvents(), 
+        MemberEvent.class, UnreachableMember.class);
+    //#subscribe
+  }
+
+  //re-subscribe when restart
+  @Override
+  public void postStop() {
+    cluster.unsubscribe(getSelf());
+  }
+
+  @Override
+  public void onReceive(Object message) {
+    if (message instanceof MemberUp) {
+      MemberUp mUp = (MemberUp) message;
+      log.info("Member is Up: {}", mUp.member());
+
+    } else if (message instanceof UnreachableMember) {
+      UnreachableMember mUnreachable = (UnreachableMember) message;
+      log.info("Member detected as unreachable: {}", mUnreachable.member());
+
+    } else if (message instanceof MemberRemoved) {
+      MemberRemoved mRemoved = (MemberRemoved) message;
+      log.info("Member is Removed: {}", mRemoved.member());
+
+    } else if (message instanceof MemberEvent) {
+      // ignore
+
+    } else {
+      unhandled(message);
+    }
+
+  }
+}

akka-docs/rst/java/code/docs/cluster/SimpleClusterListener2.java

@@ -0,0 +1,57 @@
+package docs.cluster;
+
+import akka.actor.UntypedActor;
+import akka.cluster.Cluster;
+import akka.cluster.ClusterEvent.CurrentClusterState;
+import akka.cluster.ClusterEvent.MemberEvent;
+import akka.cluster.ClusterEvent.MemberUp;
+import akka.cluster.ClusterEvent.MemberRemoved;
+import akka.cluster.ClusterEvent.UnreachableMember;
+import akka.event.Logging;
+import akka.event.LoggingAdapter;
+
+public class SimpleClusterListener2 extends UntypedActor {
+  LoggingAdapter log = Logging.getLogger(getContext().system(), this);
+  Cluster cluster = Cluster.get(getContext().system());
+
+  //subscribe to cluster changes
+  @Override
+  public void preStart() {
+    //#subscribe
+    cluster.subscribe(getSelf(), MemberEvent.class, UnreachableMember.class);
+    //#subscribe
+  }
+
+  //re-subscribe when restart
+  @Override
+  public void postStop() {
+    cluster.unsubscribe(getSelf());
+  }
+
+  @Override
+  public void onReceive(Object message) {
+    if (message instanceof CurrentClusterState) {
+      CurrentClusterState state = (CurrentClusterState) message;
+      log.info("Current members: {}", state.members());
+
+    } else if (message instanceof MemberUp) {
+      MemberUp mUp = (MemberUp) message;
+      log.info("Member is Up: {}", mUp.member());
+
+    } else if (message instanceof UnreachableMember) {
+      UnreachableMember mUnreachable = (UnreachableMember) message;
+      log.info("Member detected as unreachable: {}", mUnreachable.member());
+
+    } else if (message instanceof MemberRemoved) {
+      MemberRemoved mRemoved = (MemberRemoved) message;
+      log.info("Member is Removed: {}", mRemoved.member());
+
+    } else if (message instanceof MemberEvent) {
+      // ignore
+
+    } else {
+      unhandled(message);
+    }
+
+  }
+}

akka-docs/rst/java/code/docs/cluster/StatsAggregator.java

@@ -0,0 +1,56 @@
+package docs.cluster;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import docs.cluster.StatsMessages.JobFailed;
+import docs.cluster.StatsMessages.StatsResult;
+import scala.concurrent.duration.Duration;
+import akka.actor.ActorRef;
+import akka.actor.ReceiveTimeout;
+import akka.actor.UntypedActor;
+
+//#aggregator
+public class StatsAggregator extends UntypedActor {
+
+  final int expectedResults;
+  final ActorRef replyTo;
+  final List<Integer> results = new ArrayList<Integer>();
+
+  public StatsAggregator(int expectedResults, ActorRef replyTo) {
+    this.expectedResults = expectedResults;
+    this.replyTo = replyTo;
+  }
+
+  @Override
+  public void preStart() {
+    getContext().setReceiveTimeout(Duration.create(3, TimeUnit.SECONDS));
+  }
+
+  @Override
+  public void onReceive(Object message) {
+    if (message instanceof Integer) {
+      Integer wordCount = (Integer) message;
+      results.add(wordCount);
+      if (results.size() == expectedResults) {
+        int sum = 0;
+        for (int c : results)
+          sum += c;
+        double meanWordLength = ((double) sum) / results.size();
+        replyTo.tell(new StatsResult(meanWordLength), getSelf());
+        getContext().stop(getSelf());
+      }
+
+    } else if (message == ReceiveTimeout.getInstance()) {
+      replyTo.tell(new JobFailed("Service unavailable, try again later"),
+          getSelf());
+      getContext().stop(getSelf());
+
+    } else {
+      unhandled(message);
+    }
+  }
+
+}
+//#aggregator

akka-docs/rst/java/code/docs/cluster/StatsMessages.java

@@ -0,0 +1,55 @@
+package docs.cluster;
+
+import java.io.Serializable;
+
+//#messages
+public interface StatsMessages {
+
+  public static class StatsJob implements Serializable {
+    private final String text;
+
+    public StatsJob(String text) {
+      this.text = text;
+    }
+
+    public String getText() {
+      return text;
+    }
+  }
+
+  public static class StatsResult implements Serializable {
+    private final double meanWordLength;
+
+    public StatsResult(double meanWordLength) {
+      this.meanWordLength = meanWordLength;
+    }
+
+    public double getMeanWordLength() {
+      return meanWordLength;
+    }
+
+    @Override
+    public String toString() {
+      return "meanWordLength: " + meanWordLength;
+    }
+  }
+
+  public static class JobFailed implements Serializable {
+    private final String reason;
+
+    public JobFailed(String reason) {
+      this.reason = reason;
+    }
+
+    public String getReason() {
+      return reason;
+    }
+
+    @Override
+    public String toString() {
+      return "JobFailed(" + reason + ")";
+    }
+  }
+
+}
+//#messages

akka-docs/rst/java/code/docs/cluster/StatsSampleClient.java

@@ -0,0 +1,111 @@
+package docs.cluster;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import docs.cluster.StatsMessages.JobFailed;
+import docs.cluster.StatsMessages.StatsJob;
+import docs.cluster.StatsMessages.StatsResult;
+import java.util.concurrent.ThreadLocalRandom;
+import scala.concurrent.duration.Duration;
+import scala.concurrent.duration.FiniteDuration;
+import akka.actor.ActorSelection;
+import akka.actor.Address;
+import akka.actor.Cancellable;
+import akka.actor.UntypedActor;
+import akka.cluster.Cluster;
+import akka.cluster.ClusterEvent.UnreachableMember;
+import akka.cluster.ClusterEvent.ReachableMember;
+import akka.cluster.ClusterEvent.CurrentClusterState;
+import akka.cluster.ClusterEvent.MemberEvent;
+import akka.cluster.ClusterEvent.MemberUp;
+import akka.cluster.ClusterEvent.ReachabilityEvent;
+import akka.cluster.Member;
+import akka.cluster.MemberStatus;
+
+public class StatsSampleClient extends UntypedActor {
+
+  final String servicePath;
+  final Cancellable tickTask;
+  final Set<Address> nodes = new HashSet<Address>();
+
+  Cluster cluster = Cluster.get(getContext().system());
+
+  public StatsSampleClient(String servicePath) {
+    this.servicePath = servicePath;
+    FiniteDuration interval = Duration.create(2, TimeUnit.SECONDS);
+    tickTask = getContext()
+        .system()
+        .scheduler()
+        .schedule(interval, interval, getSelf(), "tick",
+            getContext().dispatcher(), null);
+  }
+
+  //subscribe to cluster changes, MemberEvent
+  @Override
+  public void preStart() {
+    cluster.subscribe(getSelf(), MemberEvent.class, ReachabilityEvent.class);
+  }
+
+  //re-subscribe when restart
+  @Override
+  public void postStop() {
+    cluster.unsubscribe(getSelf());
+    tickTask.cancel();
+  }
+
+  @Override
+  public void onReceive(Object message) {
+    if (message.equals("tick") && !nodes.isEmpty()) {
+      // just pick any one
+      List<Address> nodesList = new ArrayList<Address>(nodes);
+      Address address = nodesList.get(ThreadLocalRandom.current().nextInt(
+          nodesList.size()));
+      ActorSelection service = getContext().actorSelection(address + servicePath);
+      service.tell(new StatsJob("this is the text that will be analyzed"),
+          getSelf());
+
+    } else if (message instanceof StatsResult) {
+      StatsResult result = (StatsResult) message;
+      System.out.println(result);
+
+    } else if (message instanceof JobFailed) {
+      JobFailed failed = (JobFailed) message;
+      System.out.println(failed);
+
+    } else if (message instanceof CurrentClusterState) {
+      CurrentClusterState state = (CurrentClusterState) message;
+      nodes.clear();
+      for (Member member : state.getMembers()) {
+        if (member.hasRole("compute") && member.status().equals(MemberStatus.up())) {
+          nodes.add(member.address());
+        }
+      }
+
+    } else if (message instanceof MemberUp) {
+      MemberUp mUp = (MemberUp) message;
+      if (mUp.member().hasRole("compute"))
+        nodes.add(mUp.member().address());
+
+    } else if (message instanceof MemberEvent) {
+      MemberEvent other = (MemberEvent) message;
+      nodes.remove(other.member().address());
+
+    } else if (message instanceof UnreachableMember) {
+      UnreachableMember unreachable = (UnreachableMember) message;
+      nodes.remove(unreachable.member().address());
+
+    } else if (message instanceof ReachableMember) {
+      ReachableMember reachable = (ReachableMember) message;
+      if (reachable.member().hasRole("compute"))
+        nodes.add(reachable.member().address());
+
+    } else {
+      unhandled(message);
+    }
+  }
+
+}

akka-docs/rst/java/code/docs/cluster/StatsSampleOneMasterClientMain.java

@@ -0,0 +1,19 @@
+package docs.cluster;
+
+import com.typesafe.config.ConfigFactory;
+
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+
+public class StatsSampleOneMasterClientMain {
+
+  public static void main(String[] args) {
+    // note that client is not a compute node, role not defined
+    ActorSystem system = ActorSystem.create("ClusterSystem",
+        ConfigFactory.load("stats2"));
+    system.actorOf(Props.create(StatsSampleClient.class, "/user/statsServiceProxy"),
+        "client");
+
+  }
+
+}

akka-docs/rst/java/code/docs/cluster/StatsSampleOneMasterMain.java

@@ -0,0 +1,53 @@
+package docs.cluster;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+import akka.actor.ActorSystem;
+import akka.actor.PoisonPill;
+import akka.actor.Props;
+import akka.cluster.singleton.ClusterSingletonManager;
+import akka.cluster.singleton.ClusterSingletonManagerSettings;
+import akka.cluster.singleton.ClusterSingletonProxy;
+import akka.cluster.singleton.ClusterSingletonProxySettings;
+
+public class StatsSampleOneMasterMain {
+
+  public static void main(String[] args) {
+    if (args.length == 0) {
+      startup(new String[] { "2551", "2552", "0" });
+      StatsSampleOneMasterClientMain.main(new String[0]);
+    } else {
+      startup(args);
+    }
+  }
+
+  public static void startup(String[] ports) {
+    for (String port : ports) {
+      // Override the configuration of the port
+      Config config = ConfigFactory
+          .parseString("akka.remote.netty.tcp.port=" + port)
+          .withFallback(
+              ConfigFactory.parseString("akka.cluster.roles = [compute]"))
+          .withFallback(ConfigFactory.load("stats2"));
+
+      ActorSystem system = ActorSystem.create("ClusterSystem", config);
+
+      //#create-singleton-manager
+      ClusterSingletonManagerSettings settings = ClusterSingletonManagerSettings.create(system)
+          .withRole("compute");
+      system.actorOf(ClusterSingletonManager.props(
+          Props.create(StatsService.class), PoisonPill.getInstance(), settings),
+          "statsService");
+      //#create-singleton-manager
+
+      //#singleton-proxy
+      ClusterSingletonProxySettings proxySettings =
+          ClusterSingletonProxySettings.create(system).withRole("compute");
+      system.actorOf(ClusterSingletonProxy.props("/user/statsService",
+          proxySettings), "statsServiceProxy");
+      //#singleton-proxy
+    }
+
+  }
+}

akka-docs/rst/java/code/docs/cluster/StatsService.java

@@ -0,0 +1,85 @@
+package docs.cluster;
+
+import akka.cluster.routing.ClusterRouterGroup;
+import akka.cluster.routing.ClusterRouterGroupSettings;
+import akka.cluster.routing.ClusterRouterPool;
+import akka.cluster.routing.ClusterRouterPoolSettings;
+import akka.routing.ConsistentHashingGroup;
+import akka.routing.ConsistentHashingPool;
+import docs.cluster.StatsMessages.StatsJob;
+import akka.actor.ActorRef;
+import akka.actor.Props;
+import akka.actor.UntypedActor;
+import akka.routing.ConsistentHashingRouter.ConsistentHashableEnvelope;
+import akka.routing.FromConfig;
+
+import java.util.Collections;
+
+//#service
+public class StatsService extends UntypedActor {
+
+  // This router is used both with lookup and deploy of routees. If you
+  // have a router with only lookup of routees you can use Props.empty()
+  // instead of Props.create(StatsWorker.class).
+  ActorRef workerRouter = getContext().actorOf(
+      FromConfig.getInstance().props(Props.create(StatsWorker.class)),
+      "workerRouter");
+
+  @Override
+  public void onReceive(Object message) {
+    if (message instanceof StatsJob) {
+      StatsJob job = (StatsJob) message;
+      if (job.getText().equals("")) {
+        unhandled(message);
+      } else {
+        final String[] words = job.getText().split(" ");
+        final ActorRef replyTo = getSender();
+
+        // create actor that collects replies from workers
+        ActorRef aggregator = getContext().actorOf(
+            Props.create(StatsAggregator.class, words.length, replyTo));
+
+        // send each word to a worker
+        for (String word : words) {
+          workerRouter.tell(new ConsistentHashableEnvelope(word, word),
+              aggregator);
+        }
+      }
+
+    } else {
+      unhandled(message);
+    }
+  }
+}
+//#service
+
+//not used, only for documentation
+abstract class StatsService2 extends UntypedActor {
+  //#router-lookup-in-code
+  int totalInstances = 100;
+  Iterable<String> routeesPaths = Collections
+    .singletonList("/user/statsWorker");
+  boolean allowLocalRoutees = true;
+  String useRole = "compute";
+  ActorRef workerRouter = getContext().actorOf(
+    new ClusterRouterGroup(new ConsistentHashingGroup(routeesPaths),
+      new ClusterRouterGroupSettings(totalInstances, routeesPaths,
+        allowLocalRoutees, useRole)).props(), "workerRouter2");
+  //#router-lookup-in-code
+}
+
+//not used, only for documentation
+abstract class StatsService3 extends UntypedActor {
+  //#router-deploy-in-code
+  int totalInstances = 100;
+  int maxInstancesPerNode = 3;
+  boolean allowLocalRoutees = false;
+  String useRole = "compute";
+  ActorRef workerRouter = getContext().actorOf(
+    new ClusterRouterPool(new ConsistentHashingPool(0),
+      new ClusterRouterPoolSettings(totalInstances, maxInstancesPerNode,
+        allowLocalRoutees, useRole)).props(Props
+          .create(StatsWorker.class)), "workerRouter3");
+  //#router-deploy-in-code
+}
+

akka-docs/rst/java/code/docs/cluster/StatsWorker.java

@@ -0,0 +1,30 @@
+package docs.cluster;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import akka.actor.UntypedActor;
+
+//#worker
+public class StatsWorker extends UntypedActor {
+
+  Map<String, Integer> cache = new HashMap<String, Integer>();
+
+  @Override
+  public void onReceive(Object message) {
+    if (message instanceof String) {
+      String word = (String) message;
+      Integer length = cache.get(word);
+      if (length == null) {
+        length = word.length();
+        cache.put(word, length);
+      }
+      getSender().tell(length, getSelf());
+
+    } else {
+      unhandled(message);
+    }
+  }
+
+}
+//#worker

akka-docs/rst/java/code/docs/cluster/TransformationBackend.java

@@ -0,0 +1,60 @@
+package docs.cluster;
+
+import static docs.cluster.TransformationMessages.BACKEND_REGISTRATION;
+import docs.cluster.TransformationMessages.TransformationJob;
+import docs.cluster.TransformationMessages.TransformationResult;
+import akka.actor.UntypedActor;
+import akka.cluster.Cluster;
+import akka.cluster.ClusterEvent.CurrentClusterState;
+import akka.cluster.ClusterEvent.MemberUp;
+import akka.cluster.Member;
+import akka.cluster.MemberStatus;
+
+//#backend
+public class TransformationBackend extends UntypedActor {
+
+  Cluster cluster = Cluster.get(getContext().system());
+
+  //subscribe to cluster changes, MemberUp
+  @Override
+  public void preStart() {
+    cluster.subscribe(getSelf(), MemberUp.class);
+  }
+
+  //re-subscribe when restart
+  @Override
+  public void postStop() {
+    cluster.unsubscribe(getSelf());
+  }
+
+  @Override
+  public void onReceive(Object message) {
+    if (message instanceof TransformationJob) {
+      TransformationJob job = (TransformationJob) message;
+      getSender().tell(new TransformationResult(job.getText().toUpperCase()),
+          getSelf());
+
+    } else if (message instanceof CurrentClusterState) {
+      CurrentClusterState state = (CurrentClusterState) message;
+      for (Member member : state.getMembers()) {
+        if (member.status().equals(MemberStatus.up())) {
+          register(member);
+        }
+      }
+
+    } else if (message instanceof MemberUp) {
+      MemberUp mUp = (MemberUp) message;
+      register(mUp.member());
+
+    } else {
+      unhandled(message);
+    }
+  }
+
+  void register(Member member) {
+    if (member.hasRole("frontend"))
+      getContext().actorSelection(member.address() + "/user/frontend").tell(
+          BACKEND_REGISTRATION, getSelf());
+  }
+}
+//#backend

akka-docs/rst/java/code/docs/cluster/TransformationFrontend.java

@@ -0,0 +1,48 @@
+package docs.cluster;
+
+import static docs.cluster.TransformationMessages.BACKEND_REGISTRATION;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import docs.cluster.TransformationMessages.JobFailed;
+import docs.cluster.TransformationMessages.TransformationJob;
+import akka.actor.ActorRef;
+import akka.actor.Terminated;
+import akka.actor.UntypedActor;
+
+//#frontend
+public class TransformationFrontend extends UntypedActor {
+
+  List<ActorRef> backends = new ArrayList<ActorRef>();
+  int jobCounter = 0;
+
+  @Override
+  public void onReceive(Object message) {
+    if ((message instanceof TransformationJob) && backends.isEmpty()) {
+      TransformationJob job = (TransformationJob) message;
+      getSender().tell(
+          new JobFailed("Service unavailable, try again later", job),
+          getSender());
+
+    } else if (message instanceof TransformationJob) {
+      TransformationJob job = (TransformationJob) message;
+      jobCounter++;
+      backends.get(jobCounter % backends.size())
+          .forward(job, getContext());
+
+    } else if (message.equals(BACKEND_REGISTRATION)) {
+      getContext().watch(getSender());
+      backends.add(getSender());
+
+    } else if (message instanceof Terminated) {
+      Terminated terminated = (Terminated) message;
+      backends.remove(terminated.getActor());
+
+    } else {
+      unhandled(message);
+    }
+  }
+
+}
+//#frontend

akka-docs/rst/java/code/docs/cluster/TransformationMessages.java

@@ -0,0 +1,63 @@
+package docs.cluster;
+
+import java.io.Serializable;
+
+//#messages
+public interface TransformationMessages {
+
+  public static class TransformationJob implements Serializable {
+    private final String text;
+
+    public TransformationJob(String text) {
+      this.text = text;
+    }
+
+    public String getText() {
+      return text;
+    }
+  }
+
+  public static class TransformationResult implements Serializable {
+    private final String text;
+
+    public TransformationResult(String text) {
+      this.text = text;
+    }
+
+    public String getText() {
+      return text;
+    }
+
+    @Override
+    public String toString() {
+      return "TransformationResult(" + text + ")";
+    }
+  }
+
+  public static class JobFailed implements Serializable {
+    private final String reason;
+    private final TransformationJob job;
+
+    public JobFailed(String reason, TransformationJob job) {
+      this.reason = reason;
+      this.job = job;
+    }
+
+    public String getReason() {
+      return reason;
+    }
+
+    public TransformationJob getJob() {
+      return job;
+    }
+
+    @Override
+    public String toString() {
+      return "JobFailed(" + reason + ")";
+    }
+  }
+
+  public static final String BACKEND_REGISTRATION = "BackendRegistration";
+
+}
+//#messages