raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

improve AbstractActor, #21717

Browse source 
* Receive class that wraps PartialFunction, to avoid
  scary scala types
* move AbstractActorContext to AbstractActor.ActorContext
* converting docs, many, many UntypedActor
* removing UntypedActor docs
* add unit test for ReceiveBuilder
* MiMa filters
* consistent use of getContext(), self(), sender()
* rename cross references
* migration guide
* skip samples for now
* improve match type safetyi, add matchUnchecked
  * the `? extends P` caused code like this to compile:
    `match(String.class, (Integer i) -> {})`
  * added matchUnchecked, since it can still be useful (um, convenient)
    to be able to do:
    `matchUnchecked(List.class, (List<String> list) -> {})`
* eleminate some scala.Option
  * preRestart
  * findChild
  * ActorIdentity.getActorRef
This commit is contained in:
Patrik Nordwall 2016-12-13 10:59:29 +01:00
parent 3617fe8b41
commit 4bd6b7aab1
157 changed files with 3290 additions and 8882 deletions
Download patch file Download diff file Expand all files Collapse all files

639
akka-docs/rst/java/code/docs/actor/ActorDocTest.java
View file

@ -1,639 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
import akka.actor.*;
import akka.japi.pf.ReceiveBuilder;
import akka.testkit.ErrorFilter;
import akka.testkit.EventFilter;
import akka.testkit.TestEvent;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import docs.AbstractJavaTest;
import org.scalatest.junit.JUnitSuite;
import scala.PartialFunction;
import scala.runtime.BoxedUnit;
import static docs.actor.Messages.Swap.Swap;
import static akka.japi.Util.immutableSeq;
import java.util.concurrent.TimeUnit;
import akka.testkit.JavaTestKit;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import static org.junit.Assert.*;
//#import-props
import akka.actor.Props;
//#import-props
//#import-actorRef
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
//#import-actorRef
//#import-identify
import akka.actor.ActorIdentity;
import akka.actor.ActorSelection;
import akka.actor.Identify;
//#import-identify
//#import-graceFulStop
import akka.pattern.AskTimeoutException;
import scala.concurrent.Await;
import scala.concurrent.duration.Duration;
import scala.concurrent.Future;
import static akka.pattern.Patterns.gracefulStop;
//#import-graceFulStop
public class ActorDocTest extends AbstractJavaTest {
public static Config config = ConfigFactory.parseString(
"akka {\n" +
" loggers = [\"akka.testkit.TestEventListener\"]\n" +
" loglevel = \"WARNING\"\n" +
" stdout-loglevel = \"WARNING\"\n" +
"}\n"
);
static ActorSystem system = null;
@BeforeClass
public static void beforeClass() {
system = ActorSystem.create("ActorDocTest", config);
}
@AfterClass
public static void afterClass() throws Exception {
Await.result(system.terminate(), Duration.create("5 seconds"));
}
static
//#context-actorOf
public class FirstActor extends AbstractActor {
final ActorRef child = context().actorOf(Props.create(MyActor.class), "myChild");
//#plus-some-behavior
public FirstActor() {
receive(ReceiveBuilder.
matchAny(x -> {
sender().tell(x, self());
}).build()
);
}
//#plus-some-behavior
}
//#context-actorOf
static public abstract class SomeActor extends AbstractActor {
//#receive-constructor
public SomeActor() {
receive(ReceiveBuilder.
//#and-some-behavior
match(String.class, s -> System.out.println(s.toLowerCase())).
//#and-some-behavior
build());
}
//#receive-constructor
@Override
//#receive
public abstract PartialFunction<Object, BoxedUnit> receive();
//#receive
}
static public class ActorWithArgs extends AbstractActor {
private final String args;
ActorWithArgs(String args) {
this.args = args;
receive(ReceiveBuilder.
matchAny(x -> { }).build()
);
}
}
static
//#props-factory
public class DemoActor extends AbstractActor {
/**
* Create Props for an actor of this type.
* @param magicNumber The magic number to be passed to this actor’s constructor.
* @return a Props for creating this actor, which can then be further configured
* (e.g. calling `.withDispatcher()` on it)
*/
static Props props(Integer magicNumber) {
// You need to specify the actual type of the returned actor
// since Java 8 lambdas have some runtime type information erased
return Props.create(DemoActor.class, () -> new DemoActor(magicNumber));
}
private final Integer magicNumber;
DemoActor(Integer magicNumber) {
this.magicNumber = magicNumber;
receive(ReceiveBuilder.
match(Integer.class, i -> {
sender().tell(i + magicNumber, self());
}).build()
);
}
}
//#props-factory
static
//#props-factory
public class SomeOtherActor extends AbstractActor {
// Props(new DemoActor(42)) would not be safe
ActorRef demoActor = context().actorOf(DemoActor.props(42), "demo");
// ...
//#props-factory
public SomeOtherActor() {
receive(emptyBehavior());
}
//#props-factory
}
//#props-factory
public static class Hook extends AbstractActor {
ActorRef target = null;
public Hook() {
receive(emptyBehavior());
}
//#preStart
@Override
public void preStart() {
target = context().actorOf(Props.create(MyActor.class, "target"));
}
//#preStart
//#postStop
@Override
public void postStop() {
//#clean-up-some-resources
final String message = "stopped";
//#tell
// don’t forget to think about who is the sender (2nd argument)
target.tell(message, self());
//#tell
final Object result = "";
//#forward
target.forward(result, context());
//#forward
target = null;
//#clean-up-some-resources
}
//#postStop
// compilation test only
public void compileSelections() {
//#selection-local
// will look up this absolute path
context().actorSelection("/user/serviceA/actor");
// will look up sibling beneath same supervisor
context().actorSelection("../joe");
//#selection-local
//#selection-wildcard
// will look all children to serviceB with names starting with worker
context().actorSelection("/user/serviceB/worker*");
// will look up all siblings beneath same supervisor
context().actorSelection("../*");
//#selection-wildcard
//#selection-remote
context().actorSelection("akka.tcp://app@otherhost:1234/user/serviceB");
//#selection-remote
}
}
public static class ReplyException extends AbstractActor {
public ReplyException() {
receive(ReceiveBuilder.
matchAny(x -> {
//#reply-exception
try {
String result = operation();
sender().tell(result, self());
} catch (Exception e) {
sender().tell(new akka.actor.Status.Failure(e), self());
throw e;
}
//#reply-exception
}).build()
);
}
private String operation() {
return "Hi";
}
}
static
//#gracefulStop-actor
public class Manager extends AbstractActor {
private static enum Shutdown {
Shutdown
}
public static final Shutdown SHUTDOWN = Shutdown.Shutdown;
private ActorRef worker =
context().watch(context().actorOf(Props.create(Cruncher.class), "worker"));
public Manager() {
receive(ReceiveBuilder.
matchEquals("job", s -> {
worker.tell("crunch", self());
}).
matchEquals(SHUTDOWN, x -> {
worker.tell(PoisonPill.getInstance(), self());
context().become(shuttingDown);
}).build()
);
}
public PartialFunction<Object, BoxedUnit> shuttingDown =
ReceiveBuilder.
matchEquals("job", s -> {
sender().tell("service unavailable, shutting down", self());
}).
match(Terminated.class, t -> t.actor().equals(worker), t -> {
context().stop(self());
}).build();
}
//#gracefulStop-actor
@Test
public void usePatternsGracefulStop() throws Exception {
ActorRef actorRef = system.actorOf(Props.create(Manager.class));
//#gracefulStop
try {
Future<Boolean> stopped =
gracefulStop(actorRef, Duration.create(5, TimeUnit.SECONDS), Manager.SHUTDOWN);
Await.result(stopped, Duration.create(6, TimeUnit.SECONDS));
// the actor has been stopped
} catch (AskTimeoutException e) {
// the actor wasn't stopped within 5 seconds
}
//#gracefulStop
}
public static class Cruncher extends AbstractActor {
public Cruncher() {
receive(ReceiveBuilder.
matchEquals("crunch", s -> { }).build()
);
}
}
static
//#swapper
public class Swapper extends AbstractLoggingActor {
public Swapper() {
receive(ReceiveBuilder.
matchEquals(Swap, s -> {
log().info("Hi");
context().become(ReceiveBuilder.
matchEquals(Swap, x -> {
log().info("Ho");
context().unbecome(); // resets the latest 'become' (just for fun)
}).build(), false); // push on top instead of replace
}).build()
);
}
}
//#swapper
static
//#swapper
public class SwapperApp {
public static void main(String[] args) {
ActorSystem system = ActorSystem.create("SwapperSystem");
ActorRef swapper = system.actorOf(Props.create(Swapper.class), "swapper");
swapper.tell(Swap, ActorRef.noSender()); // logs Hi
swapper.tell(Swap, ActorRef.noSender()); // logs Ho
swapper.tell(Swap, ActorRef.noSender()); // logs Hi
swapper.tell(Swap, ActorRef.noSender()); // logs Ho
swapper.tell(Swap, ActorRef.noSender()); // logs Hi
swapper.tell(Swap, ActorRef.noSender()); // logs Ho
system.terminate();
}
}
//#swapper
@Test
public void creatingActorWithSystemActorOf() {
//#system-actorOf
// ActorSystem is a heavy object: create only one per application
final ActorSystem system = ActorSystem.create("MySystem", config);
final ActorRef myActor = system.actorOf(Props.create(FirstActor.class), "myactor");
//#system-actorOf
try {
new JavaTestKit(system) {
{
myActor.tell("hello", getRef());
expectMsgEquals("hello");
}
};
} finally {
JavaTestKit.shutdownActorSystem(system);
}
}
@Test
public void creatingPropsConfig() {
//#creating-props
Props props1 = Props.create(MyActor.class);
Props props2 = Props.create(ActorWithArgs.class,
() -> new ActorWithArgs("arg")); // careful, see below
Props props3 = Props.create(ActorWithArgs.class, "arg");
//#creating-props
//#creating-props-deprecated
// NOT RECOMMENDED within another actor:
// encourages to close over enclosing class
Props props7 = Props.create(ActorWithArgs.class,
() -> new ActorWithArgs("arg"));
//#creating-props-deprecated
}
@Test(expected=IllegalArgumentException.class)
public void creatingPropsIllegal() {
//#creating-props-illegal
// This will throw an IllegalArgumentException since some runtime
// type information of the lambda is erased.
// Use Props.create(actorClass, Creator) instead.
Props props = Props.create(() -> new ActorWithArgs("arg"));
//#creating-props-illegal
}
static
//#receive-timeout
public class ReceiveTimeoutActor extends AbstractActor {
//#receive-timeout
ActorRef target = context().system().deadLetters();
//#receive-timeout
public ReceiveTimeoutActor() {
// To set an initial delay
context().setReceiveTimeout(Duration.create("10 seconds"));
receive(ReceiveBuilder.
matchEquals("Hello", s -> {
// To set in a response to a message
context().setReceiveTimeout(Duration.create("1 second"));
//#receive-timeout
target = sender();
target.tell("Hello world", self());
//#receive-timeout
}).
match(ReceiveTimeout.class, r -> {
// To turn it off
context().setReceiveTimeout(Duration.Undefined());
//#receive-timeout
target.tell("timeout", self());
//#receive-timeout
}).build()
);
}
}
//#receive-timeout
@Test
public void using_receiveTimeout() {
final ActorRef myActor = system.actorOf(Props.create(ReceiveTimeoutActor.class));
new JavaTestKit(system) {
{
myActor.tell("Hello", getRef());
expectMsgEquals("Hello world");
expectMsgEquals("timeout");
}
};
}
static
//#hot-swap-actor
public class HotSwapActor extends AbstractActor {
private PartialFunction<Object, BoxedUnit> angry;
private PartialFunction<Object, BoxedUnit> happy;
public HotSwapActor() {
angry =
ReceiveBuilder.
matchEquals("foo", s -> {
sender().tell("I am already angry?", self());
}).
matchEquals("bar", s -> {
context().become(happy);
}).build();
happy = ReceiveBuilder.
matchEquals("bar", s -> {
sender().tell("I am already happy :-)", self());
}).
matchEquals("foo", s -> {
context().become(angry);
}).build();
receive(ReceiveBuilder.
matchEquals("foo", s -> {
context().become(angry);
}).
matchEquals("bar", s -> {
context().become(happy);
}).build()
);
}
}
//#hot-swap-actor
@Test
public void using_hot_swap() {
final ActorRef actor = system.actorOf(Props.create(HotSwapActor.class), "hot");
new JavaTestKit(system) {
{
actor.tell("foo", getRef());
actor.tell("foo", getRef());
expectMsgEquals("I am already angry?");
actor.tell("bar", getRef());
actor.tell("bar", getRef());
expectMsgEquals("I am already happy :-)");
actor.tell("foo", getRef());
actor.tell("foo", getRef());
expectMsgEquals("I am already angry?");
expectNoMsg(Duration.create(1, TimeUnit.SECONDS));
}
};
}
static
//#stash
public class ActorWithProtocol extends AbstractActorWithStash {
public ActorWithProtocol() {
receive(ReceiveBuilder.
matchEquals("open", s -> {
context().become(ReceiveBuilder.
matchEquals("write", ws -> { /* do writing */ }).
matchEquals("close", cs -> {
unstashAll();
context().unbecome();
}).
matchAny(msg -> stash()).build(), false);
}).
matchAny(msg -> stash()).build()
);
}
}
//#stash
@Test
public void using_Stash() {
final ActorRef actor = system.actorOf(Props.create(ActorWithProtocol.class), "stash");
}
static
//#watch
public class WatchActor extends AbstractActor {
private final ActorRef child = context().actorOf(Props.empty(), "target");
private ActorRef lastSender = system.deadLetters();
public WatchActor() {
context().watch(child); // <-- this is the only call needed for registration
receive(ReceiveBuilder.
matchEquals("kill", s -> {
context().stop(child);
lastSender = sender();
}).
match(Terminated.class, t -> t.actor().equals(child), t -> {
lastSender.tell("finished", self());
}).build()
);
}
}
//#watch
@Test
public void using_watch() {
ActorRef actor = system.actorOf(Props.create(WatchActor.class));
new JavaTestKit(system) {
{
actor.tell("kill", getRef());
expectMsgEquals("finished");
}
};
}
static
//#identify
public class Follower extends AbstractActor {
final Integer identifyId = 1;
public Follower(){
ActorSelection selection = context().actorSelection("/user/another");
selection.tell(new Identify(identifyId), self());
receive(ReceiveBuilder.
match(ActorIdentity.class, id -> id.getRef() != null, id -> {
ActorRef ref = id.getRef();
context().watch(ref);
context().become(active(ref));
}).
match(ActorIdentity.class, id -> id.getRef() == null, id -> {
context().stop(self());
}).build()
);
}
final PartialFunction<Object, BoxedUnit> active(final ActorRef another) {
return ReceiveBuilder.
match(Terminated.class, t -> t.actor().equals(another), t -> {
context().stop(self());
}).build();
}
}
//#identify
@Test
public void using_Identify() {
ActorRef a = system.actorOf(Props.empty());
ActorRef b = system.actorOf(Props.create(Follower.class));
new JavaTestKit(system) {
{
watch(b);
system.stop(a);
assertEquals(expectMsgClass(Duration.create(2, TimeUnit.SECONDS), Terminated.class).actor(), b);
}
};
}
public static class NoReceiveActor extends AbstractActor {
}
@Test
public void noReceiveActor() {
EventFilter ex1 = new ErrorFilter(ActorInitializationException.class);
EventFilter[] ignoreExceptions = { ex1 };
try {
system.eventStream().publish(new TestEvent.Mute(immutableSeq(ignoreExceptions)));
new JavaTestKit(system) {{
final ActorRef victim = new EventFilter<ActorRef>(ActorInitializationException.class) {
protected ActorRef run() {
return system.actorOf(Props.create(NoReceiveActor.class), "victim");
}
}.message("Actor behavior has not been set with receive(...)").occurrences(1).exec();
assertEquals(true, victim.isTerminated());
}};
} finally {
system.eventStream().publish(new TestEvent.UnMute(immutableSeq(ignoreExceptions)));
}
}
public static class MultipleReceiveActor extends AbstractActor {
public MultipleReceiveActor() {
receive(ReceiveBuilder.
match(String.class, s1 -> s1.toLowerCase().equals("become"), s1 -> {
sender().tell(s1.toUpperCase(), self());
receive(ReceiveBuilder.
match(String.class, s2 -> {
sender().tell(s2.toLowerCase(), self());
}).build()
);
}).
match(String.class, s1 -> {
sender().tell(s1.toUpperCase(), self());
}).build()
);
}
}
@Test
public void multipleReceiveActor() {
EventFilter ex1 = new ErrorFilter(IllegalActorStateException.class);
EventFilter[] ignoreExceptions = { ex1 };
try {
system.eventStream().publish(new TestEvent.Mute(immutableSeq(ignoreExceptions)));
new JavaTestKit(system) {{
new EventFilter<Boolean>(IllegalActorStateException.class) {
protected Boolean run() {
ActorRef victim = system.actorOf(Props.create(MultipleReceiveActor.class), "victim2");
victim.tell("Foo", getRef());
expectMsgEquals("FOO");
victim.tell("bEcoMe", getRef());
expectMsgEquals("BECOME");
victim.tell("Foo", getRef());
// if it's upper case, then the actor was restarted
expectMsgEquals("FOO");
return true;
}
}.message("Actor behavior has already been set with receive(...), " +
"use context().become(...) to change it later").occurrences(1).exec();
}};
} finally {
system.eventStream().publish(new TestEvent.UnMute(immutableSeq(ignoreExceptions)));
}
}
}

