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

ActorContext as constructor parameter in AbstractBehavior, #27689 (#27806)

Browse source 
* ActorContext as constructor parameter in AbstractBehavior, #27689

* additional test

* additional doc clarification

* another rebase
This commit is contained in:
Patrik Nordwall 2019-09-27 11:17:37 +02:00 • committed by Arnout Engelen
parent 0719de035b
commit 91db18b564
66 changed files with 442 additions and 323 deletions
Download patch file Download diff file Expand all files Collapse all files

14
akka-actor-testkit-typed/src/test/java/jdocs/akka/actor/testkit/typed/javadsl/SyncTestingExampleTest.java
View file

@ -80,10 +80,8 @@ public class SyncTestingExampleTest extends JUnitSuite {
return Behaviors.setup(Hello::new);
}
private final ActorContext<Command> context;
private Hello(ActorContext<Command> context) {
this.context = context;
super(context);
}
@Override
@ -99,23 +97,23 @@ public class SyncTestingExampleTest extends JUnitSuite {
}
private Behavior<Command> onCreateAChild(CreateAChild message) {
context.spawn(Child.create(), message.childName);
getContext().spawn(Child.create(), message.childName);
return Behaviors.same();
}
private Behavior<Command> onCreateAnonymousChild(CreateAnAnonymousChild message) {
context.spawnAnonymous(Child.create());
getContext().spawnAnonymous(Child.create());
return Behaviors.same();
}
private Behavior<Command> onSayHelloToChild(SayHelloToChild message) {
ActorRef<String> child = context.spawn(Child.create(), message.childName);
ActorRef<String> child = getContext().spawn(Child.create(), message.childName);
child.tell("hello");
return Behaviors.same();
}
private Behavior<Command> onSayHelloToAnonymousChild(SayHelloToAnonymousChild message) {
ActorRef<String> child = context.spawnAnonymous(Child.create());
ActorRef<String> child = getContext().spawnAnonymous(Child.create());
child.tell("hello stranger");
return Behaviors.same();
}
@ -126,7 +124,7 @@ public class SyncTestingExampleTest extends JUnitSuite {
}
private Behavior<Command> onLogAndSayHello(LogAndSayHello message) {
context.getLog().info("Saying hello to {}", message.who.path().name());
getContext().getLog().info("Saying hello to {}", message.who.path().name());
message.who.tell("hello");
return Behaviors.same();
}

24
akka-actor-typed-tests/src/test/java/akka/actor/typed/javadsl/ReceiveBuilderTest.java
View file

