This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 5ddbe8c2d5 feat: add logging variants for Java DSL stash and timer
base classes (#3387)
5ddbe8c2d5 is described below
commit 5ddbe8c2d550b8e1c7dab00005e9fc80c15d0a02
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sun Jul 26 23:53:35 2026 +0800
feat: add logging variants for Java DSL stash and timer base classes (#3387)
Motivation:
Java does not support multiple inheritance or trait mixin, so users
cannot combine AbstractActorWithStash/AbstractActorWithTimers with
ActorLogging. This forces Java users to manually implement logging
boilerplate when they need both stash/timer support and logging.
Modification:
Add 8 new abstract base classes that combine ActorLogging with each
stash and timer variant:
- AbstractLoggingActorWithStash / UntypedAbstractLoggingActorWithStash
- AbstractLoggingActorWithUnboundedStash / Untyped variant
- AbstractLoggingActorWithUnrestrictedStash / Untyped variant
- AbstractLoggingActorWithTimers / UntypedAbstractLoggingActorWithTimers
Result:
Java users can now extend a single base class to get both stash/timer
support and logging, matching the convenience Scala users get via
trait mixin.
Tests:
- sbt "actor-tests / Test / testOnly
org.apache.pekko.actor.LoggingVariantActorsTest"
Passed: Total 9, Failed 0, Errors 0, Passed 9
- sbt "actor / mimaReportBinaryIssues" - success (pure additions)
References:
Fixes #2520
---
.../pekko/actor/LoggingVariantActorsTest.java | 512 +++++++++++++++++++++
.../org/apache/pekko/actor/AbstractActor.scala | 75 +++
.../main/scala/org/apache/pekko/actor/Timers.scala | 32 ++
3 files changed, 619 insertions(+)
diff --git
a/actor-tests/src/test/java/org/apache/pekko/actor/LoggingVariantActorsTest.java
b/actor-tests/src/test/java/org/apache/pekko/actor/LoggingVariantActorsTest.java
new file mode 100644
index 0000000000..ca73c7e56d
--- /dev/null
+++
b/actor-tests/src/test/java/org/apache/pekko/actor/LoggingVariantActorsTest.java
@@ -0,0 +1,512 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.actor;
+
+import java.time.Duration;
+
+import scala.concurrent.duration.FiniteDuration;
+
+import com.typesafe.config.ConfigFactory;
+
+import org.apache.pekko.event.Logging;
+import org.apache.pekko.testkit.PekkoJUnitJupiterActorSystemResource;
+import org.apache.pekko.testkit.TestProbe;
+import org.apache.pekko.testkit.javadsl.EventFilter;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class LoggingVariantActorsTest {
+
+ @RegisterExtension
+ static PekkoJUnitJupiterActorSystemResource actorSystemResource =
+ new PekkoJUnitJupiterActorSystemResource(
+ "LoggingVariantActorsTest",
+ ConfigFactory.parseString("pekko.loggers =
[org.apache.pekko.testkit.TestEventListener]")
+ .withFallback(ActorWithBoundedStashSpec.testConf()));
+
+ private final ActorSystem system = actorSystemResource.getSystem();
+
+ // --- stash actor implementations ---
+
+ public static class WithStashAndLogging extends
AbstractLoggingActorWithStash {
+ int count = 0;
+
+ @Override
+ public Receive createReceive() {
+ return receiveBuilder()
+ .matchEquals(
+ "fail",
+ s -> {
+ throw new RuntimeException("restart requested");
+ })
+ .match(
+ String.class,
+ s -> {
+ if (count < 0) {
+ log().info("Replying with length of: {}", s);
+ getSender().tell(s.length(), getSelf());
+ } else if (count == 2) {
+ unstashAll();
+ count = -1;
+ } else {
+ stash();
+ count += 1;
+ }
+ })
+ .build();
+ }
+ }
+
+ public static class UntypedWithStashAndLogging extends
UntypedAbstractLoggingActorWithStash {
+ int count = 0;
+
+ @Override
+ public void onReceive(Object msg) throws Exception {
+ if ("fail".equals(msg)) {
+ throw new RuntimeException("restart requested");
+ } else if (msg instanceof String s) {
+ if (count < 0) {
+ log().info("Replying with length of: {}", s);
+ getSender().tell(s.length(), getSelf());
+ } else if (count == 2) {
+ unstashAll();
+ count = -1;
+ } else {
+ stash();
+ count += 1;
+ }
+ } else {
+ unhandled(msg);
+ }
+ }
+ }
+
+ public static class WithUnboundedStashAndLogging extends
AbstractLoggingActorWithUnboundedStash {
+ int count = 0;
+
+ @Override
+ public Receive createReceive() {
+ return receiveBuilder()
+ .match(
+ String.class,
+ s -> {
+ if (count < 0) {
+ log().info("Replying with length of: {}", s);
+ getSender().tell(s.length(), getSelf());
+ } else if (count == 2) {
+ unstashAll();
+ count = -1;
+ } else {
+ stash();
+ count += 1;
+ }
+ })
+ .build();
+ }
+ }
+
+ public static class UntypedWithUnboundedStashAndLogging
+ extends UntypedAbstractLoggingActorWithUnboundedStash {
+ int count = 0;
+
+ @Override
+ public void onReceive(Object msg) throws Exception {
+ if (msg instanceof String s) {
+ if (count < 0) {
+ log().info("Replying with length of: {}", s);
+ getSender().tell(s.length(), getSelf());
+ } else if (count == 2) {
+ unstashAll();
+ count = -1;
+ } else {
+ stash();
+ count += 1;
+ }
+ } else {
+ unhandled(msg);
+ }
+ }
+ }
+
+ public static class WithUnrestrictedStashAndLogging
+ extends AbstractLoggingActorWithUnrestrictedStash {
+ int count = 0;
+
+ @Override
+ public Receive createReceive() {
+ return receiveBuilder()
+ .match(
+ String.class,
+ s -> {
+ if (count < 0) {
+ log().info("Replying with length of: {}", s);
+ getSender().tell(s.length(), getSelf());
+ } else if (count == 2) {
+ unstashAll();
+ count = -1;
+ } else {
+ stash();
+ count += 1;
+ }
+ })
+ .build();
+ }
+ }
+
+ public static class UntypedWithUnrestrictedStashAndLogging
+ extends UntypedAbstractLoggingActorWithUnrestrictedStash {
+ int count = 0;
+
+ @Override
+ public void onReceive(Object msg) throws Exception {
+ if (msg instanceof String s) {
+ if (count < 0) {
+ log().info("Replying with length of: {}", s);
+ getSender().tell(s.length(), getSelf());
+ } else if (count == 2) {
+ unstashAll();
+ count = -1;
+ } else {
+ stash();
+ count += 1;
+ }
+ } else {
+ unhandled(msg);
+ }
+ }
+ }
+
+ // --- timer actor implementations ---
+
+ public static class WithTimersAndLogging extends
AbstractLoggingActorWithTimers {
+ private ActorRef replyTo;
+
+ @Override
+ public Receive createReceive() {
+ return receiveBuilder()
+ .matchEquals(
+ "tick",
+ s -> {
+ log().info("Timer fired");
+ replyTo.tell("done", getSelf());
+ })
+ .matchEquals(
+ "cancel",
+ s -> {
+ log().info("Cancelling timer");
+ getTimers().cancel("key");
+ replyTo.tell("cancelled", getSelf());
+ })
+ .match(
+ String.class,
+ s -> {
+ log().info("Received: {}", s);
+ replyTo = getSender();
+ getTimers().startSingleTimer("key", "tick",
Duration.ofMillis(200));
+ })
+ .build();
+ }
+ }
+
+ public static class UntypedWithTimersAndLogging extends
UntypedAbstractLoggingActorWithTimers {
+ private ActorRef replyTo;
+
+ @Override
+ public void onReceive(Object msg) throws Exception {
+ if ("tick".equals(msg)) {
+ log().info("Timer fired");
+ replyTo.tell("done", getSelf());
+ } else if ("cancel".equals(msg)) {
+ log().info("Cancelling timer");
+ getTimers().cancel("key");
+ replyTo.tell("cancelled", getSelf());
+ } else if (msg instanceof String s) {
+ log().info("Received: {}", s);
+ replyTo = getSender();
+ getTimers().startSingleTimer("key", "tick", Duration.ofMillis(200));
+ } else {
+ unhandled(msg);
+ }
+ }
+ }
+
+ // --- stash functional tests ---
+
+ private void testStashFunctionality(Props props) {
+ ActorRef ref = system.actorOf(props);
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "Hello");
+ probe.send(ref, "Hello2");
+ probe.send(ref, "Hello12");
+ probe.expectMsg(5);
+ probe.expectMsg(6);
+ }
+
+ @Test
+ public void mustSupportStashWithLogging() {
+ testStashFunctionality(Props.create(WithStashAndLogging.class));
+ }
+
+ @Test
+ public void mustSupportUntypedStashWithLogging() {
+ testStashFunctionality(Props.create(UntypedWithStashAndLogging.class));
+ }
+
+ @Test
+ public void mustSupportUnboundedStashWithLogging() {
+ testStashFunctionality(Props.create(WithUnboundedStashAndLogging.class));
+ }
+
+ @Test
+ public void mustSupportUntypedUnboundedStashWithLogging() {
+
testStashFunctionality(Props.create(UntypedWithUnboundedStashAndLogging.class));
+ }
+
+ @Test
+ public void mustSupportUnrestrictedStashWithLogging() {
+ testStashFunctionality(
+ Props.create(WithUnrestrictedStashAndLogging.class)
+ .withMailbox("pekko.actor.mailbox.unbounded-deque-based"));
+ }
+
+ @Test
+ public void mustSupportUntypedUnrestrictedStashWithLogging() {
+ testStashFunctionality(
+ Props.create(UntypedWithUnrestrictedStashAndLogging.class)
+ .withMailbox("pekko.actor.mailbox.unbounded-deque-based"));
+ }
+
+ // --- timer functional tests ---
+
+ @Test
+ public void mustSupportTimersWithLogging() {
+ ActorRef ref = system.actorOf(Props.create(WithTimersAndLogging.class));
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "start");
+ probe.expectMsg(FiniteDuration.create(3, "seconds"), "done");
+ }
+
+ @Test
+ public void mustSupportUntypedTimersWithLogging() {
+ ActorRef ref =
system.actorOf(Props.create(UntypedWithTimersAndLogging.class));
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "start");
+ probe.expectMsg(FiniteDuration.create(3, "seconds"), "done");
+ }
+
+ @Test
+ public void mustCancelTimerWithLogging() {
+ ActorRef ref = system.actorOf(Props.create(WithTimersAndLogging.class));
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "start");
+ probe.send(ref, "cancel");
+ probe.expectMsg(FiniteDuration.create(3, "seconds"), "cancelled");
+ probe.expectNoMessage(FiniteDuration.create(500, "millis"));
+ }
+
+ @Test
+ public void mustCancelUntypedTimerWithLogging() {
+ ActorRef ref =
system.actorOf(Props.create(UntypedWithTimersAndLogging.class));
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "start");
+ probe.send(ref, "cancel");
+ probe.expectMsg(FiniteDuration.create(3, "seconds"), "cancelled");
+ probe.expectNoMessage(FiniteDuration.create(500, "millis"));
+ }
+
+ // --- logging event assertions for all variants ---
+
+ @Test
+ public void mustEmitLogEventsFromStashVariant() {
+ new EventFilter(Logging.Info.class, system)
+ .startsWith("Replying with length of")
+ .occurrences(2)
+ .intercept(
+ () -> {
+ testStashFunctionality(Props.create(WithStashAndLogging.class));
+ return null;
+ });
+ }
+
+ @Test
+ public void mustEmitLogEventsFromUntypedStashVariant() {
+ new EventFilter(Logging.Info.class, system)
+ .startsWith("Replying with length of")
+ .occurrences(2)
+ .intercept(
+ () -> {
+
testStashFunctionality(Props.create(UntypedWithStashAndLogging.class));
+ return null;
+ });
+ }
+
+ @Test
+ public void mustEmitLogEventsFromUnboundedStashVariant() {
+ new EventFilter(Logging.Info.class, system)
+ .startsWith("Replying with length of")
+ .occurrences(2)
+ .intercept(
+ () -> {
+
testStashFunctionality(Props.create(WithUnboundedStashAndLogging.class));
+ return null;
+ });
+ }
+
+ @Test
+ public void mustEmitLogEventsFromUntypedUnboundedStashVariant() {
+ new EventFilter(Logging.Info.class, system)
+ .startsWith("Replying with length of")
+ .occurrences(2)
+ .intercept(
+ () -> {
+
testStashFunctionality(Props.create(UntypedWithUnboundedStashAndLogging.class));
+ return null;
+ });
+ }
+
+ @Test
+ public void mustEmitLogEventsFromUnrestrictedStashVariant() {
+ new EventFilter(Logging.Info.class, system)
+ .startsWith("Replying with length of")
+ .occurrences(2)
+ .intercept(
+ () -> {
+ testStashFunctionality(
+ Props.create(WithUnrestrictedStashAndLogging.class)
+
.withMailbox("pekko.actor.mailbox.unbounded-deque-based"));
+ return null;
+ });
+ }
+
+ @Test
+ public void mustEmitLogEventsFromUntypedUnrestrictedStashVariant() {
+ new EventFilter(Logging.Info.class, system)
+ .startsWith("Replying with length of")
+ .occurrences(2)
+ .intercept(
+ () -> {
+ testStashFunctionality(
+ Props.create(UntypedWithUnrestrictedStashAndLogging.class)
+
.withMailbox("pekko.actor.mailbox.unbounded-deque-based"));
+ return null;
+ });
+ }
+
+ @Test
+ public void mustEmitLogEventsFromTimerVariant() {
+ new EventFilter(Logging.Info.class, system)
+ .startsWith("Timer fired")
+ .occurrences(1)
+ .intercept(
+ () -> {
+ ActorRef ref =
system.actorOf(Props.create(WithTimersAndLogging.class));
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "start");
+ probe.expectMsg(FiniteDuration.create(3, "seconds"), "done");
+ return null;
+ });
+ }
+
+ @Test
+ public void mustEmitLogEventsFromUntypedTimerVariant() {
+ new EventFilter(Logging.Info.class, system)
+ .startsWith("Timer fired")
+ .occurrences(1)
+ .intercept(
+ () -> {
+ ActorRef ref =
system.actorOf(Props.create(UntypedWithTimersAndLogging.class));
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "start");
+ probe.expectMsg(FiniteDuration.create(3, "seconds"), "done");
+ return null;
+ });
+ }
+
+ // --- stash restart behavior (unstash on preRestart) ---
+
+ @Test
+ public void mustUnstashOnRestartForStashVariant() {
+ ActorRef ref = system.actorOf(Props.create(WithStashAndLogging.class));
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "Hello");
+ probe.send(ref, "fail");
+ probe.send(ref, "Hello2");
+ probe.send(ref, "Hello12");
+ probe.expectMsg(FiniteDuration.create(10, "seconds"), 5);
+ }
+
+ @Test
+ public void mustUnstashOnRestartForUntypedStashVariant() {
+ ActorRef ref =
system.actorOf(Props.create(UntypedWithStashAndLogging.class));
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "Hello");
+ probe.send(ref, "fail");
+ probe.send(ref, "Hello2");
+ probe.send(ref, "Hello12");
+ probe.expectMsg(FiniteDuration.create(10, "seconds"), 5);
+ }
+
+ // --- log adapter accessibility ---
+
+ public static class LogAccessorCheck extends AbstractLoggingActorWithStash {
+ @Override
+ public Receive createReceive() {
+ return receiveBuilder()
+ .matchEquals(
+ "check",
+ s -> {
+ assertNotNull(log());
+ assertTrue(log().isInfoEnabled());
+ getSender().tell("ok", getSelf());
+ })
+ .build();
+ }
+ }
+
+ public static class UntypedLogAccessorCheck extends
UntypedAbstractLoggingActorWithTimers {
+ @Override
+ public void onReceive(Object msg) throws Exception {
+ if ("check".equals(msg)) {
+ assertNotNull(log());
+ assertNotNull(getTimers());
+ assertTrue(log().isInfoEnabled());
+ getSender().tell("ok", getSelf());
+ } else {
+ unhandled(msg);
+ }
+ }
+ }
+
+ @Test
+ public void mustProvideLogAccessor() {
+ ActorRef ref = system.actorOf(Props.create(LogAccessorCheck.class));
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "check");
+ probe.expectMsg(FiniteDuration.create(3, "seconds"), "ok");
+ }
+
+ @Test
+ public void mustProvideLogAndTimersAccessors() {
+ ActorRef ref = system.actorOf(Props.create(UntypedLogAccessorCheck.class));
+ final TestProbe probe = new TestProbe(system);
+ probe.send(ref, "check");
+ probe.expectMsg(FiniteDuration.create(3, "seconds"), "ok");
+ }
+}
diff --git a/actor/src/main/scala/org/apache/pekko/actor/AbstractActor.scala
b/actor/src/main/scala/org/apache/pekko/actor/AbstractActor.scala
index 7bb3ce71a0..d8ed444a01 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/AbstractActor.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/AbstractActor.scala
@@ -492,3 +492,78 @@ abstract class AbstractActorWithUnrestrictedStash extends
AbstractActor with Abs
* @since 2.0.0
*/
abstract class UntypedAbstractActorWithUnrestrictedStash extends
UntypedAbstractActor with AbstractActorStashSupport
+
+/**
+ * Java API: compatible with lambda expressions
+ *
+ * Actor base class that mixes in logging and a stash.
+ * By default requests a `DequeBasedMessageQueueSemantics` mailbox.
+ *
+ * @see [[pekko.actor.AbstractActorWithStash]] for details on how `Stash`
works.
+ *
+ * @since 2.0.0
+ */
+abstract class AbstractLoggingActorWithStash extends AbstractActor with
AbstractActorStashSupport
+ with RequiresMessageQueue[DequeBasedMessageQueueSemantics] with
ActorLogging
+
+/**
+ * Java API: compatible with lambda expressions
+ *
+ * Actor base class that mixes in logging and a stash.
+ * By default requests a `DequeBasedMessageQueueSemantics` mailbox.
+ *
+ * @see [[pekko.actor.AbstractActorWithStash]] for details on how `Stash`
works.
+ *
+ * @since 2.0.0
+ */
+abstract class UntypedAbstractLoggingActorWithStash extends
UntypedAbstractActor with AbstractActorStashSupport
+ with RequiresMessageQueue[DequeBasedMessageQueueSemantics] with
ActorLogging
+
+/**
+ * Java API: compatible with lambda expressions
+ *
+ * Actor base class that mixes in logging and an unbounded stash.
+ *
+ * @see [[pekko.actor.AbstractActorWithUnboundedStash]] for details on how
`Stash` works.
+ *
+ * @since 2.0.0
+ */
+abstract class AbstractLoggingActorWithUnboundedStash extends AbstractActor
with AbstractActorStashSupport
+ with RequiresMessageQueue[UnboundedDequeBasedMessageQueueSemantics] with
ActorLogging
+
+/**
+ * Java API: compatible with lambda expressions
+ *
+ * Actor base class that mixes in logging and an unbounded stash.
+ *
+ * @see [[pekko.actor.AbstractActorWithUnboundedStash]] for details on how
`Stash` works.
+ *
+ * @since 2.0.0
+ */
+abstract class UntypedAbstractLoggingActorWithUnboundedStash extends
UntypedAbstractActor
+ with AbstractActorStashSupport
+ with RequiresMessageQueue[UnboundedDequeBasedMessageQueueSemantics] with
ActorLogging
+
+/**
+ * Java API: compatible with lambda expressions
+ *
+ * Actor base class that mixes in logging and an unrestricted stash.
+ *
+ * @see [[pekko.actor.AbstractActorWithUnrestrictedStash]] for details on how
`Stash` works.
+ *
+ * @since 2.0.0
+ */
+abstract class AbstractLoggingActorWithUnrestrictedStash extends AbstractActor
with AbstractActorStashSupport
+ with ActorLogging
+
+/**
+ * Java API: compatible with lambda expressions
+ *
+ * Actor base class that mixes in logging and an unrestricted stash.
+ *
+ * @see [[pekko.actor.AbstractActorWithUnrestrictedStash]] for details on how
`Stash` works.
+ *
+ * @since 2.0.0
+ */
+abstract class UntypedAbstractLoggingActorWithUnrestrictedStash
+ extends UntypedAbstractActor with AbstractActorStashSupport with
ActorLogging
diff --git a/actor/src/main/scala/org/apache/pekko/actor/Timers.scala
b/actor/src/main/scala/org/apache/pekko/actor/Timers.scala
index c75574be4d..7cbc1bcebc 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/Timers.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/Timers.scala
@@ -98,6 +98,38 @@ abstract class UntypedAbstractActorWithTimers extends
UntypedAbstractActor with
final def getTimers: TimerScheduler = timers
}
+/**
+ * Java API: Support for scheduled `self` messages via [[TimerScheduler]],
with logging.
+ *
+ * Timers are bound to the lifecycle of the actor that owns it,
+ * and thus are cancelled automatically when it is restarted or stopped.
+ *
+ * @since 2.0.0
+ */
+abstract class AbstractLoggingActorWithTimers extends AbstractActor with
Timers with ActorLogging {
+
+ /**
+ * Start and cancel timers via the enclosed `TimerScheduler`.
+ */
+ final def getTimers: TimerScheduler = timers
+}
+
+/**
+ * Java API: Support for scheduled `self` messages via [[TimerScheduler]],
with logging.
+ *
+ * Timers are bound to the lifecycle of the actor that owns it,
+ * and thus are cancelled automatically when it is restarted or stopped.
+ *
+ * @since 2.0.0
+ */
+abstract class UntypedAbstractLoggingActorWithTimers extends
UntypedAbstractActor with Timers with ActorLogging {
+
+ /**
+ * Start and cancel timers via the enclosed `TimerScheduler`.
+ */
+ final def getTimers: TimerScheduler = timers
+}
+
/**
* Support for scheduled `self` messages in an actor.
* It is used by mixing in trait `Timers` in Scala or extending
`AbstractActorWithTimers`
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