209
akka-docs/rst/java/code/docs/actor/FSMDocTest.java
View file

@ -1,209 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
//#imports-data
import java.util.ArrayList;
import java.util.List;
import akka.actor.ActorRef;
//#imports-data
//#imports-actor
import akka.event.LoggingAdapter;
import akka.event.Logging;
import akka.actor.UntypedActor;
//#imports-actor
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.testkit.JavaTestKit;
import akka.testkit.TestProbe;
import akka.testkit.AkkaSpec;
import docs.AbstractJavaTest;
public class FSMDocTest extends AbstractJavaTest {
static
//#data
public final class SetTarget {
final ActorRef ref;
public SetTarget(ActorRef ref) {
this.ref = ref;
}
}
//#data
static
//#data
public final class Queue {
final Object o;
public Queue(Object o) {
this.o = o;
}
}
//#data
static
//#data
public final Object flush = new Object();
//#data
static
//#data
public final class Batch {
final List<Object> objects;
public Batch(List<Object> objects) {
this.objects = objects;
}
}
//#data
static
//#base
public abstract class MyFSMBase extends UntypedActor {
/*
* This is the mutable state of this state machine.
*/
protected enum State {
IDLE, ACTIVE;
}
private State state = State.IDLE;
private ActorRef target;
private List<Object> queue;
/*
* Then come all the mutator methods:
*/
protected void init(ActorRef target) {
this.target = target;
queue = new ArrayList<Object>();
}
protected void setState(State s) {
if (state != s) {
transition(state, s);
state = s;
}
}
protected void enqueue(Object o) {
if (queue != null)
queue.add(o);
}
protected List<Object> drainQueue() {
final List<Object> q = queue;
if (q == null)
throw new IllegalStateException("drainQueue(): not yet initialized");
queue = new ArrayList<Object>();
return q;
}
/*
* Here are the interrogation methods:
*/
protected boolean isInitialized() {
return target != null;
}
protected State getState() {
return state;
}
protected ActorRef getTarget() {
if (target == null)
throw new IllegalStateException("getTarget(): not yet initialized");
return target;
}
/*
* And finally the callbacks (only one in this example: react to state change)
*/
abstract protected void transition(State old, State next);
}
//#base
static
//#actor
public class MyFSM extends MyFSMBase {
private final LoggingAdapter log =
Logging.getLogger(getContext().system(), this);
@Override
public void onReceive(Object o) {
if (getState() == State.IDLE) {
if (o instanceof SetTarget)
init(((SetTarget) o).ref);
else
whenUnhandled(o);
} else if (getState() == State.ACTIVE) {
if (o == flush)
setState(State.IDLE);
else
whenUnhandled(o);
}
}
@Override
public void transition(State old, State next) {
if (old == State.ACTIVE) {
getTarget().tell(new Batch(drainQueue()), getSelf());
}
}
private void whenUnhandled(Object o) {
if (o instanceof Queue && isInitialized()) {
enqueue(((Queue) o).o);
setState(State.ACTIVE);
} else {
log.warning("received unknown message {} in state {}", o, getState());
}
}
}
//#actor
ActorSystem system;
@org.junit.Before
public void setUp() {
system = ActorSystem.create("FSMSystem", AkkaSpec.testConf());
}
@org.junit.Test
public void mustBunch() {
final ActorRef buncher = system.actorOf(Props.create(MyFSM.class));
final TestProbe probe = new TestProbe(system);
buncher.tell(new SetTarget(probe.ref()), ActorRef.noSender());
buncher.tell(new Queue(1), ActorRef.noSender());
buncher.tell(new Queue(2), ActorRef.noSender());
buncher.tell(flush, ActorRef.noSender());
buncher.tell(new Queue(3), ActorRef.noSender());
final Batch b = probe.expectMsgClass(Batch.class);
assert b.objects.size() == 2;
assert b.objects.contains(1);
assert b.objects.contains(2);
}
@org.junit.After
public void cleanup() {
JavaTestKit.shutdownActorSystem(system);
}
}