@ -31,7 +31,7 @@ public class ReceiveBuilderTest extends JUnitSuite {
Behavior<BehaviorBuilderTest.CounterMessage> mutable =
Behaviors.setup(
context ->
new AbstractBehavior<BehaviorBuilderTest.CounterMessage>() {
new AbstractBehavior<BehaviorBuilderTest.CounterMessage>(context) {
int currentValue = 0;
private Behavior<BehaviorBuilderTest.CounterMessage> receiveIncrease(
@ -56,28 +56,6 @@ public class ReceiveBuilderTest extends JUnitSuite {
});
}
private static class MyAbstractBehavior
extends AbstractBehavior<BehaviorBuilderTest.CounterMessage> {
private int value;
public MyAbstractBehavior(int initialValue) {
super();
this.value = initialValue;
}
@Override
public Receive<BehaviorBuilderTest.CounterMessage> createReceive() {
assertEquals(42, value);
return newReceiveBuilder().build();
}
}
@Test
public void testInitializationOrder() throws Exception {
MyAbstractBehavior mutable = new MyAbstractBehavior(42);
assertEquals(Behaviors.unhandled(), mutable.receive(null, new BehaviorBuilderTest.Increase()));
}
@Test
public void caseSelectedInOrderAdded() {
final TestProbe<Object> probe = testKit.createTestProbe();

1
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/Aggregator.java
View file

@ -66,6 +66,7 @@ public class Aggregator<Reply, Aggregate> extends AbstractBehavior<Aggregator.Co
ActorRef<Aggregate> replyTo,
Function<List<Reply>, Aggregate> aggregateReplies,
Duration timeout) {
super(context);
this.expectedReplies = expectedReplies;
this.replyTo = replyTo;
this.aggregateReplies = aggregateReplies;

8
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/AggregatorTest.java
View file

@ -152,13 +152,11 @@ public class AggregatorTest extends JUnitSuite {
return Behaviors.setup(context -> new HotelCustomer(context, hotel1, hotel2));
}
private final ActorContext<Command> context;
public HotelCustomer(
ActorContext<Command> context,
ActorRef<Hotel1.RequestQuote> hotel1,
ActorRef<Hotel2.RequestPrice> hotel2) {
this.context = context;
super(context);
Consumer<ActorRef<Object>> sendRequests =
replyTo -> {
@ -209,8 +207,8 @@ public class AggregatorTest extends JUnitSuite {
}
private Behavior<Command> onAggregatedQuotes(AggregatedQuotes aggregated) {
if (aggregated.quotes.isEmpty()) context.getLog().info("Best Quote N/A");
else context.getLog().info("Best {}", aggregated.quotes.get(0));
if (aggregated.quotes.isEmpty()) getContext().getLog().info("Best Quote N/A");
else getContext().getLog().info("Best {}", aggregated.quotes.get(0));
return this;
}
}

12
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/BubblingSample.java
View file

@ -45,7 +45,11 @@ public class BubblingSample {
public static class Worker extends AbstractBehavior<Protocol.Command> {
public static Behavior<Protocol.Command> create() {
return Behaviors.setup(context -> new Worker());
return Behaviors.setup(Worker::new);
}
private Worker(ActorContext<Protocol.Command> context) {
super(context);
}
@Override
@ -72,11 +76,10 @@ public class BubblingSample {
return Behaviors.setup(MiddleManagement::new);
}
private final ActorContext<Protocol.Command> context;
private final ActorRef<Protocol.Command> child;
private MiddleManagement(ActorContext<Protocol.Command> context) {
this.context = context;
super(context);
context.getLog().info("Middle management starting up");
// default supervision of child, meaning that it will stop on failure
@ -109,11 +112,10 @@ public class BubblingSample {
.onFailure(DeathPactException.class, SupervisorStrategy.restart());
}
private final ActorContext<Protocol.Command> context;
private final ActorRef<Protocol.Command> middleManagement;
private Boss(ActorContext<Protocol.Command> context) {
this.context = context;
super(context);
context.getLog().info("Boss starting up");
// default supervision of child, meaning that it will stop on failure
middleManagement = context.spawn(MiddleManagement.create(), "middle-management");

19
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/GracefulStopDocTest.java
View file

@ -42,10 +42,8 @@ interface GracefulStopDocTest {
return Behaviors.setup(MasterControlProgram::new);
}
private final ActorContext<Command> context;
public MasterControlProgram(ActorContext<Command> context) {
this.context = context;
super(context);
}
@Override
@ -58,21 +56,21 @@ interface GracefulStopDocTest {
}
private Behavior<Command> onSpawnJob(SpawnJob message) {
context.getSystem().log().info("Spawning job {}!", message.name);
context.spawn(Job.create(message.name), message.name);
getContext().getSystem().log().info("Spawning job {}!", message.name);
getContext().spawn(Job.create(message.name), message.name);
return this;
}
private Behavior<Command> onGracefulShutdown() {
context.getSystem().log().info("Initiating graceful shutdown...");
getContext().getSystem().log().info("Initiating graceful shutdown...");
// perform graceful stop, executing cleanup before final system termination
// behavior executing cleanup is passed as a parameter to Actor.stopped
return Behaviors.stopped(() -> context.getSystem().log().info("Cleanup!"));
return Behaviors.stopped(() -> getContext().getSystem().log().info("Cleanup!"));
}
private Behavior<Command> onPostStop() {
context.getSystem().log().info("Master Control Program stopped");
getContext().getSystem().log().info("Master Control Program stopped");
return this;
}
}
@ -106,11 +104,10 @@ interface GracefulStopDocTest {
return Behaviors.setup(context -> new Job(context, name));
}
private final ActorContext<Command> context;
private final String name;
public Job(ActorContext<Command> context, String name) {
this.context = context;
super(context);
this.name = name;
}
@ -120,7 +117,7 @@ interface GracefulStopDocTest {
}
private Behavior<Command> onPostStop() {
context.getSystem().log().info("Worker {} stopped", name);
getContext().getSystem().log().info("Worker {} stopped", name);
return this;
}
}

44
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/InteractionPatternsTest.java
View file

@ -170,7 +170,6 @@ public class InteractionPatternsTest extends JUnitSuite {
}
public static class Translator extends AbstractBehavior<Command> {
private final ActorContext<Command> context;
private final ActorRef<Backend.Request> backend;
private final ActorRef<Backend.Response> backendResponseAdapter;
@ -178,7 +177,7 @@ public class InteractionPatternsTest extends JUnitSuite {
private Map<Integer, ActorRef<URI>> inProgress = new HashMap<>();
public Translator(ActorContext<Command> context, ActorRef<Backend.Request> backend) {
this.context = context;
super(context);
this.backend = backend;
this.backendResponseAdapter =
context.messageAdapter(Backend.Response.class, WrappedBackendResponse::new);
@ -204,13 +203,13 @@ public class InteractionPatternsTest extends JUnitSuite {
Backend.Response response = wrapped.response;
if (response instanceof Backend.JobStarted) {
Backend.JobStarted rsp = (Backend.JobStarted) response;
context.getLog().info("Started {}", rsp.taskId);
getContext().getLog().info("Started {}", rsp.taskId);
} else if (response instanceof Backend.JobProgress) {
Backend.JobProgress rsp = (Backend.JobProgress) response;
context.getLog().info("Progress {}", rsp.taskId);
getContext().getLog().info("Progress {}", rsp.taskId);
} else if (response instanceof Backend.JobCompleted) {
Backend.JobCompleted rsp = (Backend.JobCompleted) response;
context.getLog().info("Completed {}", rsp.taskId);
getContext().getLog().info("Completed {}", rsp.taskId);
inProgress.get(rsp.taskId).tell(rsp.result);
inProgress.remove(rsp.taskId);
} else {
@ -292,13 +291,15 @@ public class InteractionPatternsTest extends JUnitSuite {
private Behavior<Command> onIdleCommand(Command message) {
timers.startSingleTimer(TIMER_KEY, Timeout.INSTANCE, after);
return new Active(message);
return Behaviors.setup(context -> new Active(context, message));
}
private class Active extends AbstractBehavior<Command> {
private final List<Command> buffer = new ArrayList<>();
Active(Command firstCommand) {
Active(ActorContext<Command> context, Command firstCommand) {
super(context);
buffer.add(firstCommand);
}
@ -332,6 +333,14 @@ public class InteractionPatternsTest extends JUnitSuite {
// #actor-ask
public class Hal extends AbstractBehavior<Hal.Command> {
public Behavior<Hal.Command> create() {
return Behaviors.setup(Hal::new);
}
private Hal(ActorContext<Command> context) {
super(context);
}
public interface Command {}
public static final class OpenThePodBayDoorsPlease implements Command {
@ -380,10 +389,8 @@ public class InteractionPatternsTest extends JUnitSuite {
return Behaviors.setup(context -> new Dave(context, hal));
}
private final ActorContext<Command> context;
private Dave(ActorContext<Command> context, ActorRef<Hal.Command> hal) {
this.context = context;
super(context);
// asking someone requires a timeout, if the timeout hits without response
// the ask is failed with a TimeoutException
@ -435,7 +442,7 @@ public class InteractionPatternsTest extends JUnitSuite {
}
private Behavior<Command> onAdaptedResponse(AdaptedResponse response) {
context.getLog().info("Got response from HAL: {}", response.message);
getContext().getLog().info("Got response from HAL: {}", response.message);
return this;
}
}
@ -535,7 +542,7 @@ public class InteractionPatternsTest extends JUnitSuite {
private Behavior<Command> onLeaveHome(LeaveHome message) {
context.spawn(
new PrepareToLeaveHome(message.who, message.respondTo, keyCabinet, drawer),
PrepareToLeaveHome.create(message.who, message.respondTo, keyCabinet, drawer),
"leaving" + message.who);
return Behaviors.same();
}
@ -548,6 +555,15 @@ public class InteractionPatternsTest extends JUnitSuite {
// per session actor behavior
class PrepareToLeaveHome extends AbstractBehavior<Object> {
static Behavior<Object> create(
String whoIsLeaving,
ActorRef<Home.ReadyToLeaveHome> replyTo,
ActorRef<KeyCabinet.GetKeys> keyCabinet,
ActorRef<Drawer.GetWallet> drawer) {
return Behaviors.setup(
context -> new PrepareToLeaveHome(context, whoIsLeaving, replyTo, keyCabinet, drawer));
}
private final String whoIsLeaving;
private final ActorRef<Home.ReadyToLeaveHome> replyTo;
private final ActorRef<KeyCabinet.GetKeys> keyCabinet;
@ -555,11 +571,13 @@ public class InteractionPatternsTest extends JUnitSuite {
private Optional<Wallet> wallet = Optional.empty();
private Optional<Keys> keys = Optional.empty();
PrepareToLeaveHome(
private PrepareToLeaveHome(
ActorContext<Object> context,
String whoIsLeaving,
ActorRef<Home.ReadyToLeaveHome> replyTo,
ActorRef<KeyCabinet.GetKeys> keyCabinet,
ActorRef<Drawer.GetWallet> drawer) {
super(context);
this.whoIsLeaving = whoIsLeaving;
this.replyTo = replyTo;
this.keyCabinet = keyCabinet;

26
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/IntroTest.java
View file

@ -54,10 +54,8 @@ public interface IntroTest {
return Behaviors.setup(HelloWorld::new);
}
private final ActorContext<Greet> context;
private HelloWorld(ActorContext<Greet> context) {
this.context = context;
super(context);
}
@Override
@ -66,8 +64,8 @@ public interface IntroTest {
}
private Behavior<Greet> onGreet(Greet command) {
context.getLog().info("Hello {}!", command.whom);
command.replyTo.tell(new Greeted(command.whom, context.getSelf()));
getContext().getLog().info("Hello {}!", command.whom);
command.replyTo.tell(new Greeted(command.whom, getContext().getSelf()));
return this;
}
}
@ -80,12 +78,11 @@ public interface IntroTest {
return Behaviors.setup(context -> new HelloWorldBot(context, max));
}
private final ActorContext<HelloWorld.Greeted> context;
private final int max;
private int greetingCounter;
private HelloWorldBot(ActorContext<HelloWorld.Greeted> context, int max) {
this.context = context;
super(context);
this.max = max;
}
@ -96,11 +93,11 @@ public interface IntroTest {
private Behavior<HelloWorld.Greeted> onGreeted(HelloWorld.Greeted message) {
greetingCounter++;
context.getLog().info("Greeting {} for {}", greetingCounter, message.whom);
getContext().getLog().info("Greeting {} for {}", greetingCounter, message.whom);
if (greetingCounter == max) {
return Behaviors.stopped();
} else {
message.from.tell(new HelloWorld.Greet(message.whom, context.getSelf()));
message.from.tell(new HelloWorld.Greet(message.whom, getContext().getSelf()));
return this;
}
}
@ -125,11 +122,10 @@ public interface IntroTest {
return Behaviors.setup(HelloWorldMain::new);
}
private final ActorContext<Start> context;
private final ActorRef<HelloWorld.Greet> greeter;
private HelloWorldMain(ActorContext<Start> context) {
this.context = context;
super(context);
greeter = context.spawn(HelloWorld.create(), "greeter");
}
// #hello-world-main-setup
@ -140,7 +136,8 @@ public interface IntroTest {
}
private Behavior<Start> onStart(Start command) {
ActorRef<HelloWorld.Greeted> replyTo = context.spawn(HelloWorldBot.create(3), command.name);
ActorRef<HelloWorld.Greeted> replyTo =
getContext().spawn(HelloWorldBot.create(3), command.name);
greeter.tell(new HelloWorld.Greet(command.name, replyTo));
return this;
}
@ -177,15 +174,14 @@ public interface IntroTest {
return Behaviors.setup(HelloWorldMain::new);
}
private final ActorContext<Start> context;
private final ActorRef<HelloWorld.Greet> greeter;
private HelloWorldMain(ActorContext<Start> context) {
this.context = context;
super(context);
final String dispatcherPath = "akka.actor.default-blocking-io-dispatcher";
Props greeterProps = DispatcherSelector.fromConfig(dispatcherPath);
greeter = context.spawn(HelloWorld.create(), "greeter", greeterProps);
greeter = getContext().spawn(HelloWorld.create(), "greeter", greeterProps);
}
// createReceive ...

16
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/LoggingDocExamples.java
View file

@ -38,10 +38,8 @@ public interface LoggingDocExamples {
return Behaviors.setup(MyLoggingBehavior::new);
}
private final ActorContext<String> context;
private MyLoggingBehavior(ActorContext<String> context) {
this.context = context;
super(context);
}
@Override
@ -50,7 +48,7 @@ public interface LoggingDocExamples {
}
private Behavior<String> onReceive(String message) {
context.getLog().info("Received message: {}", message);
getContext().getLog().info("Received message: {}", message);
return this;
}
}
@ -68,10 +66,8 @@ public interface LoggingDocExamples {
});
}
private final ActorContext<String> context;
private BackendManager(ActorContext<String> context) {
this.context = context;
super(context);
}
@Override
@ -80,7 +76,7 @@ public interface LoggingDocExamples {
}
private Behavior<String> onReceive(String message) {
context.getLog().debug("Received message: {}", message);
getContext().getLog().debug("Received message: {}", message);
return this;
}
}
@ -122,6 +118,10 @@ public interface LoggingDocExamples {
return Behaviors.empty();
}
public BackendManager2(ActorContext<Command> context) {
super(context);
}
@Override
public Receive<Command> createReceive() {
return newReceiveBuilder().build();

29
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/OOIntroTest.java
View file

@ -108,11 +108,10 @@ public interface OOIntroTest {
}
public static class ChatRoomBehavior extends AbstractBehavior<RoomCommand> {
final ActorContext<RoomCommand> context;
final List<ActorRef<SessionCommand>> sessions = new ArrayList<>();
private ChatRoomBehavior(ActorContext<RoomCommand> context) {
this.context = context;
super(context);
}
@Override
@ -129,9 +128,10 @@ public interface OOIntroTest {
throws UnsupportedEncodingException {
ActorRef<SessionEvent> client = getSession.replyTo;
ActorRef<SessionCommand> ses =
context.spawn(
new SessionBehavior(context.getSelf(), getSession.screenName, client),
URLEncoder.encode(getSession.screenName, StandardCharsets.UTF_8.name()));
getContext()
.spawn(
SessionBehavior.create(getContext().getSelf(), getSession.screenName, client),
URLEncoder.encode(getSession.screenName, StandardCharsets.UTF_8.name()));
// narrow to only expose PostMessage
client.tell(new SessionGranted(ses.narrow()));
sessions.add(ses);
@ -151,8 +151,17 @@ public interface OOIntroTest {
private final String screenName;
private final ActorRef<SessionEvent> client;
SessionBehavior(
public static Behavior<ChatRoom.SessionCommand> create(
ActorRef<RoomCommand> room, String screenName, ActorRef<SessionEvent> client) {
return Behaviors.setup(context -> new SessionBehavior(context, room, screenName, client));
}
private SessionBehavior(
ActorContext<ChatRoom.SessionCommand> context,
ActorRef<RoomCommand> room,
String screenName,
ActorRef<SessionEvent> client) {
super(context);
this.room = room;
this.screenName = screenName;
this.client = client;
@ -187,10 +196,8 @@ public interface OOIntroTest {
return Behaviors.setup(Gabbler::new);
}
private ActorContext<ChatRoom.SessionEvent> context;
private Gabbler(ActorContext<ChatRoom.SessionEvent> context) {
this.context = context;
super(context);
}
@Override
@ -204,7 +211,7 @@ public interface OOIntroTest {
}
private Behavior<ChatRoom.SessionEvent> onSessionDenied(ChatRoom.SessionDenied message) {
context.getLog().info("cannot start chat room session: {}", message.reason);
getContext().getLog().info("cannot start chat room session: {}", message.reason);
return Behaviors.stopped();
}
@ -214,7 +221,7 @@ public interface OOIntroTest {
}
private Behavior<ChatRoom.SessionEvent> onMessagePosted(ChatRoom.MessagePosted message) {
context
getContext()
.getLog()
.info("message has been posted by '{}': {}", message.screenName, message.message);
return Behaviors.stopped();

34
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/StyleGuideDocExamples.java
View file

@ -110,12 +110,10 @@ interface StyleGuideDocExamples {
return Behaviors.setup(Counter::new);
}
private final ActorContext<Counter.Command> context;
private int n;
private Counter(ActorContext<Command> context) {
this.context = context;
super(context);
}
@Override
@ -130,7 +128,7 @@ interface StyleGuideDocExamples {
private Behavior<Command> onIncrement() {
n++;
context.getLog().debug("Incremented counter to [{}]", n);
getContext().getLog().debug("Incremented counter to [{}]", n);
return this;
}
@ -426,13 +424,15 @@ interface StyleGuideDocExamples {
// factory for the initial `Behavior`
public static Behavior<Command> create(int countDownFrom, ActorRef<Done> notifyWhenZero) {
return Behaviors.setup(context -> new CountDown(countDownFrom, notifyWhenZero));
return Behaviors.setup(context -> new CountDown(context, countDownFrom, notifyWhenZero));
}
private final ActorRef<Done> notifyWhenZero;
private int remaining;
private CountDown(int countDownFrom, ActorRef<Done> notifyWhenZero) {
private CountDown(
ActorContext<Command> context, int countDownFrom, ActorRef<Done> notifyWhenZero) {
super(context);
this.remaining = countDownFrom;
this.notifyWhenZero = notifyWhenZero;
}
@ -558,12 +558,11 @@ interface StyleGuideDocExamples {
}
private final String name;
private final ActorContext<Command> context;
private int count;
private Counter(String name, ActorContext<Command> context) {
super(context);
this.name = name;
this.context = context;
}
// #on-message-lambda
@ -589,14 +588,16 @@ interface StyleGuideDocExamples {
// #on-message-lambda
private Behavior<Command> onIncrement() {
count++;
context.getLog().debug("[{}] Incremented counter to [{}]", name, count);
getContext().getLog().debug("[{}] Incremented counter to [{}]", name, count);
return this;
}
// #on-message-lambda
private Behavior<Command> onTick() {
count++;
context.getLog().debug("[{}] Incremented counter by background tick to [{}]", name, count);
getContext()
.getLog()
.debug("[{}] Incremented counter by background tick to [{}]", name, count);
return this;
}
@ -616,14 +617,14 @@ interface StyleGuideDocExamples {
Increment.class,
notUsed -> {
count++;
context.getLog().debug("[{}] Incremented counter to [{}]", name, count);
getContext().getLog().debug("[{}] Incremented counter to [{}]", name, count);
return this;
})
.onMessage(
Tick.class,
notUsed -> {
count++;
context
getContext()
.getLog()
.debug("[{}] Incremented counter by background tick to [{}]", name, count);
return this;
@ -694,12 +695,11 @@ interface StyleGuideDocExamples {
}
private final String name;
private final ActorContext<Message> context;
private int count;
private Counter(String name, ActorContext<Message> context) {
super(context);
this.name = name;
this.context = context;
}
@Override
@ -713,13 +713,15 @@ interface StyleGuideDocExamples {
private Behavior<Message> onIncrement() {
count++;
context.getLog().debug("[{}] Incremented counter to [{}]", name, count);
getContext().getLog().debug("[{}] Incremented counter to [{}]", name, count);
return this;
}
private Behavior<Message> onTick() {
count++;
context.getLog().debug("[{}] Incremented counter by background tick to [{}]", name, count);
getContext()
.getLog()
.debug("[{}] Incremented counter by background tick to [{}]", name, count);
return this;
}

1
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/TailChopping.java
View file

@ -77,6 +77,7 @@ public class TailChopping<Reply> extends AbstractBehavior<TailChopping.Command>
ActorRef<Reply> replyTo,
Duration finalTimeout,
Reply timeoutReply) {
super(context);
this.timers = timers;
this.sendRequest = sendRequest;
this.nextRequestAfter = nextRequestAfter;

5
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/coexistence/TypedWatchingClassicTest.java
View file

@ -39,11 +39,10 @@ public class TypedWatchingClassicTest extends JUnitSuite {
INSTANCE
}
private final akka.actor.typed.javadsl.ActorContext<Command> context;
private final akka.actor.ActorRef second;
private Typed(ActorContext<Command> context, akka.actor.ActorRef second) {
this.context = context;
super(context);
this.second = second;
}
@ -70,7 +69,7 @@ public class TypedWatchingClassicTest extends JUnitSuite {
}
private Behavior<Command> onPong() {
Adapter.stop(context, second);
Adapter.stop(getContext(), second);
return this;
}
}

15
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/fromclassic/TypedSample.java
View file

@ -46,10 +46,8 @@ interface TypedSample {
return Behaviors.setup(HelloWorld::new);
}
private final ActorContext<Greet> context;
private HelloWorld(ActorContext<Greet> context) {
this.context = context;
super(context);
}
@Override
@ -58,8 +56,8 @@ interface TypedSample {
}
private Behavior<Greet> onGreet(Greet command) {
context.getLog().info("Hello {}!", command.whom);
command.replyTo.tell(new Greeted(command.whom, context.getSelf()));
getContext().getLog().info("Hello {}!", command.whom);
command.replyTo.tell(new Greeted(command.whom, getContext().getSelf()));
return this;
}
}
@ -92,11 +90,10 @@ interface TypedSample {
return Behaviors.setup(Parent::new);
}
private final ActorContext<Command> context;
private Map<String, ActorRef<Child.Command>> children = new HashMap<>();
private Parent(ActorContext<Command> context) {
this.context = context;
super(context);
}
@Override
@ -110,8 +107,8 @@ interface TypedSample {
private Behavior<Command> onDelegateToChild(DelegateToChild command) {
ActorRef<Child.Command> ref = children.get(command.name);
if (ref == null) {
ref = context.spawn(Child.create(), command.name);
context.watchWith(ref, new ChildTerminated(command.name));
ref = getContext().spawn(Child.create(), command.name);
getContext().watchWith(ref, new ChildTerminated(command.name));
children.put(command.name, ref);
}
ref.tell(command.message);

2
akka-actor-typed-tests/src/test/scala/akka/actor/typed/BehaviorSpec.scala
View file

@ -441,7 +441,7 @@ class MutableScalaBehaviorSpec extends Messages with Become with Stoppable {
def behv(monitor: ActorRef[Event]): Behavior[Command] =
SBehaviors.setup[Command] { context =>
new SAbstractBehavior[Command] {
new SAbstractBehavior[Command](context) {
private var state: State = StateA
override def onMessage(message: Command): Behavior[Command] = {

58
akka-actor-typed-tests/src/test/scala/akka/actor/typed/SupervisionSpec.scala
View file

@ -22,6 +22,7 @@ import akka.actor.typed.SupervisorStrategy.Resume
import akka.actor.typed.eventstream.EventStream
import akka.actor.typed.scaladsl.Behaviors._
import akka.actor.typed.scaladsl.AbstractBehavior
import akka.actor.typed.scaladsl.ActorContext
import akka.actor.typed.scaladsl.Behaviors
import org.scalatest.Matchers
import org.scalatest.WordSpec
@ -81,7 +82,8 @@ object SupervisionSpec {
Behaviors.same
}
class FailingConstructor(monitor: ActorRef[Event]) extends AbstractBehavior[Command] {
class FailingConstructor(context: ActorContext[Command], monitor: ActorRef[Event])
extends AbstractBehavior[Command](context) {
monitor ! Started
throw new RuntimeException("simulated exc from constructor") with NoStackTrace
@ -263,7 +265,8 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
class FailingConstructorTestSetup(failCount: Int) {
val failCounter = new AtomicInteger(0)
class FailingConstructor(monitor: ActorRef[Event]) extends AbstractBehavior[Command] {
class FailingConstructor(context: ActorContext[Command], monitor: ActorRef[Event])
extends AbstractBehavior[Command](context) {
monitor ! Started
if (failCounter.getAndIncrement() < failCount) {
throw TestException("simulated exc from constructor")
@ -1023,7 +1026,7 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
"fail when exception from AbstractBehavior constructor" in new FailingConstructorTestSetup(failCount = 1) {
val probe = TestProbe[Event]("evt")
val behv = supervise(setup[Command](_ => new FailingConstructor(probe.ref)))
val behv = supervise(setup[Command](ctx => new FailingConstructor(ctx, probe.ref)))
.onFailure[Exception](SupervisorStrategy.restart)
LoggingEventFilter.error[ActorInitializationException].intercept {
@ -1271,6 +1274,55 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
probe.expectMessage(Pong(2))
}
"not allow AbstractBehavior without setup" in {
val contextProbe = createTestProbe[ActorContext[String]]
spawn(Behaviors.setup[String] { context =>
contextProbe.ref ! context
Behaviors.empty
})
val wrongContext = contextProbe.receiveMessage()
intercept[IllegalArgumentException] {
Behaviors
.supervise(new AbstractBehavior[String](wrongContext) {
override def onMessage(msg: String): Behavior[String] = Behaviors.same
})
.onFailure(SupervisorStrategy.restart)
}
intercept[IllegalArgumentException] {
Behaviors.supervise(new AbstractBehavior[String](null) {
override def onMessage(msg: String): Behavior[String] = Behaviors.same
})
}
}
"detect AbstractBehavior with wrong ActorContext" in {
val contextProbe = createTestProbe[ActorContext[String]]
spawn(Behaviors.setup[String] { context =>
contextProbe.ref ! context
Behaviors.empty
})
val wrongContext = contextProbe.receiveMessage()
val wrong = spawn(Behaviors.setup[String](_ =>
new AbstractBehavior[String](wrongContext) {
override def onMessage(msg: String): Behavior[String] = Behaviors.same
}))
val probe = createTestProbe()
LoggingEventFilter
.error[IllegalStateException]
.withMessageContains("created with wrong ActorContext")
.withOccurrences(2) // twice because also logged for PostStop signal
.intercept {
wrong ! "boom"
}
probe.expectTerminated(wrong)
}
}
val allStrategies = Seq(

8
akka-actor-typed-tests/src/test/scala/akka/actor/typed/WatchSpec.scala
View file

@ -27,9 +27,11 @@ object WatchSpec {
case (_, Stop) => Behaviors.stopped
}
val mutableTerminatorBehavior = new AbstractBehavior[Stop.type] {
override def onMessage(message: Stop.type) = message match {
case Stop => Behaviors.stopped
val mutableTerminatorBehavior = Behaviors.setup[Stop.type] { context =>
new AbstractBehavior[Stop.type](context) {
override def onMessage(message: Stop.type) = message match {
case Stop => Behaviors.stopped
}
}
}

2
akka-actor-typed-tests/src/test/scala/akka/actor/typed/scaladsl/ActorLoggingSpec.scala
View file

@ -38,7 +38,7 @@ object WhereTheBehaviorIsDefined {
object BehaviorWhereTheLoggerIsUsed {
def behavior: Behavior[String] = Behaviors.setup(ctx => new BehaviorWhereTheLoggerIsUsed(ctx))
}
class BehaviorWhereTheLoggerIsUsed(context: ActorContext[String]) extends AbstractBehavior[String] {
class BehaviorWhereTheLoggerIsUsed(context: ActorContext[String]) extends AbstractBehavior[String](context) {
context.log.info("Starting up")
override def onMessage(msg: String): Behavior[String] = {
Behaviors.same

3
akka-actor-typed-tests/src/test/scala/akka/actor/typed/scaladsl/StashSpec.scala
View file

@ -131,7 +131,8 @@ object AbstractStashSpec {
}
}
class MutableStash(context: ActorContext[Command], buffer: StashBuffer[Command]) extends AbstractBehavior[Command] {
class MutableStash(context: ActorContext[Command], buffer: StashBuffer[Command])
extends AbstractBehavior[Command](context) {
private var stashing = false
private var processed = Vector.empty[String]

15
akka-actor-typed-tests/src/test/scala/docs/akka/typed/OOIntroSpec.scala
View file

@ -48,7 +48,7 @@ object OOIntroSpec {
def apply(): Behavior[RoomCommand] =
Behaviors.setup(context => new ChatRoomBehavior(context))
class ChatRoomBehavior(context: ActorContext[RoomCommand]) extends AbstractBehavior[RoomCommand] {
class ChatRoomBehavior(context: ActorContext[RoomCommand]) extends AbstractBehavior[RoomCommand](context) {
private var sessions: List[ActorRef[SessionCommand]] = List.empty
override def onMessage(message: RoomCommand): Behavior[RoomCommand] = {
@ -56,7 +56,7 @@ object OOIntroSpec {
case GetSession(screenName, client) =>
// create a child actor for further interaction with the client
val ses = context.spawn(
new SessionBehavior(context.self, screenName, client),
SessionBehavior(context.self, screenName, client),
name = URLEncoder.encode(screenName, StandardCharsets.UTF_8.name))
client ! SessionGranted(ses)
sessions = ses :: sessions
@ -69,11 +69,20 @@ object OOIntroSpec {
}
}
object SessionBehavior {
def apply(
room: ActorRef[PublishSessionMessage],
screenName: String,
client: ActorRef[SessionEvent]): Behavior[SessionCommand] =
Behaviors.setup(ctx => new SessionBehavior(ctx, room, screenName, client))
}
private class SessionBehavior(
context: ActorContext[SessionCommand],
room: ActorRef[PublishSessionMessage],
screenName: String,
client: ActorRef[SessionEvent])
extends AbstractBehavior[SessionCommand] {
extends AbstractBehavior[SessionCommand](context) {
override def onMessage(msg: SessionCommand): Behavior[SessionCommand] = {
msg match {

2
akka-actor-typed-tests/src/test/scala/docs/akka/typed/StyleGuideDocExamples.scala
View file

@ -77,7 +77,7 @@ object StyleGuideDocExamples {
}
}
class Counter(context: ActorContext[Counter.Command]) extends AbstractBehavior[Counter.Command] {
class Counter(context: ActorContext[Counter.Command]) extends AbstractBehavior[Counter.Command](context) {
import Counter._
private var n = 0

2
akka-actor-typed-tests/src/test/scala/docs/akka/typed/fromclassic/TypedSample.scala
View file

@ -24,7 +24,7 @@ object TypedSample {
Behaviors.setup(context => new HelloWorld(context))
}
class HelloWorld(context: ActorContext[HelloWorld.Greet]) extends AbstractBehavior[HelloWorld.Greet] {
class HelloWorld(context: ActorContext[HelloWorld.Greet]) extends AbstractBehavior[HelloWorld.Greet](context) {
import HelloWorld._
override def onMessage(message: Greet): Behavior[Greet] = {

2
akka-actor-typed/src/main/scala/akka/actor/typed/ActorSystem.scala
View file

@ -250,7 +250,7 @@ object ActorSystem {
appConfig,
cl,
executionContext,
Some(PropsAdapter[Any](() => new GuardianStartupBehavior(guardianBehavior), guardianProps)),
Some(PropsAdapter[Any](() => GuardianStartupBehavior(guardianBehavior), guardianProps)),
setup)
system.start()

6
akka-actor-typed/src/main/scala/akka/actor/typed/internal/Supervision.scala
View file

@ -32,6 +32,12 @@ import org.slf4j.event.Level
*/
@InternalApi private[akka] object Supervisor {
def apply[T, Thr <: Throwable: ClassTag](initialBehavior: Behavior[T], strategy: SupervisorStrategy): Behavior[T] = {
if (initialBehavior.isInstanceOf[scaladsl.AbstractBehavior[_]] || initialBehavior
.isInstanceOf[javadsl.AbstractBehavior[_]]) {
throw new IllegalArgumentException(
"The supervised Behavior must not be a AbstractBehavior instance directly," +
"because a different instance should be created when it is restarted. Wrap in Behaviors.setup.")
}
strategy match {
case r: RestartOrBackoff =>

34
akka-actor-typed/src/main/scala/akka/actor/typed/internal/adapter/GuardianStartupBehavior.scala
View file

@ -8,17 +8,9 @@ import akka.actor.typed.Behavior
import akka.actor.typed.BehaviorInterceptor
import akka.actor.typed.Signal
import akka.actor.typed.TypedActorContext
import akka.actor.typed.scaladsl.{ AbstractBehavior, Behaviors, StashOverflowException }
import akka.actor.typed.scaladsl.{ Behaviors, StashOverflowException }
import akka.annotation.InternalApi
/**
* INTERNAL API
*/
@InternalApi
private[akka] object GuardianStartupBehavior {
case object Start
}
/**
* INTERNAL API
*
@ -27,29 +19,29 @@ private[akka] object GuardianStartupBehavior {
* system, and we know that the bootstrap is completed and the actor context can be accessed.
*/
@InternalApi
private[akka] final class GuardianStartupBehavior[T](val guardianBehavior: Behavior[T]) extends AbstractBehavior[Any] {
private[akka] object GuardianStartupBehavior {
case object Start
import GuardianStartupBehavior.Start
private val StashCapacity = 1000
private var tempStash: List[Any] = Nil
def apply[T](guardianBehavior: Behavior[T]): Behavior[Any] =
waitingForStart(guardianBehavior, Vector.empty)
override def onMessage(msg: Any): Behavior[Any] =
msg match {
private def waitingForStart[T](guardianBehavior: Behavior[T], tempStash: Vector[Any]): Behavior[Any] = {
Behaviors.receiveMessage {
case Start =>
// ctx is not available initially so we cannot use it until here
Behaviors.withStash[Any](1000) { stash =>
tempStash.reverse.foreach(stash.stash)
tempStash = null
Behaviors.withStash[Any](StashCapacity) { stash =>
tempStash.foreach(stash.stash)
stash.unstashAll(Behaviors.intercept(() => new GuardianStopInterceptor)(guardianBehavior.unsafeCast[Any]))
}
case other =>
tempStash = other :: tempStash
if (tempStash.size > 1000) {
if (tempStash.size >= StashCapacity) {
throw new StashOverflowException("Guardian Behavior did not receive start and buffer is full.")
}
this
waitingForStart(guardianBehavior, tempStash :+ other)
}
}
}
/**

22
akka-actor-typed/src/main/scala/akka/actor/typed/internal/routing/GroupRouterImpl.scala
View file

@ -43,28 +43,30 @@ private final class InitialGroupRouterImpl[T](
ctx: ActorContext[T],
serviceKey: ServiceKey[T],
routingLogic: RoutingLogic[T])
extends AbstractBehavior[T] {
extends AbstractBehavior[T](ctx) {
// casting trix to avoid having to wrap incoming messages - note that this will cause problems if intercepting
// messages to a router
ctx.system.receptionist ! Receptionist.Subscribe(serviceKey, ctx.self.unsafeUpcast[Any].narrow[Receptionist.Listing])
context.system.receptionist ! Receptionist.Subscribe(
serviceKey,
context.self.unsafeUpcast[Any].narrow[Receptionist.Listing])
private val stash = StashBuffer[T](ctx, capacity = 10000)
private val stash = StashBuffer[T](context, capacity = 10000)
def onMessage(msg: T): Behavior[T] = msg match {
case serviceKey.Listing(update) =>
// we don't need to watch, because receptionist already does that
routingLogic.routeesUpdated(update)
val activeGroupRouter = new GroupRouterImpl[T](ctx, serviceKey, routingLogic, update.isEmpty)
val activeGroupRouter = new GroupRouterImpl[T](context, serviceKey, routingLogic, update.isEmpty)
stash.unstashAll(activeGroupRouter)
case msg: T @unchecked =>
import akka.actor.typed.scaladsl.adapter._
if (!stash.isFull) stash.stash(msg)
else
ctx.system.eventStream ! EventStream.Publish(Dropped(
context.system.eventStream ! EventStream.Publish(Dropped(
msg,
s"Stash is full in group router for [$serviceKey]",
ctx.self.toClassic)) // don't fail on full stash
context.self.toClassic)) // don't fail on full stash
this
}
}
@ -78,13 +80,13 @@ private[akka] final class GroupRouterImpl[T](
serviceKey: ServiceKey[T],
routingLogic: RoutingLogic[T],
routeesInitiallyEmpty: Boolean)
extends AbstractBehavior[T] {
extends AbstractBehavior[T](ctx) {
private var routeesEmpty = routeesInitiallyEmpty
def onMessage(msg: T): Behavior[T] = msg match {
case l @ serviceKey.Listing(update) =>
ctx.log.debug("Update from receptionist: [{}]", l)
context.log.debug("Update from receptionist: [{}]", l)
val routees =
if (update.nonEmpty) update
else
@ -98,8 +100,8 @@ private[akka] final class GroupRouterImpl[T](
import akka.actor.typed.scaladsl.adapter._
if (!routeesEmpty) routingLogic.selectRoutee() ! msg
else
ctx.system.eventStream ! EventStream.Publish(
Dropped(msg, s"No routees in group router for [$serviceKey]", ctx.self.toClassic))
context.system.eventStream ! EventStream.Publish(
Dropped(msg, s"No routees in group router for [$serviceKey]", context.self.toClassic))
this
}
}

14
akka-actor-typed/src/main/scala/akka/actor/typed/internal/routing/PoolRouterImpl.scala
View file

@ -42,17 +42,17 @@ private final class PoolRouterImpl[T](
poolSize: Int,
behaviorFactory: () => Behavior[T],
logic: RoutingLogic[T])
extends AbstractBehavior[T] {
extends AbstractBehavior[T](ctx) {
(1 to poolSize).foreach { _ =>
val child = ctx.spawnAnonymous(behaviorFactory())
ctx.watch(child)
val child = context.spawnAnonymous(behaviorFactory())
context.watch(child)
child
}
onRouteesChanged()
private def onRouteesChanged(): Unit = {
val children = ctx.children.toSet.asInstanceOf[Set[ActorRef[T]]]
val children = context.children.toSet.asInstanceOf[Set[ActorRef[T]]]
logic.routeesUpdated(children)
}
@ -67,12 +67,12 @@ private final class PoolRouterImpl[T](
// for the `Terminated` we receive for the first child. This means it is not certain that
// there will be a log entry per child in those cases (it does not make sense to keep the
// pool alive just to get the logging right when there are no routees available)
if (ctx.children.nonEmpty) {
ctx.log.debug("Pool child stopped [{}]", child.path)
if (context.children.nonEmpty) {
context.log.debug("Pool child stopped [{}]", child.path)
onRouteesChanged()
this
} else {
ctx.log.info("Last pool child stopped, stopping pool [{}]", ctx.self.path)
context.log.info("Last pool child stopped, stopping pool [{}]", context.self.path)
Behaviors.stopped
}
}

44
akka-actor-typed/src/main/scala/akka/actor/typed/javadsl/AbstractBehavior.scala
View file

@ -16,14 +16,29 @@ import akka.util.OptionVal
* alternative is provided by the factory methods in [[Behaviors]], for example
* [[Behaviors.receiveMessage]].
*
* Instances of this behavior should be created via [[Behaviors.setup]] and if
* the [[ActorContext]] is needed it can be passed as a constructor parameter
* from the factory function. This is important because a new instance
* Instances of this behavior should be created via [[Behaviors.setup]] and
* the [[ActorContext]] should be passed as a constructor parameter
* from the factory function. This is important because a new instance
* should be created when restart supervision is used.
*
* When switching `Behavior` to another `AbstractBehavior` the original `ActorContext`
* can be used as the `context` parameter instead of wrapping in a new `Behaviors.setup`,
* but it wouldn't be wrong to use `context` from `Behaviors.setup` since that is the same
* `ActorContext` instance.
*
* It must not be created with an `ActorContext` of another actor, such as the parent actor.
* Such mistake will be detected at runtime and throw `IllegalStateException` when the
* first message is received.
*
* @see [[Behaviors.setup]]
*/
abstract class AbstractBehavior[T] extends ExtensibleBehavior[T] {
abstract class AbstractBehavior[T](context: ActorContext[T]) extends ExtensibleBehavior[T] {
if (context eq null)
throw new IllegalArgumentException(
"context must not be null. Wrap in Behaviors.setup and " +
"pass the context to the constructor of AbstractBehavior.")
private var _receive: OptionVal[Receive[T]] = OptionVal.None
private def receive: Receive[T] = _receive match {
case OptionVal.None =>
@ -33,13 +48,26 @@ abstract class AbstractBehavior[T] extends ExtensibleBehavior[T] {
case OptionVal.Some(r) => r
}
@throws(classOf[Exception])
override final def receive(ctx: TypedActorContext[T], msg: T): Behavior[T] =
receive.receive(ctx, msg)
def getContext: ActorContext[T] = context
private def checkRightContext(ctx: TypedActorContext[T]): Unit =
if (ctx.asJava ne context)
throw new IllegalStateException(
s"Actor [${ctx.asJava.getSelf}] of AbstractBehavior class " +
s"[${getClass.getName}] was created with wrong ActorContext [${context.asJava.getSelf}]. " +
"Wrap in Behaviors.setup and pass the context to the constructor of AbstractBehavior.")
@throws(classOf[Exception])
override final def receiveSignal(ctx: TypedActorContext[T], msg: Signal): Behavior[T] =
override final def receive(ctx: TypedActorContext[T], msg: T): Behavior[T] = {
checkRightContext(ctx)
receive.receive(ctx, msg)
}
@throws(classOf[Exception])
override final def receiveSignal(ctx: TypedActorContext[T], msg: Signal): Behavior[T] = {
checkRightContext(ctx)
receive.receiveSignal(ctx, msg)
}
/**
* Implement this to define how messages and signals are processed. Use the

39
akka-actor-typed/src/main/scala/akka/actor/typed/scaladsl/AbstractBehavior.scala
View file

@ -16,14 +16,28 @@ import akka.actor.typed.{ Behavior, ExtensibleBehavior, Signal, TypedActorContex
* alternative is provided by the factory methods in [[Behaviors]], for example
* [[Behaviors.receiveMessage]].
*
* Instances of this behavior should be created via [[Behaviors.setup]] and if
* the [[ActorContext]] is needed it can be passed as a constructor parameter
* Instances of this behavior should be created via [[Behaviors.setup]] and
* the [[ActorContext]] should be passed as a constructor parameter
* from the factory function. This is important because a new instance
* should be created when restart supervision is used.
*
* When switching `Behavior` to another `AbstractBehavior` the original `ActorContext`
* can be used as the `context` parameter instead of wrapping in a new `Behaviors.setup`,
* but it wouldn't be wrong to use `context` from `Behaviors.setup` since that is the same
* `ActorContext` instance.
*
* It must not be created with an `ActorContext` of another actor, such as the parent actor.
* Such mistake will be detected at runtime and throw `IllegalStateException` when the
* first message is received.
*
* @see [[Behaviors.setup]]
*/
abstract class AbstractBehavior[T] extends ExtensibleBehavior[T] {
abstract class AbstractBehavior[T](protected val context: ActorContext[T]) extends ExtensibleBehavior[T] {
if (context eq null)
throw new IllegalArgumentException(
"context must not be null. Wrap in Behaviors.setup and " +
"pass the context to the constructor of AbstractBehavior.")
/**
* Implement this method to process an incoming message and return the next behavior.
@ -55,11 +69,22 @@ abstract class AbstractBehavior[T] extends ExtensibleBehavior[T] {
@throws(classOf[Exception])
def onSignal: PartialFunction[Signal, Behavior[T]] = PartialFunction.empty
@throws(classOf[Exception])
override final def receive(ctx: TypedActorContext[T], msg: T): Behavior[T] =
onMessage(msg)
private def checkRightContext(ctx: TypedActorContext[T]): Unit =
if (ctx.asJava ne context)
throw new IllegalStateException(
s"Actor [${ctx.asJava.getSelf}] of AbstractBehavior class " +
s"[${getClass.getName}] was created with wrong ActorContext [${context.asJava.getSelf}]. " +
"Wrap in Behaviors.setup and pass the context to the constructor of AbstractBehavior.")
@throws(classOf[Exception])
override final def receiveSignal(ctx: TypedActorContext[T], msg: Signal): Behavior[T] =
override final def receive(ctx: TypedActorContext[T], msg: T): Behavior[T] = {
checkRightContext(ctx)
onMessage(msg)
}
@throws(classOf[Exception])
override final def receiveSignal(ctx: TypedActorContext[T], msg: Signal): Behavior[T] = {
checkRightContext(ctx)
onSignal.applyOrElse(msg, { case _ => Behaviors.unhandled }: PartialFunction[Signal, Behavior[T]])
}
}

8
akka-cluster-sharding-typed/src/test/java/jdocs/akka/cluster/sharding/typed/ShardingCompileOnlyTest.java
View file

@ -49,12 +49,11 @@ interface ShardingCompileOnlyTest {
return Behaviors.setup(context -> new Counter(context, entityId));
}
private final ActorContext<Command> context;
private final String entityId;
private int value = 0;
private Counter(ActorContext<Command> context, String entityId) {
this.context = context;
super(context);
this.entityId = entityId;
}
@ -112,7 +111,6 @@ interface ShardingCompileOnlyTest {
});
}
private final ActorContext<Command> context;
private final ActorRef<ClusterSharding.ShardCommand> shard;
private final String entityId;
private int value = 0;
@ -121,7 +119,7 @@ interface ShardingCompileOnlyTest {
ActorContext<Command> context,
ActorRef<ClusterSharding.ShardCommand> shard,
String entityId) {
this.context = context;
super(context);
this.shard = shard;
this.entityId = entityId;
}
@ -148,7 +146,7 @@ interface ShardingCompileOnlyTest {
private Behavior<Command> onIdle() {
// after receive timeout
shard.tell(new ClusterSharding.Passivate<>(context.getSelf()));
shard.tell(new ClusterSharding.Passivate<>(getContext().getSelf()));
return this;
}

1
akka-cluster-sharding-typed/src/test/java/jdocs/akka/cluster/sharding/typed/ShardingReplyCompileOnlyTest.java
View file

@ -59,6 +59,7 @@ interface ShardingReplyCompileOnlyTest {
private int value = 0;
private Counter(ActorContext<Command> context) {
super(context);
this.sharding = ClusterSharding.get(context.getSystem());
}

3
akka-cluster-typed/src/test/java/jdocs/akka/cluster/ddata/typed/javadsl/ReplicatorDocSample.java
View file

@ -81,7 +81,6 @@ interface ReplicatorDocSample {
new Counter(ctx, replicatorAdapter, key)));
}
private final ActorContext<Command> context;
// adapter that turns the response messages from the replicator into our own protocol
private final ReplicatorMessageAdapter<Command, GCounter> replicatorAdapter;
private final SelfUniqueAddress node;
@ -93,8 +92,8 @@ interface ReplicatorDocSample {
ActorContext<Command> context,
ReplicatorMessageAdapter<Command, GCounter> replicatorAdapter,
Key<GCounter> key) {
super(context);
this.context = context;
this.replicatorAdapter = replicatorAdapter;
this.key = key;

3
akka-cluster-typed/src/test/java/jdocs/akka/cluster/typed/SingletonCompileOnlyTest.java
View file

@ -45,11 +45,10 @@ public interface SingletonCompileOnlyTest {
return Behaviors.setup(Counter::new);
}
private final ActorContext<Command> context;
private int value = 0;
private Counter(ActorContext<Command> context) {
this.context = context;
super(context);
}
@Override

6
akka-docs/src/main/paradox/project/migration-guide-2.5.x-2.6.x.md
View file

@ -542,7 +542,11 @@ made before finalizing the APIs. Compared to Akka 2.5.x the source incompatible
* `akka.cluster.sharding.typed.scaladsl.Entity.apply` changed to use two parameter lists because the new
`EntityContext.entityTypeKey` required additional type parameter that is inferred better with a secondary
parameter list.
* `EventSourcedBehavior.withEnforcedReplies` signature changed. Command is not required to extend `ExpectingReply` anymore. `ExpectingReply` has been removed therefore.
* `EventSourcedBehavior.withEnforcedReplies` signature changed. Command is not required to extend `ExpectingReply`
anymore. `ExpectingReply` has therefore been removed.
* `ActorContext` is now a mandatory constructor parameter in `AbstractBehavior`. Create via `Behaviors.setup.
The reason is to encourage right usage and detect mistakes like not creating a new instance (via `setup`)
when the behavior is supervised and restarted.
#### Akka Typed Stream API changes

4
akka-docs/src/test/java/jdocs/actor/typed/BlockingActor.java
View file

@ -13,7 +13,9 @@ public class BlockingActor extends AbstractBehavior<Integer> {
return Behaviors.setup(BlockingActor::new);
}
private BlockingActor(ActorContext<Integer> context) {}
private BlockingActor(ActorContext<Integer> context) {
super(context);
}
@Override
public Receive<Integer> createReceive() {

2
akka-docs/src/test/java/jdocs/actor/typed/BlockingDispatcherTest.java
View file

@ -14,7 +14,7 @@ public class BlockingDispatcherTest {
Behaviors.setup(
context -> {
ActorRef<Integer> actor1 = context.spawn(BlockingActor.create(), "BlockingActor");
ActorRef<Integer> actor2 = context.spawn(new PrintActor(), "PrintActor");
ActorRef<Integer> actor2 = context.spawn(PrintActor.create(), "PrintActor");
for (int i = 0; i < 100; i++) {
actor1.tell(i);

8
akka-docs/src/test/java/jdocs/actor/typed/DispatcherDocTest.java
View file

@ -21,7 +21,7 @@ public class DispatcherDocTest {
// #defining-dispatcher-in-code
ActorRef<Integer> myActor =
context.spawn(
new PrintActor(), "PrintActor", DispatcherSelector.fromConfig("my-dispatcher"));
PrintActor.create(), "PrintActor", DispatcherSelector.fromConfig("my-dispatcher"));
// #defining-dispatcher-in-code
}
@ -29,7 +29,7 @@ public class DispatcherDocTest {
// #defining-fixed-pool-size-dispatcher
ActorRef<Integer> myActor =
context.spawn(
new PrintActor(),
PrintActor.create(),
"PrintActor",
DispatcherSelector.fromConfig("blocking-io-dispatcher"));
// #defining-fixed-pool-size-dispatcher
@ -39,7 +39,9 @@ public class DispatcherDocTest {
// #defining-pinned-dispatcher
ActorRef<Integer> myActor =
context.spawn(
new PrintActor(), "PrintActor", DispatcherSelector.fromConfig("my-pinned-dispatcher"));
PrintActor.create(),
"PrintActor",
DispatcherSelector.fromConfig("my-pinned-dispatcher"));
// #defining-pinned-dispatcher
}

6
akka-docs/src/test/java/jdocs/actor/typed/PrintActor.java
View file

@ -11,7 +11,11 @@ import akka.actor.typed.javadsl.*;
class PrintActor extends AbstractBehavior<Integer> {
public static Behavior<Integer> create() {
return Behaviors.setup(context -> new PrintActor());
return Behaviors.setup(PrintActor::new);
}
private PrintActor(ActorContext<Integer> context) {
super(context);
}
@Override

1
akka-docs/src/test/java/jdocs/actor/typed/SeparateDispatcherCompletionStageActor.java
View file

@ -19,6 +19,7 @@ class SeparateDispatcherFutureActor extends AbstractBehavior<Integer> {
}
private SeparateDispatcherFutureActor(ActorContext<Integer> context) {
super(context);
ec =
context
.getSystem()

38
akka-docs/src/test/java/jdocs/actor/typed/SharedMutableStateDocTest.java
View file

@ -52,13 +52,11 @@ interface SharedMutableStateDocTest {
}
}
private final ActorContext<Command> context;
private String state = "";
private Set<String> mySet = new HashSet<>();
public MyActor(ActorContext<Command> context) {
this.context = context;
super(context);
}
@Override
@ -96,12 +94,13 @@ interface SharedMutableStateDocTest {
// Turn the future result into a message that is sent to
// self when future completes
CompletableFuture<String> futureResult = expensiveCalculation();
context.pipeToSelf(
futureResult,
(result, failure) -> {
if (result != null) return new UpdateState(result);
else throw new RuntimeException(failure);
});
getContext()
.pipeToSelf(
futureResult,
(result, failure) -> {
if (result != null) return new UpdateState(result);
else throw new RuntimeException(failure);
});
// Another example of incorrect approach
// mutating actor state from ask future callback
@ -110,7 +109,7 @@ interface SharedMutableStateDocTest {
message.otherActor,
Query::new,
Duration.ofSeconds(3),
context.getSystem().scheduler());
getContext().getSystem().scheduler());
response.whenComplete(
(result, failure) -> {
if (result != null) state = "new state: " + result;
@ -118,15 +117,16 @@ interface SharedMutableStateDocTest {
// use context.ask instead, turns the completion
// into a message sent to self
context.ask(
String.class,
message.otherActor,
Duration.ofSeconds(3),
Query::new,
(result, failure) -> {
if (result != null) return new UpdateState(result);
else throw new RuntimeException(failure);
});
getContext()
.ask(
String.class,
message.otherActor,
Duration.ofSeconds(3),
Query::new,
(result, failure) -> {
if (result != null) return new UpdateState(result);
else throw new RuntimeException(failure);
});
return this;
}

6
akka-docs/src/test/java/jdocs/persistence/ResumableProjectionExample.java
View file

@ -9,6 +9,7 @@ import akka.actor.typed.ActorRef;
import akka.actor.typed.ActorSystem;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Adapter;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
@ -79,14 +80,15 @@ public interface ResumableProjectionExample {
}
public static Behavior<Command> create(String id, ExampleStore store) {
return Behaviors.setup(context -> new TheOneWhoWritesToQueryJournal(store));
return Behaviors.setup(context -> new TheOneWhoWritesToQueryJournal(context, store));
}
private final ExampleStore store;
private ComplexState state = new ComplexState();
private TheOneWhoWritesToQueryJournal(ExampleStore store) {
private TheOneWhoWritesToQueryJournal(ActorContext<Command> context, ExampleStore store) {
super(context);
this.store = store;
}

28
akka-docs/src/test/java/jdocs/typed/tutorial_1/ActorHierarchyExperiments.java
View file

@ -28,10 +28,8 @@ class PrintMyActorRefActor extends AbstractBehavior<String> {
return Behaviors.setup(PrintMyActorRefActor::new);
}
private final ActorContext<String> context;
private PrintMyActorRefActor(ActorContext<String> context) {
this.context = context;
super(context);
}
@Override
@ -40,7 +38,7 @@ class PrintMyActorRefActor extends AbstractBehavior<String> {
}
private Behavior<String> printIt() {
ActorRef<String> secondRef = context.spawn(Behaviors.empty(), "second-actor");
ActorRef<String> secondRef = getContext().spawn(Behaviors.empty(), "second-actor");
System.out.println("Second: " + secondRef);
return this;
}
@ -51,10 +49,11 @@ class PrintMyActorRefActor extends AbstractBehavior<String> {
class StartStopActor1 extends AbstractBehavior<String> {
static Behavior<String> create() {
return Behaviors.setup(context -> new StartStopActor1());
return Behaviors.setup(StartStopActor1::new);
}
private StartStopActor1() {
private StartStopActor1(ActorContext<String> context) {
super(context);
System.out.println("first started");
}
@ -75,10 +74,11 @@ class StartStopActor1 extends AbstractBehavior<String> {
class StartStopActor2 extends AbstractBehavior<String> {
static Behavior<String> create() {
return Behaviors.setup(context -> new StartStopActor2());
return Behaviors.setup(StartStopActor2::new);
}
private StartStopActor2() {
private StartStopActor2(ActorContext<String> context) {
super(context);
System.out.println("second started");
}
@ -104,6 +104,7 @@ class SupervisingActor extends AbstractBehavior<String> {
private final ActorRef<String> child;
private SupervisingActor(ActorContext<String> context) {
super(context);
child =
context.spawn(
Behaviors.supervise(SupervisedActor.create()).onFailure(SupervisorStrategy.restart()),
@ -124,10 +125,11 @@ class SupervisingActor extends AbstractBehavior<String> {
class SupervisedActor extends AbstractBehavior<String> {
static Behavior<String> create() {
return Behaviors.setup(context -> new SupervisedActor());
return Behaviors.setup(SupervisedActor::new);
}
private SupervisedActor() {
private SupervisedActor(ActorContext<String> context) {
super(context);
System.out.println("supervised actor started");
}
@ -165,10 +167,8 @@ class Main extends AbstractBehavior<String> {
return Behaviors.setup(Main::new);
}
private final ActorContext<String> context;
private Main(ActorContext<String> context) {
this.context = context;
super(context);
}
@Override
@ -177,7 +177,7 @@ class Main extends AbstractBehavior<String> {
}
private Behavior<String> start() {
ActorRef<String> firstRef = context.spawn(PrintMyActorRefActor.create(), "first-actor");
ActorRef<String> firstRef = getContext().spawn(PrintMyActorRefActor.create(), "first-actor");
System.out.println("First: " + firstRef);
firstRef.tell("printit");

8
akka-docs/src/test/java/jdocs/typed/tutorial_2/IotSupervisor.java
View file

@ -18,10 +18,8 @@ public class IotSupervisor extends AbstractBehavior<Void> {
return Behaviors.setup(IotSupervisor::new);
}
private final ActorContext<Void> context;
public IotSupervisor(ActorContext<Void> context) {
this.context = context;
private IotSupervisor(ActorContext<Void> context) {
super(context);
context.getLog().info("IoT Application started");
}
@ -32,7 +30,7 @@ public class IotSupervisor extends AbstractBehavior<Void> {
}
private IotSupervisor onPostStop() {
context.getLog().info("IoT Application stopped");
getContext().getLog().info("IoT Application stopped");
return this;
}
}

7
akka-docs/src/test/java/jdocs/typed/tutorial_3/Device.java
View file

@ -66,14 +66,13 @@ public class Device extends AbstractBehavior<Device.Command> {
return Behaviors.setup(context -> new Device(context, groupId, deviceId));
}
private final ActorContext<Command> context;
private final String groupId;
private final String deviceId;
private Optional<Double> lastTemperatureReading = Optional.empty();
private Device(ActorContext<Command> context, String groupId, String deviceId) {
this.context = context;
super(context);
this.groupId = groupId;
this.deviceId = deviceId;
@ -90,7 +89,7 @@ public class Device extends AbstractBehavior<Device.Command> {
}
private Behavior<Command> onRecordTemperature(RecordTemperature r) {
context.getLog().info("Recorded temperature reading {} with {}", r.value, r.requestId);
getContext().getLog().info("Recorded temperature reading {} with {}", r.value, r.requestId);
lastTemperatureReading = Optional.of(r.value);
r.replyTo.tell(new TemperatureRecorded(r.requestId));
return this;
@ -102,7 +101,7 @@ public class Device extends AbstractBehavior<Device.Command> {
}
private Behavior<Command> onPostStop() {
context.getLog().info("Device actor {}-{} stopped", groupId, deviceId);
getContext().getLog().info("Device actor {}-{} stopped", groupId, deviceId);
return Behaviors.stopped();
}
}

5
akka-docs/src/test/java/jdocs/typed/tutorial_3/inprogress2/Device.java
View file

@ -45,14 +45,13 @@ public class Device extends AbstractBehavior<Device.Command> {
return Behaviors.setup(context -> new Device(context, groupId, deviceId));
}
private final ActorContext<Command> context;
private final String groupId;
private final String deviceId;
private Optional<Double> lastTemperatureReading = Optional.empty();
private Device(ActorContext<Command> context, String groupId, String deviceId) {
this.context = context;
super(context);
this.groupId = groupId;
this.deviceId = deviceId;
@ -73,7 +72,7 @@ public class Device extends AbstractBehavior<Device.Command> {
}
private Device onPostStop() {
context.getLog().info("Device actor {}-{} stopped", groupId, deviceId);
getContext().getLog().info("Device actor {}-{} stopped", groupId, deviceId);
return this;
}
// #read-protocol-2

7
akka-docs/src/test/java/jdocs/typed/tutorial_4/Device.java
View file

@ -70,14 +70,13 @@ public class Device extends AbstractBehavior<Device.Command> {
return Behaviors.setup(context -> new Device(context, groupId, deviceId));
}
private final ActorContext<Command> context;
private final String groupId;
private final String deviceId;
private Optional<Double> lastTemperatureReading = Optional.empty();
private Device(ActorContext<Command> context, String groupId, String deviceId) {
this.context = context;
super(context);
this.groupId = groupId;
this.deviceId = deviceId;
@ -95,7 +94,7 @@ public class Device extends AbstractBehavior<Device.Command> {
}
private Behavior<Command> onRecordTemperature(RecordTemperature r) {
context.getLog().info("Recorded temperature reading {} with {}", r.value, r.requestId);
getContext().getLog().info("Recorded temperature reading {} with {}", r.value, r.requestId);
lastTemperatureReading = Optional.of(r.value);
r.replyTo.tell(new TemperatureRecorded(r.requestId));
return this;
@ -107,7 +106,7 @@ public class Device extends AbstractBehavior<Device.Command> {
}
private Behavior<Command> onPostStop() {
context.getLog().info("Device actor {}-{} stopped", groupId, deviceId);
getContext().getLog().info("Device actor {}-{} stopped", groupId, deviceId);
return Behaviors.stopped();
}
}

18
akka-docs/src/test/java/jdocs/typed/tutorial_4/DeviceGroup.java
View file

@ -40,12 +40,11 @@ public class DeviceGroup extends AbstractBehavior<DeviceGroup.Command> {
return Behaviors.setup(context -> new DeviceGroup(context, groupId));
}
private final ActorContext<Command> context;
private final String groupId;
private final Map<String, ActorRef<Device.Command>> deviceIdToActor = new HashMap<>();
private DeviceGroup(ActorContext<Command> context, String groupId) {
this.context = context;
super(context);
this.groupId = groupId;
context.getLog().info("DeviceGroup {} started", groupId);
}
@ -56,18 +55,19 @@ public class DeviceGroup extends AbstractBehavior<DeviceGroup.Command> {
if (deviceActor != null) {
trackMsg.replyTo.tell(new DeviceManager.DeviceRegistered(deviceActor));
} else {
context.getLog().info("Creating device actor for {}", trackMsg.deviceId);
getContext().getLog().info("Creating device actor for {}", trackMsg.deviceId);
deviceActor =
context.spawn(Device.create(groupId, trackMsg.deviceId), "device-" + trackMsg.deviceId);
getContext()
.spawn(Device.create(groupId, trackMsg.deviceId), "device-" + trackMsg.deviceId);
// #device-group-register
context.watchWith(
deviceActor, new DeviceTerminated(deviceActor, groupId, trackMsg.deviceId));
getContext()
.watchWith(deviceActor, new DeviceTerminated(deviceActor, groupId, trackMsg.deviceId));
// #device-group-register
deviceIdToActor.put(trackMsg.deviceId, deviceActor);
trackMsg.replyTo.tell(new DeviceManager.DeviceRegistered(deviceActor));
}
} else {
context
getContext()
.getLog()
.warn(
"Ignoring TrackDevice request for {}. This actor is responsible for {}.",
@ -87,7 +87,7 @@ public class DeviceGroup extends AbstractBehavior<DeviceGroup.Command> {
// #device-group-remove
private DeviceGroup onTerminated(DeviceTerminated t) {
context.getLog().info("Device actor for {} has been terminated", t.deviceId);
getContext().getLog().info("Device actor for {} has been terminated", t.deviceId);
deviceIdToActor.remove(t.deviceId);
return this;
}
@ -111,7 +111,7 @@ public class DeviceGroup extends AbstractBehavior<DeviceGroup.Command> {
}
private DeviceGroup onPostStop() {
context.getLog().info("DeviceGroup {} stopped", groupId);
getContext().getLog().info("DeviceGroup {} stopped", groupId);
return this;
}
}

13
akka-docs/src/test/java/jdocs/typed/tutorial_4/DeviceManager.java
View file

@ -82,11 +82,10 @@ public class DeviceManager extends AbstractBehavior<DeviceManager.Command> {
return Behaviors.setup(DeviceManager::new);
}
private final ActorContext<Command> context;
private final Map<String, ActorRef<DeviceGroup.Command>> groupIdToActor = new HashMap<>();
private DeviceManager(ActorContext<Command> context) {
this.context = context;
super(context);
context.getLog().info("DeviceManager started");
}
@ -96,10 +95,10 @@ public class DeviceManager extends AbstractBehavior<DeviceManager.Command> {
if (ref != null) {
ref.tell(trackMsg);
} else {
context.getLog().info("Creating device group actor for {}", groupId);
getContext().getLog().info("Creating device group actor for {}", groupId);
ActorRef<DeviceGroup.Command> groupActor =
context.spawn(DeviceGroup.create(groupId), "group-" + groupId);
context.watchWith(groupActor, new DeviceGroupTerminated(groupId));
getContext().spawn(DeviceGroup.create(groupId), "group-" + groupId);
getContext().watchWith(groupActor, new DeviceGroupTerminated(groupId));
groupActor.tell(trackMsg);
groupIdToActor.put(groupId, groupActor);
}
@ -117,7 +116,7 @@ public class DeviceManager extends AbstractBehavior<DeviceManager.Command> {
}
private DeviceManager onTerminated(DeviceGroupTerminated t) {
context.getLog().info("Device group actor for {} has been terminated", t.groupId);
getContext().getLog().info("Device group actor for {} has been terminated", t.groupId);
groupIdToActor.remove(t.groupId);
return this;
}
@ -132,7 +131,7 @@ public class DeviceManager extends AbstractBehavior<DeviceManager.Command> {
}
private DeviceManager onPostStop() {
context.getLog().info("DeviceManager stopped");
getContext().getLog().info("DeviceManager stopped");
return this;
}
// #device-registration-msgs

7
akka-docs/src/test/java/jdocs/typed/tutorial_5/Device.java
View file

@ -68,14 +68,13 @@ public class Device extends AbstractBehavior<Device.Command> {
return Behaviors.setup(context -> new Device(context, groupId, deviceId));
}
private final ActorContext<Command> context;
private final String groupId;
private final String deviceId;
private Optional<Double> lastTemperatureReading = Optional.empty();
private Device(ActorContext<Command> context, String groupId, String deviceId) {
this.context = context;
super(context);
this.groupId = groupId;
this.deviceId = deviceId;
@ -93,7 +92,7 @@ public class Device extends AbstractBehavior<Device.Command> {
}
private Behavior<Command> onRecordTemperature(RecordTemperature r) {
context.getLog().info("Recorded temperature reading {} with {}", r.value, r.requestId);
getContext().getLog().info("Recorded temperature reading {} with {}", r.value, r.requestId);
lastTemperatureReading = Optional.of(r.value);
r.replyTo.tell(new TemperatureRecorded(r.requestId));
return this;
@ -105,7 +104,7 @@ public class Device extends AbstractBehavior<Device.Command> {
}
private Behavior<Command> onPostStop() {
context.getLog().info("Device actor {}-{} stopped", groupId, deviceId);
getContext().getLog().info("Device actor {}-{} stopped", groupId, deviceId);
return Behaviors.stopped();
}
}

25
akka-docs/src/test/java/jdocs/typed/tutorial_5/DeviceGroup.java
View file

@ -37,12 +37,11 @@ public class DeviceGroup extends AbstractBehavior<DeviceGroup.Command> {
return Behaviors.setup(context -> new DeviceGroup(context, groupId));
}
private final ActorContext<Command> context;
private final String groupId;
private final Map<String, ActorRef<Device.Command>> deviceIdToActor = new HashMap<>();
private DeviceGroup(ActorContext<Command> context, String groupId) {
this.context = context;
super(context);
this.groupId = groupId;
context.getLog().info("DeviceGroup {} started", groupId);
}
@ -54,16 +53,17 @@ public class DeviceGroup extends AbstractBehavior<DeviceGroup.Command> {
if (deviceActor != null) {
trackMsg.replyTo.tell(new DeviceManager.DeviceRegistered(deviceActor));
} else {
context.getLog().info("Creating device actor for {}", trackMsg.deviceId);
getContext().getLog().info("Creating device actor for {}", trackMsg.deviceId);
deviceActor =
context.spawn(Device.create(groupId, trackMsg.deviceId), "device-" + trackMsg.deviceId);
context.watchWith(
deviceActor, new DeviceTerminated(deviceActor, groupId, trackMsg.deviceId));
getContext()
.spawn(Device.create(groupId, trackMsg.deviceId), "device-" + trackMsg.deviceId);
getContext()
.watchWith(deviceActor, new DeviceTerminated(deviceActor, groupId, trackMsg.deviceId));
deviceIdToActor.put(trackMsg.deviceId, deviceActor);
trackMsg.replyTo.tell(new DeviceManager.DeviceRegistered(deviceActor));
}
} else {
context
getContext()
.getLog()
.warn(
"Ignoring TrackDevice request for {}. This actor is responsible for {}.",
@ -79,13 +79,13 @@ public class DeviceGroup extends AbstractBehavior<DeviceGroup.Command> {
}
private DeviceGroup onTerminated(DeviceTerminated t) {
context.getLog().info("Device actor for {} has been terminated", t.deviceId);
getContext().getLog().info("Device actor for {} has been terminated", t.deviceId);
deviceIdToActor.remove(t.deviceId);
return this;
}
private DeviceGroup onPostStop() {
context.getLog().info("DeviceGroup {} stopped", groupId);
getContext().getLog().info("DeviceGroup {} stopped", groupId);
return this;
}
@ -101,9 +101,10 @@ public class DeviceGroup extends AbstractBehavior<DeviceGroup.Command> {
// applications!
Map<String, ActorRef<Device.Command>> deviceIdToActorCopy = new HashMap<>(this.deviceIdToActor);
context.spawnAnonymous(
DeviceGroupQuery.create(
deviceIdToActorCopy, r.requestId, r.replyTo, Duration.ofSeconds(3)));
getContext()
.spawnAnonymous(
DeviceGroupQuery.create(
deviceIdToActorCopy, r.requestId, r.replyTo, Duration.ofSeconds(3)));
return this;
}

1
akka-docs/src/test/java/jdocs/typed/tutorial_5/DeviceGroupQuery.java
View file

@ -74,6 +74,7 @@ public class DeviceGroupQuery extends AbstractBehavior<DeviceGroupQuery.Command>
Duration timeout,
ActorContext<Command> context,
TimerScheduler<Command> timers) {
super(context);
this.requestId = requestId;
this.requester = requester;

13
akka-docs/src/test/java/jdocs/typed/tutorial_5/DeviceManager.java
View file

@ -144,7 +144,6 @@ public class DeviceManager extends AbstractBehavior<DeviceManager.Command> {
}
}
private final ActorContext<Command> context;
private final Map<String, ActorRef<DeviceGroup.Command>> groupIdToActor = new HashMap<>();
public static Behavior<Command> create() {
@ -152,7 +151,7 @@ public class DeviceManager extends AbstractBehavior<DeviceManager.Command> {
}
private DeviceManager(ActorContext<Command> context) {
this.context = context;
super(context);
context.getLog().info("DeviceManager started");
}
@ -162,10 +161,10 @@ public class DeviceManager extends AbstractBehavior<DeviceManager.Command> {
if (ref != null) {
ref.tell(trackMsg);
} else {
context.getLog().info("Creating device group actor for {}", groupId);
getContext().getLog().info("Creating device group actor for {}", groupId);
ActorRef<DeviceGroup.Command> groupActor =
context.spawn(DeviceGroup.create(groupId), "group-" + groupId);
context.watchWith(groupActor, new DeviceGroupTerminated(groupId));
getContext().spawn(DeviceGroup.create(groupId), "group-" + groupId);
getContext().watchWith(groupActor, new DeviceGroupTerminated(groupId));
groupActor.tell(trackMsg);
groupIdToActor.put(groupId, groupActor);
}
@ -193,7 +192,7 @@ public class DeviceManager extends AbstractBehavior<DeviceManager.Command> {
}
private DeviceManager onTerminated(DeviceGroupTerminated t) {
context.getLog().info("Device group actor for {} has been terminated", t.groupId);
getContext().getLog().info("Device group actor for {} has been terminated", t.groupId);
groupIdToActor.remove(t.groupId);
return this;
}
@ -209,7 +208,7 @@ public class DeviceManager extends AbstractBehavior<DeviceManager.Command> {
}
private DeviceManager onPostStop() {
context.getLog().info("DeviceManager stopped");
getContext().getLog().info("DeviceManager stopped");
return this;
}
}

2
akka-docs/src/test/scala/docs/actor/typed/SharedMutableStateDocSpec.scala
View file

@ -29,7 +29,7 @@ class SharedMutableStateDocSpec {
}
}
//#mutable-state
class MyActor(context: ActorContext[MyActor.Command]) extends AbstractBehavior[MyActor.Command] {
class MyActor(context: ActorContext[MyActor.Command]) extends AbstractBehavior[MyActor.Command](context) {
import MyActor._
var state = ""

16
akka-docs/src/test/scala/typed/tutorial_1/ActorHierarchyExperiments.scala
View file

@ -23,7 +23,7 @@ object PrintMyActorRefActor {
Behaviors.setup(context => new PrintMyActorRefActor(context))
}
class PrintMyActorRefActor(context: ActorContext[String]) extends AbstractBehavior[String] {
class PrintMyActorRefActor(context: ActorContext[String]) extends AbstractBehavior[String](context) {
override def onMessage(msg: String): Behavior[String] =
msg match {
@ -41,7 +41,7 @@ object StartStopActor1 {
Behaviors.setup(context => new StartStopActor1(context))
}
class StartStopActor1(context: ActorContext[String]) extends AbstractBehavior[String] {
class StartStopActor1(context: ActorContext[String]) extends AbstractBehavior[String](context) {
println("first started")
context.spawn(StartStopActor2(), "second")
@ -60,10 +60,10 @@ class StartStopActor1(context: ActorContext[String]) extends AbstractBehavior[St
object StartStopActor2 {
def apply(): Behavior[String] =
Behaviors.setup(_ => new StartStopActor2)
Behaviors.setup(new StartStopActor2(_))
}
class StartStopActor2 extends AbstractBehavior[String] {
class StartStopActor2(context: ActorContext[String]) extends AbstractBehavior[String](context) {
println("second started")
override def onMessage(msg: String): Behavior[String] = {
@ -86,7 +86,7 @@ object SupervisingActor {
Behaviors.setup(context => new SupervisingActor(context))
}
class SupervisingActor(context: ActorContext[String]) extends AbstractBehavior[String] {
class SupervisingActor(context: ActorContext[String]) extends AbstractBehavior[String](context) {
private val child = context.spawn(
Behaviors.supervise(SupervisedActor()).onFailure(SupervisorStrategy.restart),
name = "supervised-actor")
@ -101,10 +101,10 @@ class SupervisingActor(context: ActorContext[String]) extends AbstractBehavior[S
object SupervisedActor {
def apply(): Behavior[String] =
Behaviors.setup(_ => new SupervisedActor)
Behaviors.setup(context => new SupervisedActor(context))
}
class SupervisedActor extends AbstractBehavior[String] {
class SupervisedActor(context: ActorContext[String]) extends AbstractBehavior[String](context) {
println("supervised actor started")
override def onMessage(msg: String): Behavior[String] =
@ -134,7 +134,7 @@ object Main {
}
class Main(context: ActorContext[String]) extends AbstractBehavior[String] {
class Main(context: ActorContext[String]) extends AbstractBehavior[String](context) {
override def onMessage(msg: String): Behavior[String] =
msg match {
case "start" =>

2
akka-docs/src/test/scala/typed/tutorial_2/IotSupervisor.scala
View file

@ -17,7 +17,7 @@ object IotSupervisor {
Behaviors.setup[Nothing](context => new IotSupervisor(context))
}
class IotSupervisor(context: ActorContext[Nothing]) extends AbstractBehavior[Nothing] {
class IotSupervisor(context: ActorContext[Nothing]) extends AbstractBehavior[Nothing](context) {
context.log.info("IoT Application started")
override def onMessage(msg: Nothing): Behavior[Nothing] = {

2
akka-docs/src/test/scala/typed/tutorial_3/Device.scala
View file

@ -31,7 +31,7 @@ object Device {
}
class Device(context: ActorContext[Device.Command], groupId: String, deviceId: String)
extends AbstractBehavior[Device.Command] {
extends AbstractBehavior[Device.Command](context) {
import Device._
var lastTemperatureReading: Option[Double] = None

2
akka-docs/src/test/scala/typed/tutorial_3/DeviceInProgress.scala
View file

@ -43,7 +43,7 @@ object DeviceInProgress2 {
}
class Device(context: ActorContext[Device.Command], groupId: String, deviceId: String)
extends AbstractBehavior[Device.Command] {
extends AbstractBehavior[Device.Command](context) {
import Device._
var lastTemperatureReading: Option[Double] = None

2
akka-docs/src/test/scala/typed/tutorial_4/Device.scala
View file

@ -33,7 +33,7 @@ object Device {
}
class Device(context: ActorContext[Device.Command], groupId: String, deviceId: String)
extends AbstractBehavior[Device.Command] {
extends AbstractBehavior[Device.Command](context) {
import Device._
var lastTemperatureReading: Option[Double] = None

2
akka-docs/src/test/scala/typed/tutorial_4/DeviceGroup.scala
View file

@ -30,7 +30,7 @@ object DeviceGroup {
//#device-group-remove
class DeviceGroup(context: ActorContext[DeviceGroup.Command], groupId: String)
extends AbstractBehavior[DeviceGroup.Command] {
extends AbstractBehavior[DeviceGroup.Command](context) {
import DeviceGroup._
import DeviceManager.{ DeviceRegistered, ReplyDeviceList, RequestDeviceList, RequestTrackDevice }

3
akka-docs/src/test/scala/typed/tutorial_4/DeviceManager.scala
View file

@ -41,7 +41,8 @@ object DeviceManager {
//#device-manager-msgs
}
class DeviceManager(context: ActorContext[DeviceManager.Command]) extends AbstractBehavior[DeviceManager.Command] {
class DeviceManager(context: ActorContext[DeviceManager.Command])
extends AbstractBehavior[DeviceManager.Command](context) {
import DeviceManager._
var groupIdToActor = Map.empty[String, ActorRef[DeviceGroup.Command]]

2
akka-docs/src/test/scala/typed/tutorial_5/Device.scala
View file

@ -31,7 +31,7 @@ object Device {
}
class Device(context: ActorContext[Device.Command], groupId: String, deviceId: String)
extends AbstractBehavior[Device.Command] {
extends AbstractBehavior[Device.Command](context) {
import Device._
var lastTemperatureReading: Option[Double] = None

2
akka-docs/src/test/scala/typed/tutorial_5/DeviceGroup.scala
View file

@ -27,7 +27,7 @@ object DeviceGroup {
//#query-added
class DeviceGroup(context: ActorContext[DeviceGroup.Command], groupId: String)
extends AbstractBehavior[DeviceGroup.Command] {
extends AbstractBehavior[DeviceGroup.Command](context) {
import DeviceGroup._
import DeviceManager.{
DeviceRegistered,

2
akka-docs/src/test/scala/typed/tutorial_5/DeviceGroupQuery.scala
View file

@ -45,7 +45,7 @@ class DeviceGroupQuery(
timeout: FiniteDuration,
context: ActorContext[DeviceGroupQuery.Command],
timers: TimerScheduler[DeviceGroupQuery.Command])
extends AbstractBehavior[DeviceGroupQuery.Command] {
extends AbstractBehavior[DeviceGroupQuery.Command](context) {
import DeviceGroupQuery._
import DeviceManager.DeviceNotAvailable

3
akka-docs/src/test/scala/typed/tutorial_5/DeviceManager.scala
View file

@ -53,7 +53,8 @@ object DeviceManager {
//#query-protocol
}
class DeviceManager(context: ActorContext[DeviceManager.Command]) extends AbstractBehavior[DeviceManager.Command] {
class DeviceManager(context: ActorContext[DeviceManager.Command])
extends AbstractBehavior[DeviceManager.Command](context) {
import DeviceManager._
var groupIdToActor = Map.empty[String, ActorRef[DeviceGroup.Command]]

2
akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/ReplayingEvents.scala
View file

@ -65,7 +65,7 @@ private[akka] object ReplayingEvents {
private[akka] final class ReplayingEvents[C, E, S](
override val setup: BehaviorSetup[C, E, S],
var state: ReplayingState[S])
extends AbstractBehavior[InternalProtocol]
extends AbstractBehavior[InternalProtocol](setup.context)
with JournalInteractions[C, E, S]
with SnapshotInteractions[C, E, S]
with StashManagement[C, E, S]

4
akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/Running.scala
View file

@ -96,7 +96,7 @@ private[akka] object Running {
import BehaviorSetup._
final class HandlingCommands(state: RunningState[S])
extends AbstractBehavior[InternalProtocol]
extends AbstractBehavior[InternalProtocol](setup.context)
with WithSeqNrAccessible {
def onMessage(msg: InternalProtocol): Behavior[InternalProtocol] = msg match {
@ -226,7 +226,7 @@ private[akka] object Running {
shouldSnapshotAfterPersist: SnapshotAfterPersist,
var sideEffects: immutable.Seq[SideEffect[S]],
persistStartTime: Long = System.nanoTime())
extends AbstractBehavior[InternalProtocol]
extends AbstractBehavior[InternalProtocol](setup.context)
with WithSeqNrAccessible {
private var eventCounter = 0