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

Remove ActorContext parameter from javadsl.ReceiveBuilder, #27120 (#27121)

Browse source 
* Remove ActorContext parameter from javadsl.ReceiveBuilder, #27120

* functional style in javadsl
* in Java it's more practical to have an enclosing class to hold
  initialization parameters and ActorContext
* writing behaviors as pure static methods will be unlikely be used in Java
* it's still possible to write behaviors as static methods by passing
  the context around, in same way as all other things
* better to embrace the enclosing class pattern and therefore remove
  the context parameter from the message handlers
* style cleanup of ChatRoom sample
* migration guide
This commit is contained in:
Patrik Nordwall 2019-06-13 16:10:40 +02:00 • committed by GitHub
parent f004a574b4
commit 893bd8b74b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
25 changed files with 616 additions and 561 deletions
Download patch file Download diff file Expand all files Collapse all files

181
akka-actor-testkit-typed/src/test/java/akka/actor/testkit/typed/javadsl/BehaviorTestKitTest.java
View file

@ -130,95 +130,98 @@ public class BehaviorTestKitTest extends JUnitSuite {
private static Props props = Props.empty().withDispatcherFromConfig("cat");
private static Behavior<Command> behavior =
Behaviors.receive(Command.class)
.onMessage(
SpawnChildren.class,
(context, message) -> {
IntStream.range(0, message.numberOfChildren)
.forEach(
i -> {
context.spawn(childInitial, "child" + i);
});
return Behaviors.same();
})
.onMessage(
SpawnChildrenAnonymous.class,
(context, message) -> {
IntStream.range(0, message.numberOfChildren)
.forEach(
i -> {
context.spawnAnonymous(childInitial);
});
return Behaviors.same();
})
.onMessage(
SpawnChildrenWithProps.class,
(context, message) -> {
IntStream.range(0, message.numberOfChildren)
.forEach(
i -> {
context.spawn(childInitial, "child" + i, message.props);
});
return Behaviors.same();
})
.onMessage(
SpawnChildrenAnonymousWithProps.class,
(context, message) -> {
IntStream.range(0, message.numberOfChildren)
.forEach(
i -> {
context.spawnAnonymous(childInitial, message.props);
});
return Behaviors.same();
})
.onMessage(
CreateMessageAdapter.class,
(context, message) -> {
context.messageAdapter(message.clazz, message.f);
return Behaviors.same();
})
.onMessage(
SpawnWatchAndUnWatch.class,
(context, message) -> {
ActorRef<Action> c = context.spawn(childInitial, message.name);
context.watch(c);
context.unwatch(c);
return Behaviors.same();
})
.onMessage(
SpawnAndWatchWith.class,
(context, message) -> {
ActorRef<Action> c = context.spawn(childInitial, message.name);
context.watchWith(c, message);
return Behaviors.same();
})
.onMessage(
SpawnSession.class,
(context, message) -> {
ActorRef<String> session =
context.spawnAnonymous(
Behaviors.receiveMessage(
m -> {
message.sessionHandler.tell(m);
return Behaviors.same();
}));
message.replyTo.tell(session);
return Behaviors.same();
})
.onMessage(
KillSession.class,
(context, message) -> {
context.stop(message.session);
message.replyTo.tell(Done.getInstance());
return Behaviors.same();
})
.onMessage(
Log.class,
(context, message) -> {
context.getLog().info(message.what);
return Behaviors.same();
})
.build();
Behaviors.setup(
context -> {
return Behaviors.receive(Command.class)
.onMessage(
SpawnChildren.class,
message -> {
IntStream.range(0, message.numberOfChildren)
.forEach(
i -> {
context.spawn(childInitial, "child" + i);
});
return Behaviors.same();
})
.onMessage(
SpawnChildrenAnonymous.class,
message -> {
IntStream.range(0, message.numberOfChildren)
.forEach(
i -> {
context.spawnAnonymous(childInitial);
});
return Behaviors.same();
})
.onMessage(
SpawnChildrenWithProps.class,
message -> {
IntStream.range(0, message.numberOfChildren)
.forEach(
i -> {
context.spawn(childInitial, "child" + i, message.props);
});
return Behaviors.same();
})
.onMessage(
SpawnChildrenAnonymousWithProps.class,
message -> {
IntStream.range(0, message.numberOfChildren)
.forEach(
i -> {
context.spawnAnonymous(childInitial, message.props);
});
return Behaviors.same();
})
.onMessage(
CreateMessageAdapter.class,
message -> {
context.messageAdapter(message.clazz, message.f);
return Behaviors.same();
})
.onMessage(
SpawnWatchAndUnWatch.class,
message -> {
ActorRef<Action> c = context.spawn(childInitial, message.name);
context.watch(c);
context.unwatch(c);
return Behaviors.same();
})
.onMessage(
SpawnAndWatchWith.class,
message -> {
ActorRef<Action> c = context.spawn(childInitial, message.name);
context.watchWith(c, message);
return Behaviors.same();
})
.onMessage(
SpawnSession.class,
message -> {
ActorRef<String> session =
context.spawnAnonymous(
Behaviors.receiveMessage(
m -> {
message.sessionHandler.tell(m);
return Behaviors.same();
}));
message.replyTo.tell(session);
return Behaviors.same();
})
.onMessage(
KillSession.class,
message -> {
context.stop(message.session);
message.replyTo.tell(Done.getInstance());
return Behaviors.same();
})
.onMessage(
Log.class,
message -> {
context.getLog().info(message.what);
return Behaviors.same();
})
.build();
});
@Test
public void allowAssertionsOnEffectType() {

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

@ -67,47 +67,49 @@ public class SyncTestingExampleTest extends JUnitSuite {
}
public static Behavior<Command> myBehavior =
Behaviors.receive(Command.class)
.onMessage(
CreateAChild.class,
(context, message) -> {
context.spawn(childActor, message.childName);
return Behaviors.same();
})
.onMessage(
CreateAnAnonymousChild.class,
(context, message) -> {
context.spawnAnonymous(childActor);
return Behaviors.same();
})
.onMessage(
SayHelloToChild.class,
(context, message) -> {
ActorRef<String> child = context.spawn(childActor, message.childName);
child.tell("hello");
return Behaviors.same();
})
.onMessage(
SayHelloToAnonymousChild.class,
(context, message) -> {
ActorRef<String> child = context.spawnAnonymous(childActor);
child.tell("hello stranger");
return Behaviors.same();
})
.onMessage(
SayHello.class,
(context, message) -> {
message.who.tell("hello");
return Behaviors.same();
})
.onMessage(
LogAndSayHello.class,
(context, message) -> {
context.getLog().info("Saying hello to {}", message.who.path().name());
message.who.tell("hello");
return Behaviors.same();
})
.build();
Behaviors.setup(
context ->
Behaviors.receive(Command.class)
.onMessage(
CreateAChild.class,
message -> {
context.spawn(childActor, message.childName);
return Behaviors.same();
})
.onMessage(
CreateAnAnonymousChild.class,
message -> {
context.spawnAnonymous(childActor);
return Behaviors.same();
})
.onMessage(
SayHelloToChild.class,
message -> {
ActorRef<String> child = context.spawn(childActor, message.childName);
child.tell("hello");
return Behaviors.same();
})
.onMessage(
SayHelloToAnonymousChild.class,
message -> {
ActorRef<String> child = context.spawnAnonymous(childActor);
child.tell("hello stranger");
return Behaviors.same();
})
.onMessage(
SayHello.class,
message -> {
message.who.tell("hello");
return Behaviors.same();
})
.onMessage(
LogAndSayHello.class,
message -> {
context.getLog().info("Saying hello to {}", message.who.path().name());
message.who.tell("hello");
return Behaviors.same();
})
.build());
// #under-test
@Test

32
akka-actor-typed-tests/src/test/java/akka/actor/typed/javadsl/ActorLoggingTest.java
View file

@ -69,21 +69,23 @@ public class ActorLoggingTest extends JUnitSuite {
@Test
public void loggingProvidesMDC() {
Behavior<Protocol> behavior =
Behaviors.withMdc(
null,
(message) -> {
Map<String, Object> mdc = new HashMap<>();
mdc.put("txId", message.getTransactionId());
return mdc;
},
Behaviors.receive(Protocol.class)
.onMessage(
Message.class,
(context, message) -> {
context.getLog().info(message.toString());
return Behaviors.same();
})
.build());
Behaviors.setup(
context ->
Behaviors.withMdc(
null,
(message) -> {
Map<String, Object> mdc = new HashMap<>();
mdc.put("txId", message.getTransactionId());
return mdc;
},
Behaviors.receive(Protocol.class)
.onMessage(
Message.class,
message -> {
context.getLog().info(message.toString());
return Behaviors.same();
})
.build()));
CustomEventFilter eventFilter =
new CustomEventFilter(

26
akka-actor-typed-tests/src/test/java/akka/actor/typed/javadsl/BehaviorBuilderTest.java
View file

@ -40,20 +40,20 @@ public class BehaviorBuilderTest extends JUnitSuite {
Behaviors.receive(Message.class)
.onMessage(
One.class,
(context, o) -> {
o -> {
o.foo();
return same();
})
.onMessage(One.class, o -> o.foo().startsWith("a"), (context, o) -> same())
.onMessage(One.class, o -> o.foo().startsWith("a"), o -> same())
.onMessageUnchecked(
MyList.class,
(ActorContext<Message> context, MyList<String> l) -> {
(MyList<String> l) -> {
String first = l.get(0);
return Behaviors.<Message>same();
})
.onSignal(
Terminated.class,
(context, t) -> {
t -> {
System.out.println("Terminating along with " + t.getRef());
return stopped();
})
@ -67,13 +67,13 @@ public class BehaviorBuilderTest extends JUnitSuite {
BehaviorBuilder.create()
.onMessage(
String.class,
(context, msg) -> {
msg -> {
probe.ref().tell("handler 1: " + msg);
return Behaviors.same();
})
.onMessage(
String.class,
(context, msg) -> {
msg -> {
probe.ref().tell("handler 2: " + msg);
return Behaviors.same();
})
@ -90,7 +90,7 @@ public class BehaviorBuilderTest extends JUnitSuite {
BehaviorBuilder.create()
.onMessageEquals(
"message",
(context) -> {
() -> {
probe.ref().tell("got it");
return Behaviors.same();
})
@ -107,14 +107,14 @@ public class BehaviorBuilderTest extends JUnitSuite {
BehaviorBuilder.create()
.onMessage(
String.class,
(msg) -> "other".equals(msg),
(context, msg) -> {
"other"::equals,
msg -> {
probe.ref().tell("handler 1: " + msg);
return Behaviors.same();
})
.onMessage(
String.class,
(context, msg) -> {
msg -> {
probe.ref().tell("handler 2: " + msg);
return Behaviors.same();
})
@ -130,7 +130,7 @@ public class BehaviorBuilderTest extends JUnitSuite {
Behavior<Object> behavior =
BehaviorBuilder.create()
.onAnyMessage(
(context, msg) -> {
msg -> {
probe.ref().tell(msg);
return same();
})
@ -164,12 +164,12 @@ public class BehaviorBuilderTest extends JUnitSuite {
return Behaviors.receive(CounterMessage.class)
.onMessage(
Increase.class,
(context, o) -> {
o -> {
return immutableCounter(currentValue + 1);
})
.onMessage(
Get.class,
(context, o) -> {
o -> {
o.sender.tell(new Got(currentValue));
return same();
})

44
akka-actor-typed-tests/src/test/java/akka/actor/typed/javadsl/WatchTest.java
View file

@ -73,16 +73,18 @@ public class WatchTest extends JUnitSuite {
@Test
public void shouldWatchTerminatingActor() throws Exception {
Behavior<RunTest> exiting =
Behaviors.receive(RunTest.class)
.onMessage(
RunTest.class,
(context, message) -> {
ActorRef<Stop> watched = context.spawn(exitingActor, "exitingActor");
context.watch(watched);
watched.tell(new Stop());
return waitingForTermination(message.replyTo);
})
.build();
Behaviors.setup(
context ->
Behaviors.receive(RunTest.class)
.onMessage(
RunTest.class,
message -> {
ActorRef<Stop> watched = context.spawn(exitingActor, "exitingActor");
context.watch(watched);
watched.tell(new Stop());
return waitingForTermination(message.replyTo);
})
.build());
ActorRef<RunTest> exitingRef = testKit.spawn(exiting);
CompletionStage<Done> result =
@ -93,16 +95,18 @@ public class WatchTest extends JUnitSuite {
@Test
public void shouldWatchWithCustomMessage() throws Exception {
Behavior<Message> exiting =
Behaviors.receive(Message.class)
.onMessage(
RunTest.class,
(context, message) -> {
ActorRef<Stop> watched = context.spawn(exitingActor, "exitingActor");
context.watchWith(watched, new CustomTerminationMessage());
watched.tell(new Stop());
return waitingForMessage(message.replyTo);
})
.build();
Behaviors.setup(
context ->
Behaviors.receive(Message.class)
.onMessage(
RunTest.class,
message -> {
ActorRef<Stop> watched = context.spawn(exitingActor, "exitingActor");
context.watchWith(watched, new CustomTerminationMessage());
watched.tell(new Stop());
return waitingForMessage(message.replyTo);
})
.build());
ActorRef<Message> exitingRef = testKit.spawn(exiting);
// Not sure why this does not compile without an explicit cast?

4
akka-actor-typed-tests/src/test/java/akka/actor/typed/receptionist/ReceptionistApiTest.java
View file

@ -97,14 +97,14 @@ public class ReceptionistApiTest {
.onMessage(
Receptionist.Listing.class,
listing -> listing.isForKey(key),
(msgCtx, listing) -> {
listing -> {
Set<ActorRef<String>> services = listing.getServiceInstances(key);
return Behaviors.same();
})
.onMessage(
Receptionist.Registered.class,
registered -> registered.isForKey(key),
(msgCtx, registered) -> {
registered -> {
ActorRef<String> registree = registered.getServiceInstance(key);
return Behaviors.same();
})

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

@ -25,7 +25,7 @@ public class BubblingSample {
Behaviors.receive(Message.class)
.onMessage(
Fail.class,
(context, message) -> {
message -> {
throw new RuntimeException(message.text);
})
.build();
@ -45,7 +45,7 @@ public class BubblingSample {
return Behaviors.receive(Message.class)
.onMessage(
Message.class,
(innerCtx, message) -> {
message -> {
// just pass messages on to the child
child.tell(message);
return Behaviors.same();
@ -67,7 +67,7 @@ public class BubblingSample {
return Behaviors.receive(Message.class)
.onMessage(
Message.class,
(innerCtx, message) -> {
message -> {
// just pass messages on to the child
middleManagement.tell(message);
return Behaviors.same();

11
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/FSMDocTest.java
View file

@ -158,14 +158,13 @@ public class FSMDocTest {
private static Behavior<Event> uninitialized() {
return Behaviors.receive(Event.class)
.onMessage(
SetTarget.class,
(context, message) -> idle(new Todo(message.getRef(), Collections.emptyList())))
SetTarget.class, message -> idle(new Todo(message.getRef(), Collections.emptyList())))
.build();
}
private static Behavior<Event> idle(Todo data) {
return Behaviors.receive(Event.class)
.onMessage(Queue.class, (context, message) -> active(data.addElement(message)))
.onMessage(Queue.class, message -> active(data.addElement(message)))
.build();
}
@ -175,16 +174,16 @@ public class FSMDocTest {
// State timeouts done with withTimers
timers.startSingleTimer("Timeout", TIMEOUT, Duration.ofSeconds(1));
return Behaviors.receive(Event.class)
.onMessage(Queue.class, (context, message) -> active(data.addElement(message)))
.onMessage(Queue.class, message -> active(data.addElement(message)))
.onMessage(
Flush.class,
(context, message) -> {
message -> {
data.getTarget().tell(new Batch(data.queue));
return idle(data.copy(new ArrayList<>()));
})
.onMessage(
Timeout.class,
(context, message) -> {
message -> {
data.getTarget().tell(new Batch(data.queue));
return idle(data.copy(new ArrayList<>()));
})

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

@ -40,39 +40,42 @@ public class GracefulStopDocTest {
}
public static final Behavior<JobControlLanguage> mcpa =
Behaviors.receive(JobControlLanguage.class)
.onMessage(
SpawnJob.class,
(context, message) -> {
context.getSystem().log().info("Spawning job {}!", message.name);
context.spawn(Job.job(message.name), message.name);
return Behaviors.same();
})
.onSignal(
PostStop.class,
(context, signal) -> {
context.getSystem().log().info("Master Control Programme stopped");
return Behaviors.same();
})
.onMessage(
GracefulShutdown.class,
(context, message) -> {
context.getSystem().log().info("Initiating graceful shutdown...");
Behaviors.setup(
context ->
Behaviors.receive(JobControlLanguage.class)
.onMessage(
SpawnJob.class,
message -> {
context.getSystem().log().info("Spawning job {}!", message.name);
context.spawn(Job.job(message.name), message.name);
return Behaviors.same();
})
.onSignal(
PostStop.class,
signal -> {
context.getSystem().log().info("Master Control Programme stopped");
return Behaviors.same();
})
.onMessage(
GracefulShutdown.class,
message -> {
context.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!");
});
})
.onSignal(
PostStop.class,
(context, signal) -> {
context.getSystem().log().info("Master Control Programme stopped");
return Behaviors.same();
})
.build();
// 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!");
});
})
.onSignal(
PostStop.class,
signal -> {
context.getSystem().log().info("Master Control Programme stopped");
return Behaviors.same();
})
.build());
}
// #master-actor

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

@ -33,14 +33,16 @@ public class InteractionPatternsTest extends JUnitSuite {
}
static final Behavior<PrintMe> printerBehavior =
Behaviors.receive(PrintMe.class)
.onMessage(
PrintMe.class,
(context, printMe) -> {
context.getLog().info(printMe.message);
return Behaviors.same();
})
.build();
Behaviors.setup(
context ->
Behaviors.receive(PrintMe.class)
.onMessage(
PrintMe.class,
printMe -> {
context.getLog().info(printMe.message);
return Behaviors.same();
})
.build());
// #fire-and-forget-definition
// #request-response-protocol
@ -70,7 +72,7 @@ public class InteractionPatternsTest extends JUnitSuite {
Behaviors.receive(Request.class)
.onMessage(
Request.class,
(context, request) -> {
request -> {
// ... process request ...
request.respondTo.tell(new Response("Here's your response!"));
return Behaviors.same();
@ -314,7 +316,7 @@ public class InteractionPatternsTest extends JUnitSuite {
return Behaviors.receive(Msg.class)
.onMessage(
Msg.class,
(context, message) -> {
message -> {
timers.startSingleTimer(TIMER_KEY, new TimeoutMsg(), after);
List<Msg> buffer = new ArrayList<>();
buffer.add(message);
@ -332,13 +334,13 @@ public class InteractionPatternsTest extends JUnitSuite {
return Behaviors.receive(Msg.class)
.onMessage(
TimeoutMsg.class,
(context, message) -> {
message -> {
target.tell(new Batch(buffer));
return idle(timers, target, after, maxSize);
})
.onMessage(
Msg.class,
(context, message) -> {
message -> {
buffer.add(message);
if (buffer.size() == maxSize) {
timers.cancel(TIMER_KEY);
@ -399,7 +401,7 @@ public class InteractionPatternsTest extends JUnitSuite {
Behaviors.receive(HalCommand.class)
.onMessage(
OpenThePodBayDoorsPlease.class,
(context, message) -> {
message -> {
message.respondTo.tell(
new HalResponse("I'm sorry, Dave. I'm afraid I can't do that."));
return Behaviors.same();
@ -465,8 +467,8 @@ public class InteractionPatternsTest extends JUnitSuite {
// message sent to the actor
.onMessage(
AdaptedResponse.class,
(innerCtx, response) -> {
innerCtx.getLog().info("Got response from HAL: {}", response.message);
response -> {
context.getLog().info("Got response from HAL: {}", response.message);
return Behaviors.same();
})
.build();
@ -571,7 +573,7 @@ public class InteractionPatternsTest extends JUnitSuite {
return Behaviors.receive(HomeCommand.class)
.onMessage(
LeaveHome.class,
(innerCtx, message) -> {
message -> {
context.spawn(
new PrepareToLeaveHome(message.who, message.respondTo, keyCabinet, drawer),
"leaving" + message.who);

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

@ -12,10 +12,12 @@ import akka.actor.typed.Behavior;
import akka.actor.typed.Terminated;
import akka.actor.typed.Props;
import akka.actor.typed.DispatcherSelector;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
// #imports
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@ -153,8 +155,10 @@ public class IntroTest {
system.terminate();
}
// #chatroom-actor
// #chatroom-behavior
public static class ChatRoom {
// #chatroom-behavior
// #chatroom-protocol
static interface RoomCommand {}
@ -229,90 +233,109 @@ public class IntroTest {
// #chatroom-protocol
// #chatroom-behavior
public static Behavior<RoomCommand> behavior() {
return chatRoom(new ArrayList<ActorRef<SessionCommand>>());
public static Behavior<RoomCommand> create() {
return Behaviors.setup(
ctx -> new ChatRoom(ctx).chatRoom(new ArrayList<ActorRef<SessionCommand>>()));
}
private static Behavior<RoomCommand> chatRoom(List<ActorRef<SessionCommand>> sessions) {
private final ActorContext<RoomCommand> context;
private ChatRoom(ActorContext<RoomCommand> context) {
this.context = context;
}
private Behavior<RoomCommand> chatRoom(List<ActorRef<SessionCommand>> sessions) {
return Behaviors.receive(RoomCommand.class)
.onMessage(
GetSession.class,
(context, getSession) -> {
ActorRef<SessionEvent> client = getSession.replyTo;
ActorRef<SessionCommand> ses =
context.spawn(
session(context.getSelf(), getSession.screenName, client),
URLEncoder.encode(getSession.screenName, StandardCharsets.UTF_8.name()));
// narrow to only expose PostMessage
client.tell(new SessionGranted(ses.narrow()));
List<ActorRef<SessionCommand>> newSessions = new ArrayList<>(sessions);
newSessions.add(ses);
return chatRoom(newSessions);
})
.onMessage(
PublishSessionMessage.class,
(context, pub) -> {
NotifyClient notification =
new NotifyClient((new MessagePosted(pub.screenName, pub.message)));
sessions.forEach(s -> s.tell(notification));
return Behaviors.same();
})
.onMessage(GetSession.class, getSession -> onGetSession(sessions, getSession))
.onMessage(PublishSessionMessage.class, pub -> onPublishSessionMessage(sessions, pub))
.build();
}
public static Behavior<ChatRoom.SessionCommand> session(
ActorRef<RoomCommand> room, String screenName, ActorRef<SessionEvent> client) {
return Behaviors.receive(ChatRoom.SessionCommand.class)
.onMessage(
PostMessage.class,
(context, post) -> {
// from client, publish to others via the room
room.tell(new PublishSessionMessage(screenName, post.message));
return Behaviors.same();
})
.onMessage(
NotifyClient.class,
(context, notification) -> {
// published from the room
client.tell(notification.message);
return Behaviors.same();
})
.build();
private Behavior<RoomCommand> onGetSession(
List<ActorRef<SessionCommand>> sessions, GetSession getSession)
throws UnsupportedEncodingException {
ActorRef<SessionEvent> client = getSession.replyTo;
ActorRef<SessionCommand> ses =
context.spawn(
Session.create(context.getSelf(), getSession.screenName, client),
URLEncoder.encode(getSession.screenName, StandardCharsets.UTF_8.name()));
// narrow to only expose PostMessage
client.tell(new SessionGranted(ses.narrow()));
List<ActorRef<SessionCommand>> newSessions = new ArrayList<>(sessions);
newSessions.add(ses);
return chatRoom(newSessions);
}
// #chatroom-behavior
private Behavior<RoomCommand> onPublishSessionMessage(
List<ActorRef<SessionCommand>> sessions, PublishSessionMessage pub) {
NotifyClient notification =
new NotifyClient((new MessagePosted(pub.screenName, pub.message)));
sessions.forEach(s -> s.tell(notification));
return Behaviors.same();
}
static class Session {
static Behavior<ChatRoom.SessionCommand> create(
ActorRef<RoomCommand> room, String screenName, ActorRef<SessionEvent> client) {
return Behaviors.receive(ChatRoom.SessionCommand.class)
.onMessage(PostMessage.class, post -> onPostMessage(room, screenName, post))
.onMessage(NotifyClient.class, notification -> onNotifyClient(client, notification))
.build();
}
private static Behavior<SessionCommand> onPostMessage(
ActorRef<RoomCommand> room, String screenName, PostMessage post) {
// from client, publish to others via the room
room.tell(new PublishSessionMessage(screenName, post.message));
return Behaviors.same();
}
private static Behavior<SessionCommand> onNotifyClient(
ActorRef<SessionEvent> client, NotifyClient notification) {
// published from the room
client.tell(notification.message);
return Behaviors.same();
}
}
}
// #chatroom-actor
// #chatroom-behavior
// #chatroom-gabbler
public abstract static class Gabbler {
private Gabbler() {}
public static class Gabbler {
public static Behavior<ChatRoom.SessionEvent> create() {
return Behaviors.setup(ctx -> new Gabbler(ctx).behavior());
}
public static Behavior<ChatRoom.SessionEvent> behavior() {
private final ActorContext<ChatRoom.SessionEvent> context;
private Gabbler(ActorContext<ChatRoom.SessionEvent> context) {
this.context = context;
}
private Behavior<ChatRoom.SessionEvent> behavior() {
return Behaviors.receive(ChatRoom.SessionEvent.class)
.onMessage(
ChatRoom.SessionDenied.class,
(context, message) -> {
context.getLog().info("cannot start chat room session: {}", message.reason);
return Behaviors.stopped();
})
.onMessage(
ChatRoom.SessionGranted.class,
(context, message) -> {
message.handle.tell(new ChatRoom.PostMessage("Hello World!"));
return Behaviors.same();
})
.onMessage(
ChatRoom.MessagePosted.class,
(context, message) -> {
context
.getLog()
.info(
"message has been posted by '{}': {}", message.screenName, message.message);
return Behaviors.stopped();
})
.onMessage(ChatRoom.SessionDenied.class, this::onSessionDenied)
.onMessage(ChatRoom.SessionGranted.class, this::onSessionGranted)
.onMessage(ChatRoom.MessagePosted.class, this::onMessagePosted)
.build();
}
private Behavior<ChatRoom.SessionEvent> onSessionDenied(ChatRoom.SessionDenied message) {
context.getLog().info("cannot start chat room session: {}", message.reason);
return Behaviors.stopped();
}
private Behavior<ChatRoom.SessionEvent> onSessionGranted(ChatRoom.SessionGranted message) {
message.handle.tell(new ChatRoom.PostMessage("Hello World!"));
return Behaviors.same();
}
private Behavior<ChatRoom.SessionEvent> onMessagePosted(ChatRoom.MessagePosted message) {
context
.getLog()
.info("message has been posted by '{}': {}", message.screenName, message.message);
return Behaviors.stopped();
}
}
// #chatroom-gabbler
@ -323,9 +346,8 @@ public class IntroTest {
Behaviors.setup(
context -> {
ActorRef<ChatRoom.RoomCommand> chatRoom =
context.spawn(ChatRoom.behavior(), "chatRoom");
ActorRef<ChatRoom.SessionEvent> gabbler =
context.spawn(Gabbler.behavior(), "gabbler");
context.spawn(ChatRoom.create(), "chatRoom");
ActorRef<ChatRoom.SessionEvent> gabbler = context.spawn(Gabbler.create(), "gabbler");
context.watch(gabbler);
chatRoom.tell(new ChatRoom.GetSession("ol’ Gabbler", gabbler));

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

@ -11,6 +11,7 @@ import akka.actor.typed.Behavior;
import akka.actor.typed.Terminated;
import akka.actor.typed.javadsl.*;
// #imports
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@ -18,8 +19,9 @@ import java.util.List;
public class OOIntroTest {
// #chatroom-actor
// #chatroom-behavior
public static class ChatRoom {
// #chatroom-behavior
// #chatroom-protocol
static interface RoomCommand {}
@ -94,7 +96,7 @@ public class OOIntroTest {
// #chatroom-protocol
// #chatroom-behavior
public static Behavior<RoomCommand> behavior() {
public static Behavior<RoomCommand> create() {
return Behaviors.setup(ChatRoomBehavior::new);
}
@ -102,7 +104,7 @@ public class OOIntroTest {
final ActorContext<RoomCommand> context;
final List<ActorRef<SessionCommand>> sessions = new ArrayList<>();
public ChatRoomBehavior(ActorContext<RoomCommand> context) {
private ChatRoomBehavior(ActorContext<RoomCommand> context) {
this.context = context;
}
@ -110,40 +112,39 @@ public class OOIntroTest {
public Receive<RoomCommand> createReceive() {
ReceiveBuilder<RoomCommand> builder = newReceiveBuilder();
builder.onMessage(
GetSession.class,
getSession -> {
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()));
// narrow to only expose PostMessage
client.tell(new SessionGranted(ses.narrow()));
sessions.add(ses);
return this;
});
builder.onMessage(
PublishSessionMessage.class,
pub -> {
NotifyClient notification =
new NotifyClient((new MessagePosted(pub.screenName, pub.message)));
sessions.forEach(s -> s.tell(notification));
return this;
});
builder.onMessage(GetSession.class, this::onGetSession);
builder.onMessage(PublishSessionMessage.class, this::onPublishSessionMessage);
return builder.build();
}
private Behavior<RoomCommand> onGetSession(GetSession getSession)
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()));
// narrow to only expose PostMessage
client.tell(new SessionGranted(ses.narrow()));
sessions.add(ses);
return this;
}
private Behavior<RoomCommand> onPublishSessionMessage(PublishSessionMessage pub) {
NotifyClient notification =
new NotifyClient((new MessagePosted(pub.screenName, pub.message)));
sessions.forEach(s -> s.tell(notification));
return this;
}
}
public static class SessionBehavior extends AbstractBehavior<ChatRoom.SessionCommand> {
static class SessionBehavior extends AbstractBehavior<ChatRoom.SessionCommand> {
private final ActorRef<RoomCommand> room;
private final String screenName;
private final ActorRef<SessionEvent> client;
public SessionBehavior(
SessionBehavior(
ActorRef<RoomCommand> room, String screenName, ActorRef<SessionEvent> client) {
this.room = room;
this.screenName = screenName;
@ -153,33 +154,35 @@ public class OOIntroTest {
@Override
public Receive<SessionCommand> createReceive() {
return newReceiveBuilder()
.onMessage(
PostMessage.class,
post -> {
// from client, publish to others via the room
room.tell(new PublishSessionMessage(screenName, post.message));
return Behaviors.same();
})
.onMessage(
NotifyClient.class,
notification -> {
// published from the room
client.tell(notification.message);
return Behaviors.same();
})
.onMessage(PostMessage.class, this::onPostMessage)
.onMessage(NotifyClient.class, this::onNotifyClient)
.build();
}
private Behavior<SessionCommand> onPostMessage(PostMessage post) {
// from client, publish to others via the room
room.tell(new PublishSessionMessage(screenName, post.message));
return Behaviors.same();
}
private Behavior<SessionCommand> onNotifyClient(NotifyClient notification) {
// published from the room
client.tell(notification.message);
return Behaviors.same();
}
}
// #chatroom-behavior
}
// #chatroom-actor
// #chatroom-behavior
// #chatroom-gabbler
public static class Gabbler extends AbstractBehavior<ChatRoom.SessionEvent> {
public static Behavior<ChatRoom.SessionEvent> create() {
return Behaviors.setup(Gabbler::new);
}
private ActorContext<ChatRoom.SessionEvent> context;
public Gabbler(ActorContext<ChatRoom.SessionEvent> context) {
private Gabbler(ActorContext<ChatRoom.SessionEvent> context) {
this.context = context;
}
@ -187,32 +190,27 @@ public class OOIntroTest {
public Receive<ChatRoom.SessionEvent> createReceive() {
ReceiveBuilder<ChatRoom.SessionEvent> builder = newReceiveBuilder();
return builder
.onMessage(
ChatRoom.SessionDenied.class,
message -> {
context.getLog().info("cannot start chat room session: {}", message.reason);
return Behaviors.stopped();
})
.onMessage(
ChatRoom.SessionGranted.class,
message -> {
message.handle.tell(new ChatRoom.PostMessage("Hello World!"));
return Behaviors.same();
})
.onMessage(
ChatRoom.MessagePosted.class,
message -> {
context
.getLog()
.info(
"message has been posted by '{}': {}", message.screenName, message.message);
return Behaviors.stopped();
})
.onMessage(ChatRoom.SessionDenied.class, this::onSessionDenied)
.onMessage(ChatRoom.SessionGranted.class, this::onSessionGranted)
.onMessage(ChatRoom.MessagePosted.class, this::onMessagePosted)
.build();
}
public static Behavior<ChatRoom.SessionEvent> behavior() {
return Behaviors.setup(Gabbler::new);
private Behavior<ChatRoom.SessionEvent> onSessionDenied(ChatRoom.SessionDenied message) {
context.getLog().info("cannot start chat room session: {}", message.reason);
return Behaviors.stopped();
}
private Behavior<ChatRoom.SessionEvent> onSessionGranted(ChatRoom.SessionGranted message) {
message.handle.tell(new ChatRoom.PostMessage("Hello World!"));
return Behaviors.same();
}
private Behavior<ChatRoom.SessionEvent> onMessagePosted(ChatRoom.MessagePosted message) {
context
.getLog()
.info("message has been posted by '{}': {}", message.screenName, message.message);
return Behaviors.stopped();
}
}
// #chatroom-gabbler
@ -224,9 +222,8 @@ public class OOIntroTest {
Behaviors.setup(
context -> {
ActorRef<ChatRoom.RoomCommand> chatRoom =
context.spawn(ChatRoom.behavior(), "chatRoom");
ActorRef<ChatRoom.SessionEvent> gabbler =
context.spawn(Gabbler.behavior(), "gabbler");
context.spawn(ChatRoom.create(), "chatRoom");
ActorRef<ChatRoom.SessionEvent> gabbler = context.spawn(Gabbler.create(), "gabbler");
context.watch(gabbler);
chatRoom.tell(new ChatRoom.GetSession("ol’ Gabbler", gabbler));

2
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/RouterTest.java
View file

@ -43,7 +43,7 @@ public class RouterTest {
return Behaviors.receive(Command.class)
.onMessage(
DoLog.class,
(notUsed, doLog) -> {
doLog -> {
context.getLog().info("Got message {}", doLog.text);
return Behaviors.same();
})

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

@ -5,6 +5,7 @@
package jdocs.akka.typed;
// #import
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.StashBuffer;
// #import
@ -64,10 +65,8 @@ public class StashDocTest extends JUnitSuite {
}
}
static class SaveSuccess implements Command {
public static final SaveSuccess instance = new SaveSuccess();
private SaveSuccess() {}
enum SaveSuccess implements Command {
INSTANCE
}
static class DBError implements Command {
@ -78,26 +77,28 @@ public class StashDocTest extends JUnitSuite {
}
}
private final ActorContext<Command> context;
private final StashBuffer<Command> buffer = StashBuffer.create(100);
private final String id;
private final DB db;
public DataAccess(String id, DB db) {
private DataAccess(ActorContext<Command> context, String id, DB db) {
this.context = context;
this.id = id;
this.db = db;
}
Behavior<Command> behavior() {
public static Behavior<Command> create(String id, DB db) {
return Behaviors.setup(
context -> {
context.pipeToSelf(
ctx -> {
ctx.pipeToSelf(
db.load(id),
(value, cause) -> {
if (cause == null) return new InitialState(value);
else return new DBError(asRuntimeException(cause));
});
return init();
return new DataAccess(ctx, id, db).init();
});
}
@ -105,18 +106,18 @@ public class StashDocTest extends JUnitSuite {
return Behaviors.receive(Command.class)
.onMessage(
InitialState.class,
(context, message) -> {
message -> {
// now we are ready to handle stashed messages if any
return buffer.unstashAll(context, active(message.value));
})
.onMessage(
DBError.class,
(context, message) -> {
message -> {
throw message.cause;
})
.onMessage(
Command.class,
(context, message) -> {
message -> {
// stash all other messages for later processing
buffer.stash(message);
return Behaviors.same();
@ -128,17 +129,17 @@ public class StashDocTest extends JUnitSuite {
return Behaviors.receive(Command.class)
.onMessage(
Get.class,
(context, message) -> {
message -> {
message.replyTo.tell(state);
return Behaviors.same();
})
.onMessage(
Save.class,
(context, message) -> {
message -> {
context.pipeToSelf(
db.save(id, message.payload),
(value, cause) -> {
if (cause == null) return SaveSuccess.instance;
if (cause == null) return SaveSuccess.INSTANCE;
else return new DBError(asRuntimeException(cause));
});
return saving(message.payload, message.replyTo);
@ -148,20 +149,20 @@ public class StashDocTest extends JUnitSuite {
private Behavior<Command> saving(String state, ActorRef<Done> replyTo) {
return Behaviors.receive(Command.class)
.onMessageEquals(
SaveSuccess.instance,
context -> {
.onMessage(
SaveSuccess.class,
message -> {
replyTo.tell(Done.getInstance());
return buffer.unstashAll(context, active(state));
})
.onMessage(
DBError.class,
(context, message) -> {
message -> {
throw message.cause;
})
.onMessage(
Command.class,
(context, message) -> {
message -> {
buffer.stash(message);
return Behaviors.same();
})
@ -195,8 +196,7 @@ public class StashDocTest extends JUnitSuite {
}
};
final ActorRef<DataAccess.Command> dataAccess =
testKit.spawn(new DataAccess("17", db).behavior());
final ActorRef<DataAccess.Command> dataAccess = testKit.spawn(DataAccess.create("17", db));
TestProbe<String> getInbox = testKit.createTestProbe(String.class);
dataAccess.tell(new DataAccess.Get(getInbox.getRef()));
getInbox.expectMessage("TheValue");

4
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/coexistence/TypedWatchingUntypedTest.java
View file

@ -50,11 +50,11 @@ public class TypedWatchingUntypedTest extends JUnitSuite {
return akka.actor.typed.javadsl.Behaviors.receive(Typed.Command.class)
.onMessage(
Typed.Pong.class,
(_ctx, message) -> {
message -> {
Adapter.stop(context, second);
return same();
})
.onSignal(akka.actor.typed.Terminated.class, (_ctx, sig) -> stopped())
.onSignal(akka.actor.typed.Terminated.class, sig -> stopped())
.build();
});
}

2
akka-actor-typed-tests/src/test/java/jdocs/akka/typed/coexistence/UntypedWatchingTypedTest.java
View file

@ -74,7 +74,7 @@ public class UntypedWatchingTypedTest extends JUnitSuite {
return Behaviors.receive(Typed.Command.class)
.onMessage(
Typed.Ping.class,
(context, message) -> {
message -> {
message.replyTo.tell(new Pong());
return same();
})

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

@ -36,14 +36,10 @@ public class SupervisionCompileOnlyTest {
public static Behavior<CounterMessage> counter(int currentValue) {
return Behaviors.receive(CounterMessage.class)
.onMessage(
Increase.class,
(context, o) -> {
return counter(currentValue + 1);
})
.onMessage(Increase.class, o -> counter(currentValue + 1))
.onMessage(
Get.class,
(context, o) -> {
o -> {
o.sender.tell(new Got(currentValue));
return Behaviors.same();
})

60
akka-actor-typed-tests/src/test/scala/docs/akka/typed/IntroSpec.scala
View file

@ -103,8 +103,9 @@ object IntroSpec {
//#hello-world-main-with-dispatchers
}
//#chatroom-actor
//#chatroom-behavior
object ChatRoom {
//#chatroom-behavior
//#chatroom-protocol
sealed trait RoomCommand
final case class GetSession(screenName: String, replyTo: ActorRef[SessionEvent]) extends RoomCommand
@ -125,7 +126,7 @@ object IntroSpec {
//#chatroom-protocol
//#chatroom-behavior
val behavior: Behavior[RoomCommand] =
def apply(): Behavior[RoomCommand] =
chatRoom(List.empty)
private def chatRoom(sessions: List[ActorRef[SessionCommand]]): Behavior[RoomCommand] =
@ -159,9 +160,32 @@ object IntroSpec {
client ! message
Behaviors.same
}
//#chatroom-behavior
}
//#chatroom-actor
//#chatroom-behavior
//#chatroom-gabbler
object Gabbler {
import ChatRoom._
def apply(): Behavior[SessionEvent] =
Behaviors.setup { context =>
Behaviors.receiveMessage {
//#chatroom-gabbler
// We document that the compiler warns about the missing handler for `SessionDenied`
case SessionDenied(reason) =>
context.log.info("cannot start chat room session: {}", reason)
Behaviors.stopped
//#chatroom-gabbler
case SessionGranted(handle) =>
handle ! PostMessage("Hello World!")
Behaviors.same
case MessagePosted(screenName, message) =>
context.log.info("message has been posted by '{}': {}", screenName, message)
Behaviors.stopped
}
}
}
//#chatroom-gabbler
}
@ -188,35 +212,13 @@ class IntroSpec extends ScalaTestWithActorTestKit with WordSpecLike {
}
"chat" in {
//#chatroom-gabbler
import ChatRoom._
val gabbler: Behavior[SessionEvent] =
Behaviors.setup { context =>
Behaviors.receiveMessage {
//#chatroom-gabbler
// We document that the compiler warns about the missing handler for `SessionDenied`
case SessionDenied(reason) =>
context.log.info("cannot start chat room session: {}", reason)
Behaviors.stopped
//#chatroom-gabbler
case SessionGranted(handle) =>
handle ! PostMessage("Hello World!")
Behaviors.same
case MessagePosted(screenName, message) =>
context.log.info("message has been posted by '{}': {}", screenName, message)
Behaviors.stopped
}
}
//#chatroom-gabbler
//#chatroom-main
val main: Behavior[NotUsed] =
Behaviors.setup { context =>
val chatRoom = context.spawn(ChatRoom.behavior, "chatroom")
val gabblerRef = context.spawn(gabbler, "gabbler")
val chatRoom = context.spawn(ChatRoom(), "chatroom")
val gabblerRef = context.spawn(Gabbler(), "gabbler")
context.watch(gabblerRef)
chatRoom ! GetSession("ol’ Gabbler", gabblerRef)
chatRoom ! ChatRoom.GetSession("ol’ Gabbler", gabblerRef)
Behaviors.receiveSignal {
case (_, Terminated(_)) =>

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

@ -17,8 +17,9 @@ import org.scalatest.WordSpecLike
object OOIntroSpec {
//#chatroom-actor
//#chatroom-behavior
object ChatRoom {
//#chatroom-behavior
//#chatroom-protocol
sealed trait RoomCommand
final case class GetSession(screenName: String, replyTo: ActorRef[SessionEvent]) extends RoomCommand
@ -39,8 +40,8 @@ object OOIntroSpec {
//#chatroom-protocol
//#chatroom-behavior
def behavior(): Behavior[RoomCommand] =
Behaviors.setup[RoomCommand](context => new ChatRoomBehavior(context))
def apply(): Behavior[RoomCommand] =
Behaviors.setup(context => new ChatRoomBehavior(context))
class ChatRoomBehavior(context: ActorContext[RoomCommand]) extends AbstractBehavior[RoomCommand] {
private var sessions: List[ActorRef[SessionCommand]] = List.empty
@ -50,7 +51,7 @@ object OOIntroSpec {
case GetSession(screenName, client) =>
// create a child actor for further interaction with the client
val ses = context.spawn(
session(context.self, screenName, client),
new SessionBehavior(context.self, screenName, client),
name = URLEncoder.encode(screenName, StandardCharsets.UTF_8.name))
client ! SessionGranted(ses)
sessions = ses :: sessions
@ -63,23 +64,48 @@ object OOIntroSpec {
}
}
private def session(
private class SessionBehavior(
room: ActorRef[PublishSessionMessage],
screenName: String,
client: ActorRef[SessionEvent]): Behavior[SessionCommand] =
Behaviors.receiveMessage {
case PostMessage(message) =>
// from client, publish to others via the room
room ! PublishSessionMessage(screenName, message)
Behaviors.same
case NotifyClient(message) =>
// published from the room
client ! message
Behaviors.same
client: ActorRef[SessionEvent])
extends AbstractBehavior[SessionCommand] {
override def onMessage(msg: SessionCommand): Behavior[SessionCommand] = {
msg match {
case PostMessage(message) =>
// from client, publish to others via the room
room ! PublishSessionMessage(screenName, message)
Behaviors.same
case NotifyClient(message) =>
// published from the room
client ! message
Behaviors.same
}
}
//#chatroom-behavior
}
}
//#chatroom-behavior
//#chatroom-gabbler
object Gabbler {
import ChatRoom._
def apply(): Behavior[SessionEvent] =
Behaviors.setup { context =>
Behaviors.receiveMessage {
case SessionDenied(reason) =>
context.log.info("cannot start chat room session: {}", reason)
Behaviors.stopped
case SessionGranted(handle) =>
handle ! PostMessage("Hello World!")
Behaviors.same
case MessagePosted(screenName, message) =>
context.log.info("message has been posted by '{}': {}", screenName, message)
Behaviors.stopped
}
}
//#chatroom-gabbler
}
//#chatroom-actor
}
@ -89,36 +115,17 @@ class OOIntroSpec extends ScalaTestWithActorTestKit with WordSpecLike {
"A chat room" must {
"chat" in {
//#chatroom-gabbler
import ChatRoom._
val gabbler =
Behaviors.setup[SessionEvent] { context =>
Behaviors.receiveMessage[SessionEvent] {
case SessionDenied(reason) =>
context.log.info("cannot start chat room session: {}", reason)
Behaviors.stopped
case SessionGranted(handle) =>
handle ! PostMessage("Hello World!")
Behaviors.same
case MessagePosted(screenName, message) =>
context.log.info("message has been posted by '{}': {}", screenName, message)
Behaviors.stopped
}
}
//#chatroom-gabbler
//#chatroom-main
val main: Behavior[String] =
Behaviors.setup { context =>
val chatRoom = context.spawn(ChatRoom.behavior(), "chatroom")
val gabblerRef = context.spawn(gabbler, "gabbler")
val chatRoom = context.spawn(ChatRoom(), "chatroom")
val gabblerRef = context.spawn(Gabbler(), "gabbler")
context.watch(gabblerRef)
Behaviors
.receiveMessagePartial[String] {
case "go" =>
chatRoom ! GetSession("ol’ Gabbler", gabblerRef)
chatRoom ! ChatRoom.GetSession("ol’ Gabbler", gabblerRef)
Behaviors.same
}
.receiveSignal {

57
akka-actor-typed/src/main/scala/akka/actor/typed/javadsl/BehaviorBuilder.scala
View file

@ -4,9 +4,11 @@
package akka.actor.typed.javadsl
import java.util.function.Supplier
import scala.annotation.tailrec
import akka.japi.function.{ Function => JFunction }
import akka.japi.function.{ Function2 => JFunction2 }
import akka.japi.function.{ Predicate => JPredicate }
import akka.annotation.InternalApi
import akka.actor.typed.Behavior
@ -39,7 +41,7 @@ final class BehaviorBuilder[T] private (messageHandlers: List[Case[T, T]], signa
* @tparam M type of message to match
* @return a new behavior builder with the specified handling appended
*/
def onMessage[M <: T](`type`: Class[M], handler: JFunction2[ActorContext[T], M, Behavior[T]]): BehaviorBuilder[T] =
def onMessage[M <: T](`type`: Class[M], handler: JFunction[M, Behavior[T]]): BehaviorBuilder[T] =
withMessage(OptionVal.Some(`type`), OptionVal.None, handler)
/**
@ -51,10 +53,7 @@ final class BehaviorBuilder[T] private (messageHandlers: List[Case[T, T]], signa
* @tparam M type of message to match
* @return a new behavior builder with the specified handling appended
*/
def onMessage[M <: T](
`type`: Class[M],
test: JPredicate[M],
handler: JFunction2[ActorContext[T], M, Behavior[T]]): BehaviorBuilder[T] =
def onMessage[M <: T](`type`: Class[M], test: JPredicate[M], handler: JFunction[M, Behavior[T]]): BehaviorBuilder[T] =
withMessage(OptionVal.Some(`type`), OptionVal.Some((t: T) => test.test(t.asInstanceOf[M])), handler)
/**
@ -67,9 +66,7 @@ final class BehaviorBuilder[T] private (messageHandlers: List[Case[T, T]], signa
* @param handler action to apply when the type matches
* @return a new behavior builder with the specified handling appended
*/
def onMessageUnchecked[M <: T](
`type`: Class[_ <: T],
handler: JFunction2[ActorContext[T], M, Behavior[T]]): BehaviorBuilder[T] =
def onMessageUnchecked[M <: T](`type`: Class[_ <: T], handler: JFunction[M, Behavior[T]]): BehaviorBuilder[T] =
withMessage[M](OptionVal.Some(`type`.asInstanceOf[Class[M]]), OptionVal.None, handler)
/**
@ -79,12 +76,12 @@ final class BehaviorBuilder[T] private (messageHandlers: List[Case[T, T]], signa
* @param handler action to apply when the message matches
* @return a new behavior builder with the specified handling appended
*/
def onMessageEquals(msg: T, handler: JFunction[ActorContext[T], Behavior[T]]): BehaviorBuilder[T] =
def onMessageEquals(msg: T, handler: Supplier[Behavior[T]]): BehaviorBuilder[T] =
withMessage[T](
OptionVal.Some(msg.getClass.asInstanceOf[Class[T]]),
OptionVal.Some(_.equals(msg)),
new JFunction2[ActorContext[T], T, Behavior[T]] {
override def apply(ctx: ActorContext[T], msg: T): Behavior[T] = handler.apply(ctx)
new JFunction[T, Behavior[T]] {
override def apply(msg: T): Behavior[T] = handler.get()
})
/**
@ -94,7 +91,7 @@ final class BehaviorBuilder[T] private (messageHandlers: List[Case[T, T]], signa
* @param handler action to apply for any message
* @return a new behavior builder with the specified handling appended
*/
def onAnyMessage(handler: JFunction2[ActorContext[T], T, Behavior[T]]): BehaviorBuilder[T] =
def onAnyMessage(handler: JFunction[T, Behavior[T]]): BehaviorBuilder[T] =
withMessage(OptionVal.None, OptionVal.None, handler)
/**
@ -105,10 +102,8 @@ final class BehaviorBuilder[T] private (messageHandlers: List[Case[T, T]], signa
* @tparam M type of signal to match
* @return a new behavior builder with the specified handling appended
*/
def onSignal[M <: Signal](
`type`: Class[M],
handler: JFunction2[ActorContext[T], M, Behavior[T]]): BehaviorBuilder[T] =
withSignal(`type`, OptionVal.None, handler.asInstanceOf[JFunction2[ActorContext[T], Signal, Behavior[T]]])
def onSignal[M <: Signal](`type`: Class[M], handler: JFunction[M, Behavior[T]]): BehaviorBuilder[T] =
withSignal(`type`, OptionVal.None, handler.asInstanceOf[JFunction[Signal, Behavior[T]]])
/**
* Add a new predicated case to the signal handling.
@ -122,11 +117,11 @@ final class BehaviorBuilder[T] private (messageHandlers: List[Case[T, T]], signa
def onSignal[M <: Signal](
`type`: Class[M],
test: JPredicate[M],
handler: JFunction2[ActorContext[T], M, Behavior[T]]): BehaviorBuilder[T] =
handler: JFunction[M, Behavior[T]]): BehaviorBuilder[T] =
withSignal(
`type`,
OptionVal.Some((t: Signal) => test.test(t.asInstanceOf[M])),
handler.asInstanceOf[JFunction2[ActorContext[T], Signal, Behavior[T]]])
handler.asInstanceOf[JFunction[Signal, Behavior[T]]])
/**
* Add a new case to the signal handling matching equal signals.
@ -135,17 +130,17 @@ final class BehaviorBuilder[T] private (messageHandlers: List[Case[T, T]], signa
* @param handler action to apply when the message matches
* @return a new behavior builder with the specified handling appended
*/
def onSignalEquals(signal: Signal, handler: Function[ActorContext[T], Behavior[T]]): BehaviorBuilder[T] =
withSignal(signal.getClass, OptionVal.Some(_.equals(signal)), new JFunction2[ActorContext[T], Signal, Behavior[T]] {
override def apply(ctx: ActorContext[T], signal: Signal): Behavior[T] = {
handler.apply(ctx)
def onSignalEquals(signal: Signal, handler: Supplier[Behavior[T]]): BehaviorBuilder[T] =
withSignal(signal.getClass, OptionVal.Some(_.equals(signal)), new JFunction[Signal, Behavior[T]] {
override def apply(signal: Signal): Behavior[T] = {
handler.get()
}
})
private def withMessage[M <: T](
clazz: OptionVal[Class[M]],
test: OptionVal[M => Boolean],
handler: JFunction2[ActorContext[T], M, Behavior[T]]): BehaviorBuilder[T] = {
handler: JFunction[M, Behavior[T]]): BehaviorBuilder[T] = {
val newCase = Case(clazz, test, handler)
new BehaviorBuilder[T](newCase.asInstanceOf[Case[T, T]] +: messageHandlers, signalHandlers)
}
@ -153,7 +148,7 @@ final class BehaviorBuilder[T] private (messageHandlers: List[Case[T, T]], signa
private def withSignal[M <: Signal](
`type`: Class[M],
test: OptionVal[Signal => Boolean],
handler: JFunction2[ActorContext[T], Signal, Behavior[T]]): BehaviorBuilder[T] = {
handler: JFunction[Signal, Behavior[T]]): BehaviorBuilder[T] = {
new BehaviorBuilder[T](
messageHandlers,
Case(OptionVal.Some(`type`), test, handler).asInstanceOf[Case[T, Signal]] +: signalHandlers)
@ -170,7 +165,7 @@ object BehaviorBuilder {
private[javadsl] final case class Case[BT, MT](
`type`: OptionVal[Class[_ <: MT]],
test: OptionVal[MT => Boolean],
handler: JFunction2[ActorContext[BT], MT, Behavior[BT]])
handler: JFunction[MT, Behavior[BT]])
/**
* @return new empty immutable behavior builder.
@ -187,18 +182,18 @@ object BehaviorBuilder {
private final class BuiltBehavior[T](messageHandlers: List[Case[T, T]], signalHandlers: List[Case[T, Signal]])
extends ExtensibleBehavior[T] {
override def receive(ctx: TypedActorContext[T], msg: T): Behavior[T] = receive(ctx.asJava, msg, messageHandlers)
override def receive(ctx: TypedActorContext[T], msg: T): Behavior[T] = receive(msg, messageHandlers)
override def receiveSignal(ctx: TypedActorContext[T], msg: Signal): Behavior[T] =
receive(ctx.asJava, msg, signalHandlers)
receive(msg, signalHandlers)
@tailrec
private def receive[M](ctx: ActorContext[T], msg: M, handlers: List[Case[T, M]]): Behavior[T] =
private def receive[M](msg: M, handlers: List[Case[T, M]]): Behavior[T] =
handlers match {
case Case(cls, predicate, handler) :: tail =>
if ((cls.isEmpty || cls.get.isAssignableFrom(msg.getClass)) && (predicate.isEmpty || predicate.get.apply(msg)))
handler(ctx, msg)
else receive(ctx, msg, tail)
handler(msg)
else receive(msg, tail)
case Nil =>
Behaviors.unhandled[T]
}

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

@ -43,14 +43,10 @@ public class ShardingCompileOnlyTest {
public static Behavior<CounterCommand> counter(String entityId, Integer value) {
return Behaviors.receive(CounterCommand.class)
.onMessage(
Increment.class,
(ctx, msg) -> {
return counter(entityId, value + 1);
})
.onMessage(Increment.class, msg -> counter(entityId, value + 1))
.onMessage(
GetValue.class,
(ctx, msg) -> {
msg -> {
msg.replyTo.tell(value);
return Behaviors.same();
})
@ -74,32 +70,30 @@ public class ShardingCompileOnlyTest {
private static Behavior<CounterCommand> counter2(
ActorRef<ClusterSharding.ShardCommand> shard, String entityId, Integer value) {
return Behaviors.receive(CounterCommand.class)
.onMessage(
Increment.class,
(ctx, msg) -> {
return counter(entityId, value + 1);
})
.onMessage(
GetValue.class,
(ctx, msg) -> {
msg.replyTo.tell(value);
return Behaviors.same();
})
.onMessage(
Idle.class,
(ctx, msg) -> {
// after receive timeout
shard.tell(new ClusterSharding.Passivate<>(ctx.getSelf()));
return Behaviors.same();
})
.onMessage(
GoodByeCounter.class,
(ctx, msg) -> {
// the stopMessage, used for rebalance and passivate
return Behaviors.stopped();
})
.build();
return Behaviors.setup(
context ->
Behaviors.receive(CounterCommand.class)
.onMessage(Increment.class, msg -> counter(entityId, value + 1))
.onMessage(
GetValue.class,
msg -> {
msg.replyTo.tell(value);
return Behaviors.same();
})
.onMessage(
Idle.class,
msg -> {
// after receive timeout
shard.tell(new ClusterSharding.Passivate<>(context.getSelf()));
return Behaviors.same();
})
.onMessage(
GoodByeCounter.class,
msg -> {
// the stopMessage, used for rebalance and passivate
return Behaviors.stopped();
})
.build());
}
// #counter-passivate

50
akka-cluster-typed/src/test/java/jdocs/akka/cluster/typed/ReceptionistExample.java
View file

@ -16,11 +16,18 @@ import akka.actor.typed.ActorSystem;
public class ReceptionistExample {
public
// #ping-service
public static class PingService {
static class PingService {
private final ActorContext<Ping> context;
static final ServiceKey<Ping> pingServiceKey = ServiceKey.create(Ping.class, "pingService");
private PingService(ActorContext<Ping> context) {
this.context = context;
}
public static class Pong {}
public static class Ping {
@ -31,7 +38,7 @@ public class ReceptionistExample {
}
}
static Behavior<Ping> createBehavior() {
public static Behavior<Ping> createBehavior() {
return Behaviors.setup(
context -> {
context
@ -39,11 +46,15 @@ public class ReceptionistExample {
.receptionist()
.tell(Receptionist.register(pingServiceKey, context.getSelf()));
return Behaviors.receive(Ping.class).onMessage(Ping.class, PingService::onPing).build();
return new PingService(context).behavior();
});
}
private static Behavior<Ping> onPing(ActorContext<Ping> context, Ping msg) {
private Behavior<Ping> behavior() {
return Behaviors.receive(Ping.class).onMessage(Ping.class, this::onPing).build();
}
private Behavior<Ping> onPing(Ping msg) {
context.getLog().info("Pinged by {}", msg.replyTo);
msg.replyTo.tell(new Pong());
return Behaviors.same();
@ -51,20 +62,33 @@ public class ReceptionistExample {
}
// #ping-service
public
// #pinger
public static class Pinger {
static Behavior<PingService.Pong> createBehavior(ActorRef<PingService.Ping> pingService) {
static class Pinger {
private final ActorContext<PingService.Pong> context;
private final ActorRef<PingService.Ping> pingService;
private Pinger(ActorContext<PingService.Pong> context, ActorRef<PingService.Ping> pingService) {
this.context = context;
this.pingService = pingService;
}
public static Behavior<PingService.Pong> createBehavior(
ActorRef<PingService.Ping> pingService) {
return Behaviors.setup(
(ctx) -> {
ctx -> {
pingService.tell(new PingService.Ping(ctx.getSelf()));
return Behaviors.receive(PingService.Pong.class)
.onMessage(PingService.Pong.class, Pinger::onPong)
.build();
return new Pinger(ctx, pingService).behavior();
});
}
private static Behavior<PingService.Pong> onPong(
ActorContext<PingService.Pong> context, PingService.Pong msg) {
private Behavior<PingService.Pong> behavior() {
return Behaviors.receive(PingService.Pong.class)
.onMessage(PingService.Pong.class, this::onPong)
.build();
}
private Behavior<PingService.Pong> onPong(PingService.Pong msg) {
context.getLog().info("{} was ponged!!", context.getSelf());
return Behaviors.stopped();
}
@ -85,7 +109,7 @@ public class ReceptionistExample {
return Behaviors.receive(Object.class)
.onMessage(
Receptionist.Listing.class,
(c, msg) -> {
msg -> {
msg.getServiceInstances(PingService.pingServiceKey)
.forEach(
pingService ->

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

@ -31,14 +31,14 @@ public class SingletonCompileOnlyTest {
public static Behavior<CounterCommand> counter(String entityId, Integer value) {
return Behaviors.receive(CounterCommand.class)
.onMessage(Increment.class, (ctx, msg) -> counter(entityId, value + 1))
.onMessage(Increment.class, msg -> counter(entityId, value + 1))
.onMessage(
GetValue.class,
(ctx, msg) -> {
msg -> {
msg.replyTo.tell(value);
return Behaviors.same();
})
.onMessage(GoodByeCounter.class, (ctx, msg) -> Behaviors.stopped())
.onMessage(GoodByeCounter.class, msg -> Behaviors.stopped())
.build();
}
// #counter

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

@ -327,6 +327,9 @@ made before finalizing the APIs. Compared to Akka 2.5.x the source incompatible
* The `request` parameter in Distributed Data commands was removed, in favor of using `ask`.
* Removed `Behavior.same`, `Behavior.unhandled`, `Behavior.stopped`, `Behavior.empty`, and `Behavior.ignore` since
they were redundant with corresponding @scala[scaladsl.Behaviors.x]@java[javadsl.Behaviors.x].
* `ActorContext` parameter removed in `javadsl.ReceiveBuilder` for the functional style in Java. Use `Behaviors.setup`
to retrieve `ActorContext`, and use an enclosing class to hold initialization parameters and `ActorContext`.
#### Akka Typed Stream API changes

2
akka-docs/src/main/paradox/typed/actors.md
View file

@ -366,7 +366,7 @@ screen name.
To implement the logic where we spawn a child for the session we need access
to the `ActorContext`. This is injected as a constructor parameter upon creation
of the behavior, note how we combine the `AbstractBehavior` with `Behaviors.setup`
to do this in the `behavior` method.
to do this in the @scala[`apply`]@java[`create`] factory method.
The behavior that we declare here can handle both subtypes of `RoomCommand`.
`GetSession` has been explained already and the