+act #3911 Adding Java Lambda compatibility for Supervisor Strategy

Conflicts:
	akka-actor/src/main/scala/akka/util/Timeout.scala

akka-samples/akka-docs-java-lambda/src/test/java/docs/actor/FaultHandlingTest.java

@@ -0,0 +1,204 @@
+/**
+ * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
+ */
+package docs.actor;
+
+//#testkit
+import akka.actor.*;
+
+import static akka.actor.SupervisorStrategy.resume;
+import static akka.actor.SupervisorStrategy.restart;
+import static akka.actor.SupervisorStrategy.stop;
+import static akka.actor.SupervisorStrategy.escalate;
+import akka.japi.pf.DeciderBuilder;
+import akka.japi.pf.ReceiveBuilder;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import scala.PartialFunction;
+import scala.concurrent.Await;
+import static akka.pattern.Patterns.ask;
+import scala.concurrent.duration.Duration;
+import akka.testkit.TestProbe;
+
+//#testkit
+import akka.testkit.ErrorFilter;
+import akka.testkit.EventFilter;
+import akka.testkit.TestEvent;
+import akka.testkit.JavaTestKit;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static akka.japi.Util.immutableSeq;
+import scala.Option;
+
+import org.junit.Test;
+import org.junit.BeforeClass;
+import org.junit.AfterClass;
+import scala.runtime.BoxedUnit;
+
+//#testkit
+public class FaultHandlingTest {
+//#testkit
+
+  public static Config config = ConfigFactory.parseString(
+    "akka {\n" +
+    "  loggers = [\"akka.testkit.TestEventListener\"]\n" +
+    "  loglevel = \"WARNING\"\n" +
+    "  stdout-loglevel = \"WARNING\"\n" +
+    "}\n");
+
+  static
+  //#supervisor
+  public class Supervisor extends AbstractActor {
+
+    //#strategy
+    private static SupervisorStrategy strategy =
+      new OneForOneStrategy(10, Duration.create("1 minute"), DeciderBuilder.
+        match(ArithmeticException.class, e -> resume()).
+        match(NullPointerException.class, e -> restart()).
+        match(IllegalArgumentException.class, e -> stop()).
+        matchAny(o -> escalate()).build());
+
+    @Override
+    public SupervisorStrategy supervisorStrategy() {
+      return strategy;
+    }
+
+    //#strategy
+
+    @Override
+    public PartialFunction<Object, BoxedUnit> receive() {
+      return ReceiveBuilder.
+        match(Props.class, props -> {
+          sender().tell(context().actorOf(props), self());
+        }).build();
+    }
+  }
+
+  //#supervisor
+
+  static
+  //#supervisor2
+  public class Supervisor2 extends AbstractActor {
+
+    //#strategy2
+    private static SupervisorStrategy strategy =
+      new OneForOneStrategy(10, Duration.create("1 minute"), DeciderBuilder.
+        match(ArithmeticException.class, e -> resume()).
+        match(NullPointerException.class, e -> restart()).
+        match(IllegalArgumentException.class, e -> stop()).
+        matchAny(o -> escalate()).build());
+
+    @Override
+    public SupervisorStrategy supervisorStrategy() {
+      return strategy;
+    }
+
+    //#strategy2
+
+    @Override
+    public PartialFunction<Object, BoxedUnit> receive() {
+      return ReceiveBuilder.
+        match(Props.class, props -> {
+          sender().tell(context().actorOf(props), self());
+        }).build();
+    }
+
+    @Override
+    public void preRestart(Throwable cause, Option<Object> msg) {
+      // do not kill all children, which is the default here
+    }
+  }
+
+  //#supervisor2
+
+  static
+  //#child
+  public class Child extends AbstractActor {
+    int state = 0;
+
+    @Override
+    public PartialFunction<Object, BoxedUnit> receive() {
+      return ReceiveBuilder.
+        match(Exception.class, exception -> { throw exception; }).
+        match(Integer.class, i -> state = i).
+        matchEquals("get", s -> sender().tell(state, self())).build();
+    }
+  }
+
+  //#child
+
+  //#testkit
+  static ActorSystem system;
+  Duration timeout = Duration.create(5, SECONDS);
+
+  @BeforeClass
+  public static void start() {
+    system = ActorSystem.create("FaultHandlingTest", config);
+  }
+
+  @AfterClass
+  public static void cleanup() {
+    JavaTestKit.shutdownActorSystem(system);
+    system = null;
+  }
+
+  @Test
+  public void mustEmploySupervisorStrategy() throws Exception {
+    // code here
+    //#testkit
+    EventFilter ex1 = new ErrorFilter(ArithmeticException.class);
+    EventFilter ex2 = new ErrorFilter(NullPointerException.class);
+    EventFilter ex3 = new ErrorFilter(IllegalArgumentException.class);
+    EventFilter ex4 = new ErrorFilter(Exception.class);
+    EventFilter[] ignoreExceptions = { ex1, ex2, ex3, ex4 };
+    system.eventStream().publish(new TestEvent.Mute(immutableSeq(ignoreExceptions)));
+
+    //#create
+    Props superprops = Props.create(Supervisor.class);
+    ActorRef supervisor = system.actorOf(superprops, "supervisor");
+    ActorRef child = (ActorRef) Await.result(ask(supervisor,
+      Props.create(Child.class), 5000), timeout);
+    //#create
+
+    //#resume
+    child.tell(42, ActorRef.noSender());
+    assert Await.result(ask(child, "get", 5000), timeout).equals(42);
+    child.tell(new ArithmeticException(), ActorRef.noSender());
+    assert Await.result(ask(child, "get", 5000), timeout).equals(42);
+    //#resume
+
+    //#restart
+    child.tell(new NullPointerException(), ActorRef.noSender());
+    assert Await.result(ask(child, "get", 5000), timeout).equals(0);
+    //#restart
+
+    //#stop
+    final TestProbe probe = new TestProbe(system);
+    probe.watch(child);
+    child.tell(new IllegalArgumentException(), ActorRef.noSender());
+    probe.expectMsgClass(Terminated.class);
+    //#stop
+
+    //#escalate-kill
+    child = (ActorRef) Await.result(ask(supervisor,
+      Props.create(Child.class), 5000), timeout);
+    probe.watch(child);
+    assert Await.result(ask(child, "get", 5000), timeout).equals(0);
+    child.tell(new Exception(), ActorRef.noSender());
+    probe.expectMsgClass(Terminated.class);
+    //#escalate-kill
+
+    //#escalate-restart
+    superprops = Props.create(Supervisor2.class);
+    supervisor = system.actorOf(superprops);
+    child = (ActorRef) Await.result(ask(supervisor,
+      Props.create(Child.class), 5000), timeout);
+    child.tell(23, ActorRef.noSender());
+    assert Await.result(ask(child, "get", 5000), timeout).equals(23);
+    child.tell(new Exception(), ActorRef.noSender());
+    assert Await.result(ask(child, "get", 5000), timeout).equals(0);
+    //#escalate-restart
+    //#testkit
+  }
+
+}
+//#testkit