224
akka-docs/rst/java/code/docs/actor/FaultHandlingTest.java
View file

@ -1,224 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
//#testkit
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.SupervisorStrategy;
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.actor.SupervisorStrategy.Directive;
import akka.actor.OneForOneStrategy;
import akka.actor.Props;
import akka.actor.Terminated;
import akka.actor.UntypedActor;
import docs.AbstractJavaTest;
import scala.collection.immutable.Seq;
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 akka.japi.Function;
import scala.Option;
import org.junit.Test;
import org.junit.BeforeClass;
import org.junit.AfterClass;
//#testkit
public class FaultHandlingTest extends AbstractJavaTest {
//#testkit
static
//#supervisor
public class Supervisor extends UntypedActor {
//#strategy
private static SupervisorStrategy strategy =
new OneForOneStrategy(10, Duration.create("1 minute"),
new Function<Throwable, Directive>() {
@Override
public Directive apply(Throwable t) {
if (t instanceof ArithmeticException) {
return resume();
} else if (t instanceof NullPointerException) {
return restart();
} else if (t instanceof IllegalArgumentException) {
return stop();
} else {
return escalate();
}
}
});
@Override
public SupervisorStrategy supervisorStrategy() {
return strategy;
}
//#strategy
public void onReceive(Object o) {
if (o instanceof Props) {
getSender().tell(getContext().actorOf((Props) o), getSelf());
} else {
unhandled(o);
}
}
}
//#supervisor
static
//#supervisor2
public class Supervisor2 extends UntypedActor {
//#strategy2
private static SupervisorStrategy strategy = new OneForOneStrategy(10,
Duration.create("1 minute"),
new Function<Throwable, Directive>() {
@Override
public Directive apply(Throwable t) {
if (t instanceof ArithmeticException) {
return resume();
} else if (t instanceof NullPointerException) {
return restart();
} else if (t instanceof IllegalArgumentException) {
return stop();
} else {
return escalate();
}
}
});
@Override
public SupervisorStrategy supervisorStrategy() {
return strategy;
}
//#strategy2
public void onReceive(Object o) {
if (o instanceof Props) {
getSender().tell(getContext().actorOf((Props) o), getSelf());
} else {
unhandled(o);
}
}
@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 UntypedActor {
int state = 0;
public void onReceive(Object o) throws Exception {
if (o instanceof Exception) {
throw (Exception) o;
} else if (o instanceof Integer) {
state = (Integer) o;
} else if (o.equals("get")) {
getSender().tell(state, getSelf());
} else {
unhandled(o);
}
}
}
//#child
//#testkit
static ActorSystem system;
Duration timeout = Duration.create(5, SECONDS);
@BeforeClass
public static void start() {
system = ActorSystem.create("FaultHandlingTest");
}
@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 };
Seq<EventFilter> seq = immutableSeq(ignoreExceptions);
system.eventStream().publish(new TestEvent.Mute(seq));
//#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

205
akka-docs/rst/java/code/docs/actor/FaultHandlingTestJava8.java
View file

@ -1,205 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.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 org.scalatest.junit.JUnitSuite;
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 FaultHandlingTestJava8 extends JUnitSuite {
//#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
public Supervisor() {
receive(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
public Supervisor2() {
receive(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;
public Child() {
receive(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

21
akka-docs/rst/java/code/docs/actor/FirstUntypedActor.java
View file

@ -1,21 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.PoisonPill;
import akka.actor.UntypedActor;
//#context-actorOf
public class FirstUntypedActor extends UntypedActor {
ActorRef myActor = getContext().actorOf(Props.create(MyActor.class), "myactor");
//#context-actorOf
public void onReceive(Object message) {
myActor.forward(message, getContext());
myActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
}
}

87
akka-docs/rst/java/code/docs/actor/InitializationDocSpecJava.java
View file

@ -1,87 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
import akka.actor.*;
import akka.japi.Procedure;
import akka.testkit.AkkaJUnitActorSystemResource;
import akka.testkit.JavaTestKit;
import docs.AbstractJavaTest;
import org.junit.ClassRule;
import org.junit.Test;
import scala.Option;
public class InitializationDocSpecJava extends AbstractJavaTest {
static public class PreStartInitExample extends UntypedActor {
public void onReceive(Object message) throws Exception {}
//#preStartInit
@Override
public void preStart() {
// Initialize children here
}
// Overriding postRestart to disable the call to preStart()
// after restarts
@Override
public void postRestart(Throwable reason) {
}
// The default implementation of preRestart() stops all the children
// of the actor. To opt-out from stopping the children, we
// have to override preRestart()
@Override
public void preRestart(Throwable reason, Option<Object> message)
throws Exception {
// Keep the call to postStop(), but no stopping of children
postStop();
}
//#preStartInit
}
public static class MessageInitExample extends UntypedActor {
//#messageInit
private String initializeMe = null;
@Override
public void onReceive(Object message) throws Exception {
if (message.equals("init")) {
initializeMe = "Up and running";
getContext().become(new Procedure<Object>() {
@Override
public void apply(Object message) throws Exception {
if (message.equals("U OK?"))
getSender().tell(initializeMe, getSelf());
}
});
}
}
//#messageInit
}
@ClassRule
public static AkkaJUnitActorSystemResource actorSystemResource =
new AkkaJUnitActorSystemResource("InitializationDocSpecJava");
private final ActorSystem system = actorSystemResource.getSystem();
@Test
public void testIt() {
new JavaTestKit(system) {{
ActorRef testactor = system.actorOf(Props.create(MessageInitExample.class), "testactor");
String probe = "U OK?";
testactor.tell(probe, getRef());
expectNoMsg();
testactor.tell("init", getRef());
testactor.tell(probe, getRef());
expectMsgEquals("Up and running");
}};
}
}

68
akka-docs/rst/java/code/docs/actor/InitializationDocTest.java
View file

@ -1,68 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
import akka.actor.*;
import akka.japi.pf.ReceiveBuilder;
import akka.testkit.JavaTestKit;
import docs.AbstractJavaTest;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import scala.PartialFunction;
import scala.concurrent.duration.Duration;
import scala.runtime.BoxedUnit;
import scala.concurrent.Await;
import java.util.concurrent.TimeUnit;
public class InitializationDocTest extends AbstractJavaTest {
static ActorSystem system = null;
@BeforeClass
public static void beforeClass() {
system = ActorSystem.create("InitializationDocTest");
}
@AfterClass
public static void afterClass() throws Exception {
Await.result(system.terminate(), Duration.create("5 seconds"));
}
public static class MessageInitExample extends AbstractActor {
private String initializeMe = null;
public MessageInitExample() {
//#messageInit
receive(ReceiveBuilder.
matchEquals("init", m1 -> {
initializeMe = "Up and running";
context().become(ReceiveBuilder.
matchEquals("U OK?", m2 -> {
sender().tell(initializeMe, self());
}).build());
}).build()
//#messageInit
);
}
}
@Test
public void testIt() {
new JavaTestKit(system) {{
ActorRef testactor = system.actorOf(Props.create(MessageInitExample.class), "testactor");
String msg = "U OK?";
testactor.tell(msg, getRef());
expectNoMsg(Duration.create(1, TimeUnit.SECONDS));
testactor.tell("init", getRef());
testactor.tell(msg, getRef());
expectMsgEquals("Up and running");
}};
}
}

14
akka-docs/rst/java/code/docs/actor/MyBoundedUntypedActor.java
View file

@ -1,14 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
//#my-bounded-untyped-actor
import akka.dispatch.BoundedMessageQueueSemantics;
import akka.dispatch.RequiresMessageQueue;
public class MyBoundedUntypedActor extends MyUntypedActor
implements RequiresMessageQueue<BoundedMessageQueueSemantics> {
}
//#my-bounded-untyped-actor

33
akka-docs/rst/java/code/docs/actor/MyJavaActor.java
View file

@ -1,33 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
//#imports
import akka.actor.AbstractActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.japi.pf.ReceiveBuilder;
//#imports
//#my-actor
public class MyJavaActor extends AbstractActor {
private final LoggingAdapter log = Logging.getLogger(context().system(), this);
public MyJavaActor() {
receive(ReceiveBuilder.
match(String.class, s -> {
log.info("Received String message: {}", s);
//#my-actor
//#reply
sender().tell(s, self());
//#reply
//#my-actor
}).
matchAny(o -> log.info("received unknown message")).build()
);
}
}
//#my-actor

41
akka-docs/rst/java/code/docs/actor/MyReceiveTimeoutUntypedActor.java
View file

@ -1,41 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
//#receive-timeout
import akka.actor.ActorRef;
import akka.actor.ReceiveTimeout;
import akka.actor.UntypedActor;
import scala.concurrent.duration.Duration;
public class MyReceiveTimeoutUntypedActor extends UntypedActor {
//#receive-timeout
ActorRef target = getContext().system().deadLetters();
//#receive-timeout
public MyReceiveTimeoutUntypedActor() {
// To set an initial delay
getContext().setReceiveTimeout(Duration.create("30 seconds"));
}
public void onReceive(Object message) {
if (message.equals("Hello")) {
// To set in a response to a message
getContext().setReceiveTimeout(Duration.create("1 second"));
//#receive-timeout
target = getSender();
target.tell("Hello world", getSelf());
//#receive-timeout
} else if (message instanceof ReceiveTimeout) {
// To turn it off
getContext().setReceiveTimeout(Duration.Undefined());
//#receive-timeout
target.tell("timeout", getSelf());
//#receive-timeout
} else {
unhandled(message);
}
}
}
//#receive-timeout

29
akka-docs/rst/java/code/docs/actor/MyStoppingActor.java
View file

@ -1,29 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
//#my-stopping-actor
import akka.actor.ActorRef;
import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
public class MyStoppingActor extends UntypedActor {
ActorRef child = null;
// ... creation of child ...
public void onReceive(Object message) throws Exception {
if (message.equals("interrupt-child")) {
context().stop(child);
} else if (message.equals("done")) {
context().stop(getSelf());
} else {
unhandled(message);
}
}
}
//#my-stopping-actor

23
akka-docs/rst/java/code/docs/actor/MyUntypedActor.java
View file

@ -1,23 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
//#my-untyped-actor
import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
public class MyUntypedActor extends UntypedActor {
LoggingAdapter log = Logging.getLogger(getContext().system(), this);
public void onReceive(Object message) throws Exception {
if (message instanceof String) {
log.info("Received String message: {}", message);
getSender().tell(message, getSelf());
} else
unhandled(message);
}
}
//#my-untyped-actor

36
akka-docs/rst/java/code/docs/actor/SampleActor.java
View file

@ -1,36 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
//#sample-actor
import akka.actor.AbstractActor;
import akka.japi.pf.ReceiveBuilder;
import scala.PartialFunction;
import scala.runtime.BoxedUnit;
public class SampleActor extends AbstractActor {
private PartialFunction<Object, BoxedUnit> guarded = ReceiveBuilder.
match(String.class, s -> s.contains("guard"), s -> {
sender().tell("contains(guard): " + s, self());
context().unbecome();
}).build();
public SampleActor() {
receive(ReceiveBuilder.
match(Double.class, d -> {
sender().tell(d.isNaN() ? 0 : d, self());
}).
match(Integer.class, i -> {
sender().tell(i * 10, self());
}).
match(String.class, s -> s.startsWith("guard"), s -> {
sender().tell("startsWith(guard): " + s.toUpperCase(), self());
context().become(guarded, false);
}).build()
);
}
}
//#sample-actor

56
akka-docs/rst/java/code/docs/actor/SampleActorTest.java
View file

@ -1,56 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.testkit.JavaTestKit;
import docs.AbstractJavaTest;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import static org.junit.Assert.*;
public class SampleActorTest extends AbstractJavaTest {
static ActorSystem system;
@BeforeClass
public static void setup() {
system = ActorSystem.create("SampleActorTest");
}
@AfterClass
public static void tearDown() {
JavaTestKit.shutdownActorSystem(system);
system = null;
}
@Test
public void testSampleActor()
{
new JavaTestKit(system) {{
final ActorRef subject = system.actorOf(Props.create(SampleActor.class), "sample-actor");
final ActorRef probeRef = getRef();
subject.tell(47.11, probeRef);
subject.tell("and no guard in the beginning", probeRef);
subject.tell("guard is a good thing", probeRef);
subject.tell(47.11, probeRef);
subject.tell(4711, probeRef);
subject.tell("and no guard in the beginning", probeRef);
subject.tell(4711, probeRef);
subject.tell("and an unmatched message", probeRef);
expectMsgEquals(47.11);
assertTrue(expectMsgClass(String.class).startsWith("startsWith(guard):"));
assertTrue(expectMsgClass(String.class).startsWith("contains(guard):"));
expectMsgEquals(47110);
expectNoMsg();
}};
}
}

19
akka-docs/rst/java/code/docs/actor/SchedulerDocTest.java
View file

@ -11,10 +11,11 @@ import java.util.concurrent.TimeUnit;
//#imports1
//#imports2
import akka.actor.UntypedActor;
import akka.actor.Cancellable;
//#imports2
import docs.actorlambda.MyActor;
import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.testkit.AkkaSpec;
@ -28,7 +29,7 @@ public class SchedulerDocTest extends AbstractJavaTest {
AkkaSpec.testConf());
private final ActorSystem system = actorSystemResource.getSystem();
private ActorRef testActor = system.actorOf(Props.create(MyUntypedActor.class));
private ActorRef testActor = system.actorOf(Props.create(MyActor.class));
@Test
public void scheduleOneOffTask() {
@ -51,14 +52,14 @@ public class SchedulerDocTest extends AbstractJavaTest {
@Test
public void scheduleRecurringTask() {
//#schedule-recurring
class Ticker extends UntypedActor {
class Ticker extends AbstractActor {
@Override
public void onReceive(Object message) {
if (message.equals("Tick")) {
// Do someting
} else {
unhandled(message);
}
public Receive createReceive() {
return receiveBuilder()
.matchEquals("Tick", m -> {
// Do someting
})
.build();
}
}

794
akka-docs/rst/java/code/docs/actor/UntypedActorDocTest.java
View file

@ -1,794 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
//#import-ask
import static akka.pattern.Patterns.ask;
import static akka.pattern.Patterns.pipe;
//#import-ask
//#import-gracefulStop
import static akka.pattern.Patterns.gracefulStop;
//#import-gracefulStop
import akka.actor.PoisonPill;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;
import akka.testkit.AkkaJUnitActorSystemResource;
import docs.AbstractJavaTest;
import org.junit.ClassRule;
import org.junit.Test;
//#import-gracefulStop
import scala.concurrent.Await;
//#import-ask
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;
//#import-ask
//#import-gracefulStop
//#import-indirect
import akka.actor.Actor;
//#import-indirect
//#import-identify
import akka.actor.ActorIdentity;
//#import-identify
import akka.actor.ActorKilledException;
//#import-identify
import akka.actor.ActorSelection;
//#import-identify
//#import-actorRef
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
//#import-actorRef
//#import-identify
import akka.actor.Identify;
//#import-identify
//#import-indirect
import akka.actor.IndirectActorProducer;
//#import-indirect
import akka.actor.OneForOneStrategy;
//#import-props
import akka.actor.Props;
import akka.japi.Creator;
//#import-props
import akka.actor.SupervisorStrategy;
import akka.actor.SupervisorStrategy.Directive;
//#import-terminated
import akka.actor.Terminated;
//#import-terminated
//#import-untypedActor
import akka.actor.UntypedActor;
//#import-untypedActor
//#import-stash
import akka.actor.UntypedActorWithStash;
//#import-stash
//#import-ask
import akka.dispatch.Futures;
import akka.dispatch.Mapper;
//#import-ask
import akka.japi.Function;
//#import-procedure
import akka.japi.Procedure;
//#import-procedure
//#import-gracefulStop
import akka.pattern.AskTimeoutException;
//#import-gracefulStop
import akka.pattern.Patterns;
import akka.testkit.AkkaSpec;
import akka.testkit.JavaTestKit;
//#import-ask
import akka.util.Timeout;
//#import-ask
public class UntypedActorDocTest extends AbstractJavaTest {
@ClassRule
public static AkkaJUnitActorSystemResource actorSystemResource =
new AkkaJUnitActorSystemResource("UntypedActorDocTest", AkkaSpec.testConf());
private final ActorSystem system = actorSystemResource.getSystem();
//#creating-props-config
static class MyActorC implements Creator<MyActor> {
@Override public MyActor create() {
return new MyActor("...");
}
}
//#creating-props-config
@SuppressWarnings("unused")
@Test
public void createProps() {
//#creating-props-config
Props props1 = Props.create(MyUntypedActor.class);
Props props2 = Props.create(MyActor.class, "...");
Props props3 = Props.create(new MyActorC());
//#creating-props-config
}
//#parametric-creator
static class ParametricCreator<T extends MyActor> implements Creator<T> {
@Override public T create() {
// ... fabricate actor here
//#parametric-creator
return null;
//#parametric-creator
}
}
//#parametric-creator
@Test
public void systemActorOf() {
//#system-actorOf
// ActorSystem is a heavy object: create only one per application
final ActorSystem system = ActorSystem.create("MySystem");
final ActorRef myActor = system.actorOf(Props.create(MyUntypedActor.class),
"myactor");
//#system-actorOf
try {
new JavaTestKit(system) {
{
myActor.tell("hello", getRef());
expectMsgEquals("hello");
}
};
} finally {
JavaTestKit.shutdownActorSystem(system);
}
}
@Test
public void contextActorOf() {
new JavaTestKit(system) {
{
//#context-actorOf
class A extends UntypedActor {
final ActorRef child =
getContext().actorOf(Props.create(MyUntypedActor.class), "myChild");
//#plus-some-behavior
@Override
public void onReceive(Object msg) {
getSender().tell(child, getSelf());
}
//#plus-some-behavior
}
//#context-actorOf
final ActorRef top = system.actorOf(Props.create(A.class, this));
top.tell("hello", getRef());
final ActorRef child = expectMsgClass(ActorRef.class);
child.tell("hello", getRef());
expectMsgEquals("hello");
}
};
}
// this is just to make the test below a tiny fraction nicer
private ActorSystem getContext() {
return system;
}
static
//#creating-indirectly
class DependencyInjector implements IndirectActorProducer {
final Object applicationContext;
final String beanName;
public DependencyInjector(Object applicationContext, String beanName) {
this.applicationContext = applicationContext;
this.beanName = beanName;
}
@Override
public Class<? extends Actor> actorClass() {
return MyActor.class;
}
@Override
public MyActor produce() {
MyActor result;
//#obtain-fresh-Actor-instance-from-DI-framework
result = new MyActor((String) applicationContext);
//#obtain-fresh-Actor-instance-from-DI-framework
return result;
}
}
//#creating-indirectly
@Test
public void indirectActorOf() {
final String applicationContext = "...";
//#creating-indirectly
final ActorRef myActor = getContext().actorOf(
Props.create(DependencyInjector.class, applicationContext, "MyActor"),
"myactor3");
//#creating-indirectly
new JavaTestKit(system) {
{
myActor.tell("hello", getRef());
expectMsgEquals("...");
}
};
}
@SuppressWarnings("unused")
@Test
public void usingAsk() throws Exception {
ActorRef myActor = system.actorOf(Props.create(MyAskActor.class, this), "myactor5");
//#using-ask
Future<Object> future = Patterns.ask(myActor, "Hello", 1000);
Object result = Await.result(future, Duration.create(1, TimeUnit.SECONDS));
//#using-ask
}
@Test
public void receiveTimeout() {
final ActorRef myActor = system.actorOf(Props.create(MyReceiveTimeoutUntypedActor.class));
new JavaTestKit(system) {
{
new Within(Duration.create(1, TimeUnit.SECONDS), Duration.create(1500,
TimeUnit.MILLISECONDS)) {
@Override
protected void run() {
myActor.tell("Hello", getRef());
expectMsgEquals("Hello world");
expectMsgEquals("timeout");
}
};
}
};
}
@Test
public void usePoisonPill() {
final ActorRef myActor = system.actorOf(Props.create(MyUntypedActor.class));
new JavaTestKit(system) {
{
final ActorRef sender = getRef();
//#poison-pill
myActor.tell(akka.actor.PoisonPill.getInstance(), sender);
//#poison-pill
watch(myActor);
expectTerminated(myActor);
}
};
}
@Test
public void useKill() {
new JavaTestKit(system) {
{
class Master extends UntypedActor {
private SupervisorStrategy strategy = new OneForOneStrategy(-1,
Duration.Undefined(), new Function<Throwable, Directive>() {
@Override
public Directive apply(Throwable thr) {
if (thr instanceof ActorKilledException) {
target.tell("killed", getSelf());
getContext().stop(getSelf());
return SupervisorStrategy.stop();
}
return SupervisorStrategy.escalate();
}
});
final ActorRef target;
ActorRef child;
//#preStart
@Override
public void preStart() {
child = getContext().actorOf(Props.empty());
}
//#preStart
@SuppressWarnings("unused")
public Master(ActorRef target) {
this.target = target;
/*
* Only compilation of `forward` is verified here.
*/
final Object result = "";
//#forward
target.forward(result, getContext());
//#forward
}
@Override
public SupervisorStrategy supervisorStrategy() {
return strategy;
}
//#reply
@Override
public void onReceive(Object msg) {
Object result =
//#calculate-result
child;
//#calculate-result
// do not forget the second argument!
getSender().tell(result, getSelf());
}
//#reply
//#postStop
@Override
public void postStop() {
//#clean-up-resources-here
final String message = "stopped";
//#tell
// don’t forget to think about who is the sender (2nd argument)
target.tell(message, getSelf());
//#tell
//#clean-up-resources-here
}
//#postStop
}
final ActorRef master = system.actorOf(Props.create(Master.class, this, getRef()));
expectMsgEquals("");
master.tell("", getRef());
final ActorRef victim = expectMsgClass(ActorRef.class);
//#kill
victim.tell(akka.actor.Kill.getInstance(), ActorRef.noSender());
//#kill
expectMsgEquals("killed");
expectMsgEquals("stopped");
assert getLastSender().equals(master);
}
};
}
@Test
public void useBecome() {
new JavaTestKit(system) {
{
ActorRef myActor = system.actorOf(Props.create(HotSwapActor.class));
myActor.tell("foo", getRef());
myActor.tell("bar", getRef());
expectMsgEquals("I am already happy :-)");
myActor.tell("bar", getRef());
expectMsgEquals("I am already happy :-)");
}
};
}
@Test
public void useWatch() throws Exception {
ActorRef myActor = system.actorOf(Props.create(WatchActor.class));
Future<Object> future = Patterns.ask(myActor, "kill", 1000);
assert Await.result(future, Duration.create("1 second")).equals("finished");
}
// compilation test only
public void compileSelections() {
//#selection-local
// will look up this absolute path
getContext().actorSelection("/user/serviceA/actor");
// will look up sibling beneath same supervisor
getContext().actorSelection("../joe");
//#selection-local
//#selection-wildcard
// will look all children to serviceB with names starting with worker
getContext().actorSelection("/user/serviceB/worker*");
// will look up all siblings beneath same supervisor
getContext().actorSelection("../*");
//#selection-wildcard
//#selection-remote
getContext().actorSelection("akka.tcp://app@otherhost:1234/user/serviceB");
//#selection-remote
}
@Test
public void useIdentify() throws Exception {
new JavaTestKit(system) {
{
ActorRef a = system.actorOf(Props.create(MyUntypedActor.class), "another");
ActorRef b = system.actorOf(Props.create(Follower.class, getRef()));
expectMsgEquals(a);
system.stop(a);
watch(b);
expectTerminated(b);
}
};
}
@Test
public void usePatternsGracefulStop() throws Exception {
ActorRef actorRef = system.actorOf(Props.create(Manager.class));
//#gracefulStop
try {
Future<Boolean> stopped =
gracefulStop(actorRef, Duration.create(5, TimeUnit.SECONDS), Manager.SHUTDOWN);
Await.result(stopped, Duration.create(6, TimeUnit.SECONDS));
// the actor has been stopped
} catch (AskTimeoutException e) {
// the actor wasn't stopped within 5 seconds
}
//#gracefulStop
}
class Result {
final String x;
final String s;
public Result(String x, String s) {
this.x = x;
this.s = s;
}
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + ((s == null) ? 0 : s.hashCode());
result = prime * result + ((x == null) ? 0 : x.hashCode());
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj)
return true;
if (obj == null)
return false;
if (getClass() != obj.getClass())
return false;
Result other = (Result) obj;
if (s == null) {
if (other.s != null)
return false;
} else if (!s.equals(other.s))
return false;
if (x == null) {
if (other.x != null)
return false;
} else if (!x.equals(other.x))
return false;
return true;
}
}
@Test
public void usePatternsAskPipe() {
new JavaTestKit(system) {
{
ActorRef actorA = system.actorOf(Props.create(MyUntypedActor.class));
ActorRef actorB = system.actorOf(Props.create(MyUntypedActor.class));
ActorRef actorC = getRef();
//#ask-pipe
final Timeout t = new Timeout(Duration.create(5, TimeUnit.SECONDS));
final ArrayList<Future<Object>> futures = new ArrayList<Future<Object>>();
futures.add(ask(actorA, "request", 1000)); // using 1000ms timeout
futures.add(ask(actorB, "another request", t)); // using timeout from
// above
final Future<Iterable<Object>> aggregate = Futures.sequence(futures,
system.dispatcher());
final Future<Result> transformed = aggregate.map(
new Mapper<Iterable<Object>, Result>() {
public Result apply(Iterable<Object> coll) {
final Iterator<Object> it = coll.iterator();
final String x = (String) it.next();
final String s = (String) it.next();
return new Result(x, s);
}
}, system.dispatcher());
pipe(transformed, system.dispatcher()).to(actorC);
//#ask-pipe
expectMsgEquals(new Result("request", "another request"));
}
};
}
static
//#props-factory
public class DemoActor extends UntypedActor {
/**
* Create Props for an actor of this type.
* @param magicNumber The magic number to be passed to this actor’s constructor.
* @return a Props for creating this actor, which can then be further configured
* (e.g. calling `.withDispatcher()` on it)
*/
public static Props props(final int magicNumber) {
return Props.create(new Creator<DemoActor>() {
private static final long serialVersionUID = 1L;
@Override
public DemoActor create() throws Exception {
return new DemoActor(magicNumber);
}
});
}
final int magicNumber;
public DemoActor(int magicNumber) {
this.magicNumber = magicNumber;
}
@Override
public void onReceive(Object msg) {
// some behavior here
}
}
//#props-factory
@Test
public void demoActor() {
//#props-factory
system.actorOf(DemoActor.props(42), "demo");
//#props-factory
}
static
//#messages-in-companion
public class DemoMessagesActor extends UntypedActor {
static public class Greeting {
private final String from;
public Greeting(String from) {
this.from = from;
}
public String getGreeter() {
return from;
}
}
public void onReceive(Object message) throws Exception {
if (message instanceof Greeting) {
getSender().tell("Hello " + ((Greeting) message).getGreeter(), getSelf());
} else
unhandled(message);
}
}
//#messages-in-companion
public static class MyActor extends UntypedActor {
final String s;
public MyActor(String s) {
this.s = s;
}
public void onReceive(Object message) {
getSender().tell(s, getSelf());
}
/*
* This section must be kept in sync with the actual Actor trait.
*
* BOYSCOUT RULE: whenever you read this, verify that!
*/
//#lifecycle-callbacks
public void preStart() {
}
public void preRestart(Throwable reason, scala.Option<Object> message) {
for (ActorRef each : getContext().getChildren()) {
getContext().unwatch(each);
getContext().stop(each);
}
postStop();
}
public void postRestart(Throwable reason) {
preStart();
}
public void postStop() {
}
//#lifecycle-callbacks
}
public class MyAskActor extends UntypedActor {
public void onReceive(Object message) throws Exception {
//#reply-exception
try {
String result = operation();
getSender().tell(result, getSelf());
} catch (Exception e) {
getSender().tell(new akka.actor.Status.Failure(e), getSelf());
throw e;
}
//#reply-exception
}
private String operation() {
return "Hi";
}
}
static
//#gracefulStop-actor
public class Manager extends UntypedActor {
public static final String SHUTDOWN = "shutdown";
ActorRef worker = getContext().watch(getContext().actorOf(
Props.create(Cruncher.class), "worker"));
public void onReceive(Object message) {
if (message.equals("job")) {
worker.tell("crunch", getSelf());
} else if (message.equals(SHUTDOWN)) {
worker.tell(PoisonPill.getInstance(), getSelf());
getContext().become(shuttingDown);
}
}
Procedure<Object> shuttingDown = new Procedure<Object>() {
@Override
public void apply(Object message) {
if (message.equals("job")) {
getSender().tell("service unavailable, shutting down", getSelf());
} else if (message instanceof Terminated) {
getContext().stop(getSelf());
}
}
};
}
//#gracefulStop-actor
static class Cruncher extends UntypedActor {
public void onReceive(Object message) {
// crunch...
}
}
static
//#hot-swap-actor
public class HotSwapActor extends UntypedActor {
Procedure<Object> angry = new Procedure<Object>() {
@Override
public void apply(Object message) {
if (message.equals("bar")) {
getSender().tell("I am already angry?", getSelf());
} else if (message.equals("foo")) {
getContext().become(happy);
}
}
};
Procedure<Object> happy = new Procedure<Object>() {
@Override
public void apply(Object message) {
if (message.equals("bar")) {
getSender().tell("I am already happy :-)", getSelf());
} else if (message.equals("foo")) {
getContext().become(angry);
}
}
};
public void onReceive(Object message) {
if (message.equals("bar")) {
getContext().become(angry);
} else if (message.equals("foo")) {
getContext().become(happy);
} else {
unhandled(message);
}
}
}
//#hot-swap-actor
static
//#stash
public class ActorWithProtocol extends UntypedActorWithStash {
public void onReceive(Object msg) {
if (msg.equals("open")) {
unstashAll();
getContext().become(new Procedure<Object>() {
public void apply(Object msg) throws Exception {
if (msg.equals("write")) {
// do writing...
} else if (msg.equals("close")) {
unstashAll();
getContext().unbecome();
} else {
stash();
}
}
}, false); // add behavior on top instead of replacing
} else {
stash();
}
}
}
//#stash
static
//#watch
public class WatchActor extends UntypedActor {
final ActorRef child = this.getContext().actorOf(Props.empty(), "child");
{
this.getContext().watch(child); // <-- the only call needed for registration
}
ActorRef lastSender = getContext().system().deadLetters();
@Override
public void onReceive(Object message) {
if (message.equals("kill")) {
getContext().stop(child);
lastSender = getSender();
} else if (message instanceof Terminated) {
final Terminated t = (Terminated) message;
if (t.getActor() == child) {
lastSender.tell("finished", getSelf());
}
} else {
unhandled(message);
}
}
}
//#watch
static
//#identify
public class Follower extends UntypedActor {
final String identifyId = "1";
{
ActorSelection selection =
getContext().actorSelection("/user/another");
selection.tell(new Identify(identifyId), getSelf());
}
ActorRef another;
//#test-omitted
final ActorRef probe;
public Follower(ActorRef probe) {
this.probe = probe;
}
//#test-omitted
@Override
public void onReceive(Object message) {
if (message instanceof ActorIdentity) {
ActorIdentity identity = (ActorIdentity) message;
if (identity.correlationId().equals(identifyId)) {
ActorRef ref = identity.getRef();
if (ref == null)
getContext().stop(getSelf());
else {
another = ref;
getContext().watch(another);
//#test-omitted
probe.tell(ref, getSelf());
//#test-omitted
}
}
} else if (message instanceof Terminated) {
final Terminated t = (Terminated) message;
if (t.getActor().equals(another)) {
getContext().stop(getSelf());
}
} else {
unhandled(message);
}
}
}
//#identify
}

56
akka-docs/rst/java/code/docs/actor/UntypedActorSwapper.java
View file

@ -1,56 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor;
import static docs.actor.UntypedActorSwapper.Swap.SWAP;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.ActorSystem;
import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.japi.Procedure;
//#swapper
public class UntypedActorSwapper {
public static class Swap {
public static Swap SWAP = new Swap();
private Swap() {
}
}
public static class Swapper extends UntypedActor {
LoggingAdapter log = Logging.getLogger(getContext().system(), this);
public void onReceive(Object message) {
if (message == SWAP) {
log.info("Hi");
getContext().become(new Procedure<Object>() {
@Override
public void apply(Object message) {
log.info("Ho");
getContext().unbecome(); // resets the latest 'become'
}
}, false); // this signals stacking of the new behavior
} else {
unhandled(message);
}
}
}
public static void main(String... args) {
ActorSystem system = ActorSystem.create("MySystem");
ActorRef swap = system.actorOf(Props.create(Swapper.class));
swap.tell(SWAP, ActorRef.noSender()); // logs Hi
swap.tell(SWAP, ActorRef.noSender()); // logs Ho
swap.tell(SWAP, ActorRef.noSender()); // logs Hi
swap.tell(SWAP, ActorRef.noSender()); // logs Ho
swap.tell(SWAP, ActorRef.noSender()); // logs Hi
swap.tell(SWAP, ActorRef.noSender()); // logs Ho
}
}
//#swapper

136
akka-docs/rst/java/code/docs/actor/fsm/Buncher.java
View file

@ -1,136 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor.fsm;
//#simple-imports
import akka.actor.AbstractFSM;
import akka.actor.ActorRef;
import akka.japi.pf.UnitMatch;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import scala.concurrent.duration.Duration;
//#simple-imports
import static docs.actor.fsm.Buncher.Data;
import static docs.actor.fsm.Buncher.State.*;
import static docs.actor.fsm.Buncher.State;
import static docs.actor.fsm.Buncher.Uninitialized.*;
import static docs.actor.fsm.Events.*;
//#simple-fsm
public class Buncher extends AbstractFSM<State, Data> {
{
//#fsm-body
startWith(Idle, Uninitialized);
//#when-syntax
when(Idle,
matchEvent(SetTarget.class, Uninitialized.class,
(setTarget, uninitialized) ->
stay().using(new Todo(setTarget.getRef(), new LinkedList<>()))));
//#when-syntax
//#transition-elided
onTransition(
matchState(Active, Idle, () -> {
// reuse this matcher
final UnitMatch<Data> m = UnitMatch.create(
matchData(Todo.class,
todo -> todo.getTarget().tell(new Batch(todo.getQueue()), self())));
m.match(stateData());
}).
state(Idle, Active, () -> {/* Do something here */}));
//#transition-elided
when(Active, Duration.create(1, "second"),
matchEvent(Arrays.asList(Flush.class, StateTimeout()), Todo.class,
(event, todo) -> goTo(Idle).using(todo.copy(new LinkedList<>()))));
//#unhandled-elided
whenUnhandled(
matchEvent(Queue.class, Todo.class,
(queue, todo) -> goTo(Active).using(todo.addElement(queue.getObj()))).
anyEvent((event, state) -> {
log().warning("received unhandled request {} in state {}/{}",
event, stateName(), state);
return stay();
}));
//#unhandled-elided
initialize();
//#fsm-body
}
//#simple-fsm
static
//#simple-state
// states
enum State {
Idle, Active
}
//#simple-state
static
//#simple-state
// state data
interface Data {
}
//#simple-state
static
//#simple-state
enum Uninitialized implements Data {
Uninitialized
}
//#simple-state
static
//#simple-state
final class Todo implements Data {
private final ActorRef target;
private final List<Object> queue;
public Todo(ActorRef target, List<Object> queue) {
this.target = target;
this.queue = queue;
}
public ActorRef getTarget() {
return target;
}
public List<Object> getQueue() {
return queue;
}
//#boilerplate
@Override
public String toString() {
return "Todo{" +
"target=" + target +
", queue=" + queue +
'}';
}
public Todo addElement(Object element) {
List<Object> nQueue = new LinkedList<>(queue);
nQueue.add(element);
return new Todo(this.target, nQueue);
}
public Todo copy(List<Object> queue) {
return new Todo(this.target, queue);
}
public Todo copy(ActorRef target) {
return new Todo(target, this.queue);
}
//#boilerplate
}
//#simple-state
//#simple-fsm
}
//#simple-fsm

79
akka-docs/rst/java/code/docs/actor/fsm/BuncherTest.java
View file

@ -1,79 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor.fsm;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.testkit.JavaTestKit;
import docs.AbstractJavaTest;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.util.LinkedList;
import docs.actor.fsm.*;
import static docs.actor.fsm.Events.Batch;
import static docs.actor.fsm.Events.Queue;
import static docs.actor.fsm.Events.SetTarget;
import static docs.actor.fsm.Events.Flush.Flush;
//#test-code
public class BuncherTest extends AbstractJavaTest {
static ActorSystem system;
@BeforeClass
public static void setup() {
system = ActorSystem.create("BuncherTest");
}
@AfterClass
public static void tearDown() {
JavaTestKit.shutdownActorSystem(system);
system = null;
}
@Test
public void testBuncherActorBatchesCorrectly() {
new JavaTestKit(system) {{
final ActorRef buncher =
system.actorOf(Props.create(Buncher.class));
final ActorRef probe = getRef();
buncher.tell(new SetTarget(probe), probe);
buncher.tell(new Queue(42), probe);
buncher.tell(new Queue(43), probe);
LinkedList<Object> list1 = new LinkedList<>();
list1.add(42);
list1.add(43);
expectMsgEquals(new Batch(list1));
buncher.tell(new Queue(44), probe);
buncher.tell(Flush, probe);
buncher.tell(new Queue(45), probe);
LinkedList<Object> list2 = new LinkedList<>();
list2.add(44);
expectMsgEquals(new Batch(list2));
LinkedList<Object> list3 = new LinkedList<>();
list3.add(45);
expectMsgEquals(new Batch(list3));
system.stop(buncher);
}};
}
@Test
public void testBuncherActorDoesntBatchUninitialized() {
new JavaTestKit(system) {{
final ActorRef buncher =
system.actorOf(Props.create(Buncher.class));
final ActorRef probe = getRef();
buncher.tell(new Queue(42), probe);
expectNoMsg();
system.stop(buncher);
}};
}
}
//#test-code

108
akka-docs/rst/java/code/docs/actor/fsm/Events.java
View file

@ -1,108 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor.fsm;
import akka.actor.ActorRef;
import java.util.List;
public class Events {
static
//#simple-events
public final class SetTarget {
private final ActorRef ref;
public SetTarget(ActorRef ref) {
this.ref = ref;
}
public ActorRef getRef() {
return ref;
}
//#boilerplate
@Override
public String toString() {
return "SetTarget{" +
"ref=" + ref +
'}';
}
//#boilerplate
}
//#simple-events
static
//#simple-events
public final class Queue {
private final Object obj;
public Queue(Object obj) {
this.obj = obj;
}
public Object getObj() {
return obj;
}
//#boilerplate
@Override
public String toString() {
return "Queue{" +
"obj=" + obj +
'}';
}
//#boilerplate
}
//#simple-events
static
//#simple-events
public final class Batch {
private final List<Object> list;
public Batch(List<Object> list) {
this.list = list;
}
public List<Object> getList() {
return list;
}
//#boilerplate
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Batch batch = (Batch) o;
return list.equals(batch.list);
}
@Override
public int hashCode() {
return list.hashCode();
}
@Override
public String toString() {
final StringBuilder builder = new StringBuilder();
builder.append( "Batch{list=");
list.stream().forEachOrdered(e -> { builder.append(e); builder.append(","); });
int len = builder.length();
builder.replace(len, len, "}");
return builder.toString();
}
//#boilerplate
}
//#simple-events
static
//#simple-events
public enum Flush {
Flush
}
//#simple-events
}

180
akka-docs/rst/java/code/docs/actor/fsm/FSMDocTest.java
View file

@ -1,180 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package docs.actor.fsm;
import akka.actor.*;
import akka.testkit.JavaTestKit;
import docs.AbstractJavaTest;
import org.hamcrest.CoreMatchers;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import scala.concurrent.duration.Duration;
import static org.junit.Assert.*;
import static docs.actor.fsm.FSMDocTest.StateType.*;
import static docs.actor.fsm.FSMDocTest.Messages.*;
import static java.util.concurrent.TimeUnit.*;
public class FSMDocTest extends AbstractJavaTest {
static ActorSystem system;
@BeforeClass
public static void setup() {
system = ActorSystem.create("FSMDocTest");
}
@AfterClass
public static void tearDown() {
JavaTestKit.shutdownActorSystem(system);
system = null;
}
public static enum StateType {
SomeState,
Processing,
Idle,
Active,
Error
}
public static enum Messages {
WillDo,
Tick
}
public static enum Data {
Foo,
Bar
};
public static interface X {};
public static class DummyFSM extends AbstractFSM<StateType, Integer> {
Integer newData = 42;
//#alt-transition-syntax
public void handler(StateType from, StateType to) {
// handle transition here
}
//#alt-transition-syntax
{
//#modifier-syntax
when(SomeState, matchAnyEvent((msg, data) -> {
return goTo(Processing).using(newData).
forMax(Duration.create(5, SECONDS)).replying(WillDo);
}));
//#modifier-syntax
//#NullFunction
when(SomeState, AbstractFSM.NullFunction());
//#NullFunction
//#transition-syntax
onTransition(
matchState(Active, Idle, () -> setTimer("timeout",
Tick, Duration.create(1, SECONDS), true)).
state(Active, null, () -> cancelTimer("timeout")).
state(null, Idle, (f, t) -> log().info("entering Idle from " + f)));
//#transition-syntax
//#alt-transition-syntax
onTransition(this::handler);
//#alt-transition-syntax
//#stop-syntax
when(Error, matchEventEquals("stop", (event, data) -> {
// do cleanup ...
return stop();
}));
//#stop-syntax
//#termination-syntax
onTermination(
matchStop(Normal(),
(state, data) -> {/* Do something here */}).
stop(Shutdown(),
(state, data) -> {/* Do something here */}).
stop(Failure.class,
(reason, state, data) -> {/* Do something here */}));
//#termination-syntax
//#unhandled-syntax
whenUnhandled(
matchEvent(X.class, (x, data) -> {
log().info("Received unhandled event: " + x);
return stay();
}).
anyEvent((event, data) -> {
log().warning("Received unknown event: " + event);
return goTo(Error);
}));
}
//#unhandled-syntax
}
static
//#logging-fsm
public class MyFSM extends AbstractLoggingFSM<StateType, Data> {
//#body-elided
//#logging-fsm
ActorRef target = null;
//#logging-fsm
@Override
public int logDepth() { return 12; }
{
onTermination(
matchStop(Failure.class, (reason, state, data) -> {
String lastEvents = getLog().mkString("\n\t");
log().warning("Failure in state " + state + " with data " + data + "\n" +
"Events leading up to this point:\n\t" + lastEvents);
//#logging-fsm
target.tell(reason.cause(), self());
target.tell(state, self());
target.tell(data, self());
target.tell(lastEvents, self());
//#logging-fsm
})
);
//...
//#logging-fsm
startWith(SomeState, Data.Foo);
when(SomeState, matchEvent(ActorRef.class, Data.class, (ref, data) -> {
target = ref;
target.tell("going active", self());
return goTo(Active);
}));
when(Active, matchEventEquals("stop", (event, data) -> {
target.tell("stopping", self());
return stop(new Failure("This is not the error you're looking for"));
}));
initialize();
//#logging-fsm
}
//#body-elided
}
//#logging-fsm
@Test
public void testLoggingFSM()
{
new JavaTestKit(system) {{
final ActorRef logger =
system.actorOf(Props.create(MyFSM.class));
final ActorRef probe = getRef();
logger.tell(probe, probe);
expectMsgEquals("going active");
logger.tell("stop", probe);
expectMsgEquals("stopping");
expectMsgEquals("This is not the error you're looking for");
expectMsgEquals(Active);
expectMsgEquals(Data.Foo);
String msg = expectMsgClass(String.class);
assertThat(msg, CoreMatchers.startsWith("LogEntry(SomeState,Foo,Actor[akka://FSMDocTest/system/"));
}};
}
}

489
akka-docs/rst/java/code/docs/actor/japi/FaultHandlingDocSample.java
View file

@ -1,489 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.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.japi.Function;
import scala.concurrent.duration.Duration;
import akka.util.Timeout;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import static akka.japi.Util.classTag;
import static akka.actor.SupervisorStrategy.restart;
import static akka.actor.SupervisorStrategy.stop;
import static akka.actor.SupervisorStrategy.escalate;
import akka.actor.SupervisorStrategy.Directive;
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.lifecycle = on");
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 UntypedActor {
final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
@Override
public void preStart() {
// If we don't get any progress within 15 seconds then the service
// is unavailable
getContext().setReceiveTimeout(Duration.create("15 seconds"));
}
public void onReceive(Object msg) {
log.debug("received message {}", msg);
if (msg instanceof Progress) {
Progress progress = (Progress) msg;
log.info("Current progress: {} %", progress.percent);
if (progress.percent >= 100.0) {
log.info("That's all, shutting down");
getContext().system().terminate();
}
} else if (msg == ReceiveTimeout.getInstance()) {
// No progress within 15 seconds, ServiceUnavailable
log.error("Shutting down due to unavailable service");
getContext().system().terminate();
} else {
unhandled(msg);
}
}
}
//#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 UntypedActor {
final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
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 = getContext().actorOf(
Props.create(CounterService.class), "counter");
final int totalCount = 51;
// Stop the CounterService child if it throws ServiceUnavailable
private static SupervisorStrategy strategy = new OneForOneStrategy(-1,
Duration.Inf(), new Function<Throwable, Directive>() {
@Override
public Directive apply(Throwable t) {
if (t instanceof ServiceUnavailable) {
return stop();
} else {
return escalate();
}
}
});
@Override
public SupervisorStrategy supervisorStrategy() {
return strategy;
}
public void onReceive(Object msg) {
log.debug("received message {}", msg);
if (msg.equals(Start) && progressListener == null) {
progressListener = getSender();
getContext().system().scheduler().schedule(
Duration.Zero(), Duration.create(1, "second"), getSelf(), Do,
getContext().dispatcher(), null
);
} else if (msg.equals(Do)) {
counterService.tell(new Increment(1), getSelf());
counterService.tell(new Increment(1), getSelf());
counterService.tell(new Increment(1), getSelf());
// 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);
}
}, getContext().dispatcher()), getContext().dispatcher())
.to(progressListener);
} else {
unhandled(msg);
}
}
}
//#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 UntypedActor {
// 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 LoggingAdapter log = Logging.getLogger(getContext().system(), this);
final String key = getSelf().path().name();
ActorRef storage;
ActorRef counter;
final List<SenderMsgPair> backlog = new ArrayList<SenderMsgPair>();
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 SupervisorStrategy strategy = new OneForOneStrategy(3,
Duration.create("5 seconds"), new Function<Throwable, Directive>() {
@Override
public Directive apply(Throwable t) {
if (t instanceof StorageException) {
return restart();
} else {
return escalate();
}
}
});
@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 = getContext().watch(getContext().actorOf(
Props.create(Storage.class), "storage"));
// Tell the counter, if any, to use the new storage
if (counter != null)
counter.tell(new UseStorage(storage), getSelf());
// We need the initial value to be able to operate
storage.tell(new Get(key), getSelf());
}
@Override
public void onReceive(Object msg) {
log.debug("received message {}", msg);
if (msg instanceof Entry && ((Entry) msg).key.equals(key) &&
counter == null) {
// Reply from Storage of the initial value, now we can create the Counter
final long value = ((Entry) msg).value;
counter = getContext().actorOf(Props.create(Counter.class, key, value));
// Tell the counter to use current storage
counter.tell(new UseStorage(storage), getSelf());
// and send the buffered backlog to the counter
for (SenderMsgPair each : backlog) {
counter.tell(each.msg, each.sender);
}
backlog.clear();
} else if (msg instanceof Increment) {
forwardOrPlaceInBacklog(msg);
} else if (msg.equals(GetCurrentCount)) {
forwardOrPlaceInBacklog(msg);
} else if (msg instanceof Terminated) {
// 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), getSelf());
// Try to re-establish storage after while
getContext().system().scheduler().scheduleOnce(
Duration.create(10, "seconds"), getSelf(), Reconnect,
getContext().dispatcher(), null);
} else if (msg.equals(Reconnect)) {
// Re-establish storage after the scheduled delay
initStorage();
} else {
unhandled(msg);
}
}
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(getSender(), msg));
} else {
counter.forward(msg, getContext());
}
}
}
//#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 UntypedActor {
final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
final String key;
long count;
ActorRef storage;
public Counter(String key, long initialValue) {
this.key = key;
this.count = initialValue;
}
@Override
public void onReceive(Object msg) {
log.debug("received message {}", msg);
if (msg instanceof UseStorage) {
storage = ((UseStorage) msg).storage;
storeCount();
} else if (msg instanceof Increment) {
count += ((Increment) msg).n;
storeCount();
} else if (msg.equals(GetCurrentCount)) {
getSender().tell(new CurrentCount(key, count), getSelf());
} else {
unhandled(msg);
}
}
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)), getSelf());
}
}
}
//#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 UntypedActor {
final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
final DummyDB db = DummyDB.instance;
@Override
public void onReceive(Object msg) {
log.debug("received message {}", msg);
if (msg instanceof Store) {
Store store = (Store) msg;
db.save(store.entry.key, store.entry.value);
} else if (msg instanceof Get) {
Get get = (Get) msg;
Long value = db.load(get.key);
getSender().tell(new Entry(get.key, value == null ?
Long.valueOf(0L) : value), getSelf());
} else {
unhandled(msg);
}
}
}
//#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

470
akka-docs/rst/java/code/docs/actor/japi/FaultHandlingDocSampleJava8.java
View file

@ -1,470 +0,0 @@
/**
* Copyright (C) 2009-2017 Lightbend Inc. <http://www.lightbend.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 FaultHandlingDocSampleJava8 {
/**
* 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"));
}
public Listener() {
receive(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;
}
public Worker() {
receive(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());
}
public CounterService() {
receive(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;
receive(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;
public Storage() {
receive(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