+sam #3925 Adding Supervision for Java with Lambda Support Activator template

akka-samples/akka-sample-supervision-java-lambda/src/main/java/supervision/ArithmeticService.java

@@ -0,0 +1,75 @@
+package supervision;
+
+import akka.actor.*;
+import akka.japi.pf.DeciderBuilder;
+import akka.japi.pf.ReceiveBuilder;
+import scala.PartialFunction;
+import scala.runtime.BoxedUnit;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static  supervision.FlakyExpressionCalculator.FlakinessException;
+import static  supervision.FlakyExpressionCalculator.Result;
+import static  supervision.FlakyExpressionCalculator.Position.Left;
+
+// A very simple service that accepts arithmetic expressions and tries to
+// evaluate them. Since the calculation is dangerous (at least for the sake
+// of this example) it is delegated to a worker actor of type
+// FlakyExpressionCalculator.
+class ArithmeticService extends AbstractLoggingActor {
+
+  // Map of workers to the original actors requesting the calculation
+  Map<ActorRef, ActorRef> pendingWorkers = new HashMap<>();
+
+  private SupervisorStrategy strategy = new OneForOneStrategy(false, DeciderBuilder.
+    match(FlakinessException.class, e -> {
+      log().warning("Evaluation of a top level expression failed, restarting.");
+      return SupervisorStrategy.restart();
+    }).
+    match(ArithmeticException.class, e -> {
+      log().error("Evaluation failed because of: {}", e.getMessage());
+      notifyConsumerFailure(sender(), e);
+      return SupervisorStrategy.stop();
+    }).
+    match(Throwable.class, e -> {
+      log().error("Unexpected failure: {}", e.getMessage());
+      notifyConsumerFailure(sender(), e);
+      return SupervisorStrategy.stop();
+    }).build());
+  @Override
+  public SupervisorStrategy supervisorStrategy() {
+    return strategy;
+  }
+
+  private void notifyConsumerFailure(ActorRef worker, Throwable failure) {
+    // Status.Failure is a message type provided by the Akka library. The
+    // reason why it is used is because it is recognized by the "ask" pattern
+    // and the Future returned by ask will fail with the provided exception.
+    ActorRef pending = pendingWorkers.get(worker);
+    if (pending != null) {
+      pending.tell(new Status.Failure(failure), self());
+      pendingWorkers.remove(worker);
+    }
+  }
+
+  private void notifyConsumerSuccess(ActorRef worker, Integer result) {
+    ActorRef pending = pendingWorkers.get(worker);
+    if (pending != null) {
+      pending.tell(result, self());
+      pendingWorkers.remove(worker);
+    }
+  }
+
+  @Override
+  public PartialFunction<Object, BoxedUnit> receive() {
+    return ReceiveBuilder.
+      match(Expression.class, expr -> {
+        // We delegate the dangerous task of calculation to a worker, passing the
+        // expression as a constructor argument to the actor.
+        ActorRef worker = context().actorOf(FlakyExpressionCalculator.props(expr, Left));
+        pendingWorkers.put(worker, sender());
+      }).
+      match(Result.class, r -> notifyConsumerSuccess(sender(), r.getValue())).build();
+  }
+}