akka-samples/akka-docs-java-lambda/src/test/java/docs/actor/fsm/BuncherTest.java

@@ -36,8 +36,7 @@ public class BuncherTest {
   }
 
   @Test
-  public void testBuncherActorBatchesCorrectly()
-  {
+  public void testBuncherActorBatchesCorrectly() {
     new JavaTestKit(system) {{
       final ActorRef buncher =
         system.actorOf(Props.create(Buncher.class));
@@ -64,8 +63,7 @@ public class BuncherTest {
   }
 
   @Test
-  public void testBuncherActorDoesntBatchUninitialized()
-  {
+  public void testBuncherActorDoesntBatchUninitialized() {
     new JavaTestKit(system) {{
       final ActorRef buncher =
         system.actorOf(Props.create(Buncher.class));

akka-samples/akka-docs-java-lambda/src/test/java/docs/actor/japi/FaultHandlingDocSample.java

@@ -0,0 +1,472 @@
+/**
+ * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
+ */
+package docs.actor.japi;
+
+//#all
+//#imports
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import akka.actor.*;
+import akka.dispatch.Mapper;
+import akka.event.LoggingReceive;
+import akka.japi.pf.DeciderBuilder;
+import akka.japi.pf.ReceiveBuilder;
+import akka.util.Timeout;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import scala.concurrent.duration.Duration;
+import scala.PartialFunction;
+import scala.runtime.BoxedUnit;
+
+import static akka.japi.Util.classTag;
+import static akka.actor.SupervisorStrategy.resume;
+import static akka.actor.SupervisorStrategy.restart;
+import static akka.actor.SupervisorStrategy.stop;
+import static akka.actor.SupervisorStrategy.escalate;
+
+import static akka.pattern.Patterns.ask;
+import static akka.pattern.Patterns.pipe;
+
+import static docs.actor.japi.FaultHandlingDocSample.WorkerApi.*;
+import static docs.actor.japi.FaultHandlingDocSample.CounterServiceApi.*;
+import static docs.actor.japi.FaultHandlingDocSample.CounterApi.*;
+import static docs.actor.japi.FaultHandlingDocSample.StorageApi.*;
+
+//#imports
+
+public class FaultHandlingDocSample {
+
+  /**
+   * Runs the sample
+   */
+  public static void main(String[] args) {
+    Config config = ConfigFactory.parseString(
+      "akka.loglevel = \"DEBUG\"\n" +
+      "akka.actor.debug {\n" +
+      "  receive = on\n" +
+      "  lifecycle = on\n" +
+      "}\n");
+
+    ActorSystem system = ActorSystem.create("FaultToleranceSample", config);
+    ActorRef worker = system.actorOf(Props.create(Worker.class), "worker");
+    ActorRef listener = system.actorOf(Props.create(Listener.class), "listener");
+    // start the work and listen on progress
+    // note that the listener is used as sender of the tell,
+    // i.e. it will receive replies from the worker
+    worker.tell(Start, listener);
+  }
+
+  /**
+   * Listens on progress from the worker and shuts down the system when enough
+   * work has been done.
+   */
+  public static class Listener extends AbstractLoggingActor {
+
+    @Override
+    public void preStart() {
+      // If we don't get any progress within 15 seconds then the service
+      // is unavailable
+      context().setReceiveTimeout(Duration.create("15 seconds"));
+    }
+
+    @Override
+    public PartialFunction<Object, BoxedUnit> receive() {
+      return LoggingReceive.create(ReceiveBuilder.
+        match(Progress.class, progress -> {
+          log().info("Current progress: {} %", progress.percent);
+          if (progress.percent >= 100.0) {
+            log().info("That's all, shutting down");
+            context().system().shutdown();
+          }
+        }).
+        matchEquals(ReceiveTimeout.getInstance(), x -> {
+          // No progress within 15 seconds, ServiceUnavailable
+          log().error("Shutting down due to unavailable service");
+          context().system().shutdown();
+        }).build(), context());
+    }
+  }
+
+  //#messages
+  public interface WorkerApi {
+    public static final Object Start = "Start";
+    public static final Object Do = "Do";
+
+    public static class Progress {
+      public final double percent;
+
+      public Progress(double percent) {
+        this.percent = percent;
+      }
+
+      public String toString() {
+        return String.format("%s(%s)", getClass().getSimpleName(), percent);
+      }
+    }
+  }
+
+  //#messages
+
+  /**
+   * Worker performs some work when it receives the Start message. It will
+   * continuously notify the sender of the Start message of current Progress.
+   * The Worker supervise the CounterService.
+   */
+  public static class Worker extends AbstractLoggingActor {
+    final Timeout askTimeout = new Timeout(Duration.create(5, "seconds"));
+
+    // The sender of the initial Start message will continuously be notified
+    // about progress
+    ActorRef progressListener;
+    final ActorRef counterService = context().actorOf(
+      Props.create(CounterService.class), "counter");
+    final int totalCount = 51;
+
+    // Stop the CounterService child if it throws ServiceUnavailable
+    private static final SupervisorStrategy strategy =
+      new OneForOneStrategy(DeciderBuilder.
+        match(ServiceUnavailable.class, e -> stop()).
+        matchAny(o -> escalate()).build());
+
+    @Override
+    public SupervisorStrategy supervisorStrategy() {
+      return strategy;
+    }
+
+    @Override
+    public PartialFunction<Object, BoxedUnit> receive() {
+      return LoggingReceive.create(ReceiveBuilder.
+        matchEquals(Start, x -> progressListener == null, x -> {
+          progressListener = sender();
+          context().system().scheduler().schedule(
+            Duration.Zero(), Duration.create(1, "second"), self(), Do,
+            context().dispatcher(), null
+          );
+        }).
+        matchEquals(Do, x -> {
+          counterService.tell(new Increment(1), self());
+          counterService.tell(new Increment(1), self());
+          counterService.tell(new Increment(1), self());
+          // Send current progress to the initial sender
+          pipe(ask(counterService, GetCurrentCount, askTimeout)
+            .mapTo(classTag(CurrentCount.class))
+            .map(new Mapper<CurrentCount, Progress>() {
+              public Progress apply(CurrentCount c) {
+                return new Progress(100.0 * c.count / totalCount);
+              }
+            }, context().dispatcher()), context().dispatcher())
+            .to(progressListener);
+        }).build(), context());
+    }
+  }
+
+  //#messages
+  public interface CounterServiceApi {
+
+    public static final Object GetCurrentCount = "GetCurrentCount";
+
+    public static class CurrentCount {
+      public final String key;
+      public final long count;
+
+      public CurrentCount(String key, long count) {
+        this.key = key;
+        this.count = count;
+      }
+
+      public String toString() {
+        return String.format("%s(%s, %s)", getClass().getSimpleName(), key, count);
+      }
+    }
+
+    public static class Increment {
+      public final long n;
+
+      public Increment(long n) {
+        this.n = n;
+      }
+
+      public String toString() {
+        return String.format("%s(%s)", getClass().getSimpleName(), n);
+      }
+    }
+
+    public static class ServiceUnavailable extends RuntimeException {
+      private static final long serialVersionUID = 1L;
+      public ServiceUnavailable(String msg) {
+        super(msg);
+      }
+    }
+
+  }
+
+  //#messages
+
+  /**
+   * Adds the value received in Increment message to a persistent counter.
+   * Replies with CurrentCount when it is asked for CurrentCount. CounterService
+   * supervise Storage and Counter.
+   */
+  public static class CounterService extends AbstractLoggingActor {
+
+    // Reconnect message
+    static final Object Reconnect = "Reconnect";
+
+    private static class SenderMsgPair {
+      final ActorRef sender;
+      final Object msg;
+
+      SenderMsgPair(ActorRef sender, Object msg) {
+        this.msg = msg;
+        this.sender = sender;
+      }
+    }
+
+    final String key = self().path().name();
+    ActorRef storage;
+    ActorRef counter;
+    final List<SenderMsgPair> backlog = new ArrayList<>();
+    final int MAX_BACKLOG = 10000;
+
+    // Restart the storage child when StorageException is thrown.
+    // After 3 restarts within 5 seconds it will be stopped.
+    private static final SupervisorStrategy strategy =
+      new OneForOneStrategy(3, Duration.create("5 seconds"), DeciderBuilder.
+       match(StorageException.class, e -> restart()).
+       matchAny(o -> escalate()).build());
+
+    @Override
+    public SupervisorStrategy supervisorStrategy() {
+      return strategy;
+    }
+
+    @Override
+    public void preStart() {
+      initStorage();
+    }
+
+    /**
+     * The child storage is restarted in case of failure, but after 3 restarts,
+     * and still failing it will be stopped. Better to back-off than
+     * continuously failing. When it has been stopped we will schedule a
+     * Reconnect after a delay. Watch the child so we receive Terminated message
+     * when it has been terminated.
+     */
+    void initStorage() {
+      storage = context().watch(context().actorOf(
+        Props.create(Storage.class), "storage"));
+      // Tell the counter, if any, to use the new storage
+      if (counter != null)
+        counter.tell(new UseStorage(storage), self());
+      // We need the initial value to be able to operate
+      storage.tell(new Get(key), self());
+    }
+
+    @Override
+    public PartialFunction<Object, BoxedUnit> receive() {
+      return LoggingReceive.create(ReceiveBuilder.
+        match(Entry.class, entry -> entry.key.equals(key) && counter == null, entry -> {
+          // Reply from Storage of the initial value, now we can create the Counter
+          final long value = entry.value;
+          counter = context().actorOf(Props.create(Counter.class, key, value));
+          // Tell the counter to use current storage
+          counter.tell(new UseStorage(storage), self());
+          // and send the buffered backlog to the counter
+          for (SenderMsgPair each : backlog) {
+            counter.tell(each.msg, each.sender);
+          }
+          backlog.clear();
+        }).
+        match(Increment.class, increment -> {
+          forwardOrPlaceInBacklog(increment);
+        }).
+        matchEquals(GetCurrentCount, gcc -> {
+          forwardOrPlaceInBacklog(gcc);
+        }).
+        match(Terminated.class, o -> {
+          // After 3 restarts the storage child is stopped.
+          // We receive Terminated because we watch the child, see initStorage.
+          storage = null;
+          // Tell the counter that there is no storage for the moment
+          counter.tell(new UseStorage(null), self());
+          // Try to re-establish storage after while
+          context().system().scheduler().scheduleOnce(
+            Duration.create(10, "seconds"), self(), Reconnect,
+            context().dispatcher(), null);
+        }).
+        matchEquals(Reconnect, o -> {
+          // Re-establish storage after the scheduled delay
+          initStorage();
+        }).build(), context());
+    }
+
+    void forwardOrPlaceInBacklog(Object msg) {
+      // We need the initial value from storage before we can start delegate to
+      // the counter. Before that we place the messages in a backlog, to be sent
+      // to the counter when it is initialized.
+      if (counter == null) {
+        if (backlog.size() >= MAX_BACKLOG)
+          throw new ServiceUnavailable("CounterService not available," +
+            " lack of initial value");
+        backlog.add(new SenderMsgPair(sender(), msg));
+      } else {
+        counter.forward(msg, context());
+      }
+    }
+  }
+
+  //#messages
+  public interface CounterApi {
+    public static class UseStorage {
+      public final ActorRef storage;
+
+      public UseStorage(ActorRef storage) {
+        this.storage = storage;
+      }
+
+      public String toString() {
+        return String.format("%s(%s)", getClass().getSimpleName(), storage);
+      }
+    }
+  }
+
+  //#messages
+
+  /**
+   * The in memory count variable that will send current value to the Storage,
+   * if there is any storage available at the moment.
+   */
+  public static class Counter extends AbstractLoggingActor {
+    final String key;
+    long count;
+    ActorRef storage;
+
+    public Counter(String key, long initialValue) {
+      this.key = key;
+      this.count = initialValue;
+    }
+
+    @Override
+    public PartialFunction<Object, BoxedUnit> receive() {
+      return LoggingReceive.create(ReceiveBuilder.
+        match(UseStorage.class, useStorage -> {
+          storage = useStorage.storage;
+          storeCount();
+        }).
+        match(Increment.class, increment -> {
+          count += increment.n;
+          storeCount();
+        }).
+        matchEquals(GetCurrentCount, gcc -> {
+          sender().tell(new CurrentCount(key, count), self());
+        }).build(), context());
+    }
+
+    void storeCount() {
+      // Delegate dangerous work, to protect our valuable state.
+      // We can continue without storage.
+      if (storage != null) {
+        storage.tell(new Store(new Entry(key, count)), self());
+      }
+    }
+  }
+
+  //#messages
+  public interface StorageApi {
+
+    public static class Store {
+      public final Entry entry;
+
+      public Store(Entry entry) {
+        this.entry = entry;
+      }
+
+      public String toString() {
+        return String.format("%s(%s)", getClass().getSimpleName(), entry);
+      }
+    }
+
+    public static class Entry {
+      public final String key;
+      public final long value;
+
+      public Entry(String key, long value) {
+        this.key = key;
+        this.value = value;
+      }
+
+      public String toString() {
+        return String.format("%s(%s, %s)", getClass().getSimpleName(), key, value);
+      }
+    }
+
+    public static class Get {
+      public final String key;
+
+      public Get(String key) {
+        this.key = key;
+      }
+
+      public String toString() {
+        return String.format("%s(%s)", getClass().getSimpleName(), key);
+      }
+    }
+
+    public static class StorageException extends RuntimeException {
+      private static final long serialVersionUID = 1L;
+      public StorageException(String msg) {
+        super(msg);
+      }
+    }
+  }
+
+  //#messages
+
+  /**
+   * Saves key/value pairs to persistent storage when receiving Store message.
+   * Replies with current value when receiving Get message. Will throw
+   * StorageException if the underlying data store is out of order.
+   */
+  public static class Storage extends AbstractLoggingActor {
+
+    final DummyDB db = DummyDB.instance;
+
+    @Override
+    public PartialFunction<Object, BoxedUnit> receive() {
+      return LoggingReceive.create(ReceiveBuilder.
+        match(Store.class, store -> {
+          db.save(store.entry.key, store.entry.value);
+        }).
+        match(Get.class, get -> {
+          Long value = db.load(get.key);
+          sender().tell(new Entry(get.key, value == null ?
+            Long.valueOf(0L) : value), self());
+        }).build(), context());
+    }
+  }
+
+  //#dummydb
+  public static class DummyDB {
+    public static final DummyDB instance = new DummyDB();
+    private final Map<String, Long> db = new HashMap<String, Long>();
+
+    private DummyDB() {
+    }
+
+    public synchronized void save(String key, Long value) throws StorageException {
+      if (11 <= value && value <= 14)
+        throw new StorageException("Simulated store failure " + value);
+      db.put(key, value);
+    }
+
+    public synchronized Long load(String key) throws StorageException {
+      return db.get(key);
+    }
+  }
+  //#dummydb
+}
+//#all