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 2e5c450633 feat: add timeout duration to ReceiveTimeout message (#3399)
2e5c450633 is described below
commit 2e5c45063393b65a6af3300f74650fda6ed147ea
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Fri Aug 7 20:48:27 2026 +0800
feat: add timeout duration to ReceiveTimeout message (#3399)
* feat: add timeout duration to ReceiveTimeout message
Motivation:
ReceiveTimeout was a case object singleton carrying no information about
the configured timeout duration, making it impossible to log or inspect
the timeout value from the message handler without accessing
context.receiveTimeout separately (#2569).
Modification:
Convert ReceiveTimeout from a case object to a final case class with a
`timeout: FiniteDuration` field. The scheduler now sends
ReceiveTimeout(duration) instead of the singleton. All pattern matches
updated from stable identifier patterns to type patterns. Java API
updated from matchEquals(getInstance()) to match(ReceiveTimeout.class).
Result:
Users can now access the timeout duration directly from the message:
case timeout: ReceiveTimeout => log.info("timeout: {}", timeout.timeout)
* chore: apply scalafmt and fix MiMa filters
Apply code formatting and update MiMa exclusion filters to match the
exact binary compatibility problems reported by CI.
* fix: add Scala 3 MiMa filters for ReceiveTimeout and cover getTimeout
Motivation:
The Binary Compatibility CI job failed on the Scala 3 lane: converting
ReceiveTimeout from case object to case class produced three unfiltered
problems for pekko-actor_3 (fromProduct result type, productElementNames
and productIterator generic signatures). The new Java API getTimeout also
lacked test coverage.
Modification:
Add the three Scala 3 MiMa exclusion filters reported by CI and assert
getTimeout in the ReceiveTimeoutSpec timeout-duration test.
Result:
MiMa passes on both Scala 2.13 and Scala 3 and the Java API is covered.
Tests:
- sbt "actor-tests / Test / testOnly
org.apache.pekko.actor.ReceiveTimeoutSpec" - 15 tests pass
- sbt "++2.13.18!" "actor/mimaReportBinaryIssues" "++3.3.8!"
"actor/mimaReportBinaryIssues" - pass
- sbt checkMimaFilterDirectories - pass
References:
Refs #2569
* fix: use scala.jdk.DurationConverters in ReceiveTimeout.getTimeout
Motivation:
Address review feedback on #3399 to use the standard Scala/Java duration
conversion, and document the ReceiveTimeout breaking change for 2.x users.
Modification:
- Replace java.time.Duration.ofNanos(timeout.toNanos) with timeout.toJava
via scala.jdk.DurationConverters in Actor.scala
- Add a ReceiveTimeout entry to migration-guide-1.x-2.x.md
Result:
The Java API conversion follows the established codebase convention and
the 2.x migration guide documents the pattern matching and getInstance
breaking changes.
Tests:
- sbt "actor-tests / Test / testOnly
org.apache.pekko.actor.ReceiveTimeoutSpec" - 15 tests passed
- sbt "actor-typed-tests / Test / testOnly
org.apache.pekko.actor.typed.CancelReceiveTimeoutSpec" - passed
- sbt actor/mimaReportBinaryIssues - no issues
- scalafmt --mode diff-ref=origin/main - no extra changes
- git diff --check - clean
References:
Refs #3399
---
.../test/java/org/apache/pekko/actor/JavaAPI.java | 1 -
.../apache/pekko/actor/ReceiveTimeoutSpec.scala | 57 ++++++++++++++--------
.../actor/typed/CancelReceiveTimeoutSpec.scala | 2 +-
.../typed/internal/adapter/ActorAdapter.scala | 2 +-
.../receive-timeout-with-duration.excludes | 43 ++++++++++++++++
.../main/scala/org/apache/pekko/actor/Actor.scala | 13 +++--
.../pekko/actor/dungeon/ReceiveTimeout.scala | 2 +-
.../scala/org/apache/pekko/io/TcpConnection.scala | 2 +-
.../apache/pekko/io/TcpOutgoingConnection.scala | 6 +--
.../pekko/actor/ReceiveTimeoutBenchmark.scala | 2 +-
.../cluster/metrics/sample/StatsService.scala | 2 +-
.../pekko/cluster/sharding/ShardCoordinator.scala | 6 +--
.../cluster/sharding/ClusterShardingSpec.scala | 12 ++---
.../pekko/cluster/client/ClusterClient.scala | 4 +-
.../org/apache/pekko/cluster/SeedNodeProcess.scala | 6 +--
.../apache/pekko/cluster/ddata/Replicator.scala | 10 ++--
.../paradox/migration/migration-guide-1.x-2.x.md | 6 +++
.../java/jdocs/actor/FaultHandlingDocSample.java | 4 +-
.../java/jdocs/sharding/ClusterShardingTest.java | 2 +-
docs/src/test/scala/docs/actor/ActorDocSpec.scala | 2 +-
.../scala/docs/actor/FaultHandlingDocSample.scala | 2 +-
.../circuitbreaker/CircuitBreakerDocSpec.scala | 2 +-
.../scala/docs/cluster/FactorialFrontend.scala | 2 +-
.../persistence/journal/AsyncWriteProxy.scala | 2 +-
.../journal/PersistencePluginProxy.scala | 6 +--
.../pekko/remote/artery/MaxThroughputSpec.scala | 2 +-
.../apache/pekko/testkit/TestActorRefSpec.scala | 2 +-
27 files changed, 133 insertions(+), 69 deletions(-)
diff --git a/actor-tests/src/test/java/org/apache/pekko/actor/JavaAPI.java
b/actor-tests/src/test/java/org/apache/pekko/actor/JavaAPI.java
index 1b813aa6e6..13d2126537 100644
--- a/actor-tests/src/test/java/org/apache/pekko/actor/JavaAPI.java
+++ b/actor-tests/src/test/java/org/apache/pekko/actor/JavaAPI.java
@@ -43,7 +43,6 @@ public class JavaAPI {
public void mustCompile() {
final Kill kill = Kill.getInstance();
final PoisonPill pill = PoisonPill.getInstance();
- final ReceiveTimeout t = ReceiveTimeout.getInstance();
final LocalScope ls = LocalScope.getInstance();
final NoScopeGiven noscope = NoScopeGiven.getInstance();
diff --git
a/actor-tests/src/test/scala/org/apache/pekko/actor/ReceiveTimeoutSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/actor/ReceiveTimeoutSpec.scala
index 72002a693e..4c2c11a1a0 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/ReceiveTimeoutSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/ReceiveTimeoutSpec.scala
@@ -56,8 +56,8 @@ object ReceiveTimeoutSpec {
restarting.set(true)
probe ! "crashing"
throw TestException("boom bang")
- case ReceiveTimeout =>
- probe ! ReceiveTimeout
+ case timeout: ReceiveTimeout =>
+ probe ! timeout
case other =>
probe ! other
}
@@ -94,7 +94,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
context.setReceiveTimeout(500.milliseconds)
def receive = {
- case ReceiveTimeout => timeoutLatch.open()
+ case _: ReceiveTimeout => timeoutLatch.open()
}
}))
@@ -102,6 +102,23 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
system.stop(timeoutActor)
}
+ "carry the configured timeout duration" taggedAs TimingTest in {
+ val probe = TestProbe()
+
+ val timeoutActor = system.actorOf(Props(new Actor {
+ context.setReceiveTimeout(500.milliseconds)
+
+ def receive = {
+ case timeout: ReceiveTimeout => probe.ref ! timeout
+ }
+ }))
+
+ val msg = probe.expectMsgType[ReceiveTimeout]
+ msg.timeout should ===(500.milliseconds)
+ msg.getTimeout should ===(java.time.Duration.ofMillis(500))
+ system.stop(timeoutActor)
+ }
+
"reschedule timeout after regular receive" taggedAs TimingTest in {
val timeoutLatch = TestLatch()
val messageCount = 100
@@ -111,8 +128,8 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
context.setReceiveTimeout(1.second)
def receive = {
- case Tick => processedLatch.countDown()
- case ReceiveTimeout => timeoutLatch.open()
+ case Tick => processedLatch.countDown()
+ case _: ReceiveTimeout => timeoutLatch.open()
}
}))
@@ -132,8 +149,8 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
context.setReceiveTimeout(500.milliseconds)
def receive = {
- case Tick => ()
- case ReceiveTimeout =>
+ case Tick => ()
+ case _: ReceiveTimeout =>
count.incrementAndGet
timeoutLatch.open()
context.setReceiveTimeout(Duration.Undefined)
@@ -152,7 +169,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
val timeoutActor = system.actorOf(Props(new Actor {
def receive = {
- case ReceiveTimeout => timeoutLatch.open()
+ case _: ReceiveTimeout => timeoutLatch.open()
}
}))
@@ -167,8 +184,8 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
context.setReceiveTimeout(1.second)
def receive = {
- case ReceiveTimeout => timeoutLatch.open()
- case TransparentTick =>
+ case _: ReceiveTimeout => timeoutLatch.open()
+ case TransparentTick =>
}
}))
@@ -189,7 +206,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
context.setReceiveTimeout(1.second)
def receive = {
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
self ! TransparentTick
timeoutLatch.countDown()
case TransparentTick =>
@@ -210,7 +227,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
context.setReceiveTimeout(1.second)
def receive: Receive = {
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
timeoutLatch.open()
case TransparentTick =>
count.incrementAndGet()
@@ -229,8 +246,8 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
val timeoutActor = system.actorOf(Props(new Actor {
def receive = {
- case TransparentTick => context.setReceiveTimeout(500.milliseconds)
- case ReceiveTimeout => timeoutLatch.open()
+ case TransparentTick => context.setReceiveTimeout(500.milliseconds)
+ case _: ReceiveTimeout => timeoutLatch.open()
}
}))
@@ -247,8 +264,8 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
context.setReceiveTimeout(500.milliseconds)
def receive = {
- case TransparentTick => context.setReceiveTimeout(Duration.Inf)
- case ReceiveTimeout => timeoutLatch.open()
+ case TransparentTick => context.setReceiveTimeout(Duration.Inf)
+ case _: ReceiveTimeout => timeoutLatch.open()
}
}))
@@ -266,8 +283,8 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
context.setReceiveTimeout(initialTimeout)
def receive: Receive = {
- case TransparentTick => context.setReceiveTimeout(Duration.Undefined)
- case ReceiveTimeout => timeoutLatch.open()
+ case TransparentTick =>
context.setReceiveTimeout(Duration.Undefined)
+ case _: ReceiveTimeout => timeoutLatch.open()
}
}))
@@ -292,7 +309,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
count += 1 // do some work then
context.setReceiveTimeout(initialTimeout)
- case ReceiveTimeout => probe.ref ! ReceiveTimeout
+ case timeout: ReceiveTimeout => probe.ref ! timeout
}
}))
@@ -320,7 +337,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
probe.expectMsg("crashing")
probe.expectMsg("stopping")
probe.expectMsg("restarting")
- probe.expectMsg(ReceiveTimeout)
+ probe.expectMsgType[ReceiveTimeout]
}
"Will cancel receive timeout task if stopped" in {
diff --git
a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/CancelReceiveTimeoutSpec.scala
b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/CancelReceiveTimeoutSpec.scala
index c342ba9cfd..e227e624e8 100644
---
a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/CancelReceiveTimeoutSpec.scala
+++
b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/CancelReceiveTimeoutSpec.scala
@@ -96,7 +96,7 @@ class CancelReceiveTimeoutSpec extends
ScalaTestWithActorTestKit with AnyWordSpe
// This is what happens when the scheduler fires ReceiveTimeout before
// cancelReceiveTimeout() is processed but the actor dequeues them in the
// opposite order.
- ref.toClassic ! pekko.actor.ReceiveTimeout
+ ref.toClassic ! pekko.actor.ReceiveTimeout(1.second)
// Step 3: verify the actor is still alive and responsive (no NPE crash).
ref ! Ping
diff --git
a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala
b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala
index 7bee9f44f1..2b9ae0bd84 100644
---
a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala
+++
b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/adapter/ActorAdapter.scala
@@ -99,7 +99,7 @@ import pekko.util.OptionVal
ChildFailed(ActorRefAdapter(ref), ex)
} else Terminated(ActorRefAdapter(ref))
handleSignal(msg)
- case classic.ReceiveTimeout =>
+ case _: classic.ReceiveTimeout =>
// cancelReceiveTimeout() sets receiveTimeoutMsg to null, but a
classic ReceiveTimeout
// that was already enqueued in the mailbox before the cancel cannot
be retracted.
// Discard the stale timeout to avoid passing null into the typed
behavior stack (#3084).
diff --git
a/actor/src/main/mima-filters/2.0.x.backwards.excludes/receive-timeout-with-duration.excludes
b/actor/src/main/mima-filters/2.0.x.backwards.excludes/receive-timeout-with-duration.excludes
new file mode 100644
index 0000000000..8b5bb33515
--- /dev/null
+++
b/actor/src/main/mima-filters/2.0.x.backwards.excludes/receive-timeout-with-duration.excludes
@@ -0,0 +1,43 @@
+# 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.
+
+# ReceiveTimeout changed from case object to case class with a timeout field
+# https://github.com/apache/pekko/issues/2569
+ProblemFilters.exclude[FinalClassProblem]("org.apache.pekko.actor.ReceiveTimeout")
+ProblemFilters.exclude[StaticVirtualMemberProblem]("org.apache.pekko.actor.ReceiveTimeout.canEqual")
+ProblemFilters.exclude[StaticVirtualMemberProblem]("org.apache.pekko.actor.ReceiveTimeout.productIterator")
+ProblemFilters.exclude[StaticVirtualMemberProblem]("org.apache.pekko.actor.ReceiveTimeout.productElement")
+ProblemFilters.exclude[StaticVirtualMemberProblem]("org.apache.pekko.actor.ReceiveTimeout.productArity")
+ProblemFilters.exclude[StaticVirtualMemberProblem]("org.apache.pekko.actor.ReceiveTimeout.productPrefix")
+ProblemFilters.exclude[StaticVirtualMemberProblem]("org.apache.pekko.actor.ReceiveTimeout.productElementNames")
+ProblemFilters.exclude[StaticVirtualMemberProblem]("org.apache.pekko.actor.ReceiveTimeout.productElementName")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.actor.ReceiveTimeout.this")
+ProblemFilters.exclude[MissingTypesProblem]("org.apache.pekko.actor.ReceiveTimeout$")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.actor.ReceiveTimeout.productElementName")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.actor.ReceiveTimeout.productElementNames")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.actor.ReceiveTimeout.productPrefix")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.actor.ReceiveTimeout.productArity")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.actor.ReceiveTimeout.productElement")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.actor.ReceiveTimeout.productIterator")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.actor.ReceiveTimeout.canEqual")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.actor.ReceiveTimeout$")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.actor.ReceiveTimeout.getInstance")
+ProblemFilters.exclude[IncompatibleTemplateDefProblem]("org.apache.pekko.actor.ReceiveTimeout")
+# Scala 3 only: Mirror/product members of the former case object
+ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.pekko.actor.ReceiveTimeout.fromProduct")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.actor.ReceiveTimeout.productElementNames")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.actor.ReceiveTimeout.productIterator")
diff --git a/actor/src/main/scala/org/apache/pekko/actor/Actor.scala
b/actor/src/main/scala/org/apache/pekko/actor/Actor.scala
index 3ba8abafc0..3b73d71729 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/Actor.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/Actor.scala
@@ -18,6 +18,7 @@ import java.util.Optional
import scala.annotation.nowarn
import scala.annotation.tailrec
import scala.beans.BeanProperty
+import scala.jdk.DurationConverters._
import scala.util.control.NoStackTrace
import org.apache.pekko
@@ -144,19 +145,17 @@ private[pekko] final case class
AddressTerminated(address: Address)
with PossiblyHarmful
with DeadLetterSuppression
-abstract class ReceiveTimeout extends PossiblyHarmful
-
/**
- * When using ActorContext.setReceiveTimeout, the singleton instance of
ReceiveTimeout will be sent
- * to the Actor when there hasn't been any message for that long.
+ * When using ActorContext.setReceiveTimeout, a ReceiveTimeout message
carrying the configured
+ * timeout duration will be sent to the Actor when there hasn't been any
message for that long.
*/
@SerialVersionUID(1L)
-case object ReceiveTimeout extends ReceiveTimeout {
+final case class ReceiveTimeout(timeout:
scala.concurrent.duration.FiniteDuration) extends PossiblyHarmful {
/**
- * Java API: get the singleton instance
+ * Java API: get the timeout duration
*/
- def getInstance = this
+ def getTimeout: java.time.Duration = timeout.toJava
}
/**
diff --git
a/actor/src/main/scala/org/apache/pekko/actor/dungeon/ReceiveTimeout.scala
b/actor/src/main/scala/org/apache/pekko/actor/dungeon/ReceiveTimeout.scala
index 2c897e2872..21b4ce216d 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/dungeon/ReceiveTimeout.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/dungeon/ReceiveTimeout.scala
@@ -73,7 +73,7 @@ private[pekko] trait ReceiveTimeout { this: ActorCell =>
private def rescheduleReceiveTimeout(data: State, timeout: FiniteDuration):
Unit = {
data.task.cancel() // Cancel any ongoing future
- data.task = system.scheduler.scheduleOnce(timeout, self,
pekko.actor.ReceiveTimeout)(this.dispatcher)
+ data.task = system.scheduler.scheduleOnce(timeout, self,
pekko.actor.ReceiveTimeout(timeout))(this.dispatcher)
data.version += 1
}
diff --git a/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala
b/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala
index b866f99ed0..3ae1366e13 100644
--- a/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala
+++ b/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala
@@ -100,7 +100,7 @@ private[io] abstract class TcpConnection(val tcp: TcpExt,
val channel: SocketCha
val info = ConnectionInfo(registration, commander, keepOpenOnPeerClosed
= false, useResumeWriting = false)
handleClose(info, Some(sender()), cmd.event)
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
// after sending `Register` user should watch this actor to make sure
// it didn't die because of the timeout
log.debug("Configured registration timeout of [{}] expired, stopping",
RegisterTimeout)
diff --git
a/actor/src/main/scala/org/apache/pekko/io/TcpOutgoingConnection.scala
b/actor/src/main/scala/org/apache/pekko/io/TcpOutgoingConnection.scala
index 34c678ce20..2d6b1dfaee 100644
--- a/actor/src/main/scala/org/apache/pekko/io/TcpOutgoingConnection.scala
+++ b/actor/src/main/scala/org/apache/pekko/io/TcpOutgoingConnection.scala
@@ -87,7 +87,7 @@ private[io] class TcpOutgoingConnection(
register(remoteAddress, registration)
}
}
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
connectionTimeout()
}
@@ -96,7 +96,7 @@ private[io] class TcpOutgoingConnection(
reportConnectFailure {
register(new InetSocketAddress(resolved.address(),
remoteAddress.getPort), registration)
}
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
connectionTimeout()
case Failure(ex) =>
// async-dns responds with a Failure on DNS server lookup failure
@@ -141,7 +141,7 @@ private[io] class TcpOutgoingConnection(
reportConnectFailure {
channelRegistry.register(channel, SelectionKey.OP_CONNECT)
}
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
connectionTimeout()
}
}
diff --git
a/bench-jmh/src/main/scala/org/apache/pekko/actor/ReceiveTimeoutBenchmark.scala
b/bench-jmh/src/main/scala/org/apache/pekko/actor/ReceiveTimeoutBenchmark.scala
index 67556b08f1..5757e0de18 100644
---
a/bench-jmh/src/main/scala/org/apache/pekko/actor/ReceiveTimeoutBenchmark.scala
+++
b/bench-jmh/src/main/scala/org/apache/pekko/actor/ReceiveTimeoutBenchmark.scala
@@ -90,7 +90,7 @@ object ReceiveTimeoutBenchmark {
override def receive: Receive = {
case Message =>
case finished: Finished => finished.latch.countDown()
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
}
}
}
diff --git
a/cluster-metrics/src/multi-jvm/scala/org/apache/pekko/cluster/metrics/sample/StatsService.scala
b/cluster-metrics/src/multi-jvm/scala/org/apache/pekko/cluster/metrics/sample/StatsService.scala
index eef45b2c1a..9aaf98690a 100644
---
a/cluster-metrics/src/multi-jvm/scala/org/apache/pekko/cluster/metrics/sample/StatsService.scala
+++
b/cluster-metrics/src/multi-jvm/scala/org/apache/pekko/cluster/metrics/sample/StatsService.scala
@@ -51,7 +51,7 @@ class StatsAggregator(expectedResults: Int, replyTo:
ActorRef) extends Actor {
replyTo ! StatsResult(meanWordLength)
context.stop(self)
}
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
replyTo ! JobFailed("Service unavailable, try again later")
context.stop(self)
}
diff --git
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
index 1d9baa46dc..752b02fc03 100644
---
a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
+++
b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala
@@ -595,7 +595,7 @@ object ShardCoordinator {
shardRegionFrom,
regions.size)
- timers.startSingleTimer("hand-off-timeout", ReceiveTimeout, handOffTimeout)
+ timers.startSingleTimer("hand-off-timeout",
ReceiveTimeout(handOffTimeout), handOffTimeout)
def receive: Receive = {
case BeginHandOffAck(`shard`) =>
@@ -610,7 +610,7 @@ object ShardCoordinator {
shard)
acked(shardRegion)
}
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
if (isRebalance)
log.debug("{}: Rebalance of [{}] from [{}] timed out", typeName,
shard, shardRegionFrom)
else
@@ -632,7 +632,7 @@ object ShardCoordinator {
def stoppingShard: Receive = {
case ShardStopped(`shard`) => done(ok
= true)
- case ReceiveTimeout => done(ok
= false)
+ case _: ReceiveTimeout => done(ok
= false)
case RebalanceWorker.ShardRegionTerminated(`shardRegionFrom`) =>
log.debug(
"{}: ShardRegion [{}] terminated while waiting for ShardStopped for
shard [{}].",
diff --git
a/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/ClusterShardingSpec.scala
b/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/ClusterShardingSpec.scala
index b939a1b4c9..1653dcf8cb 100644
---
a/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/ClusterShardingSpec.scala
+++
b/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/ClusterShardingSpec.scala
@@ -70,11 +70,11 @@ object ClusterShardingSpec {
}
override def receiveCommand: Receive = {
- case Increment => persist(CounterChanged(+1))(updateState)
- case Decrement => persist(CounterChanged(-1))(updateState)
- case Get(_) => sender() ! count
- case ReceiveTimeout => context.parent ! Passivate(stopMessage = Stop)
- case Stop => context.stop(self)
+ case Increment => persist(CounterChanged(+1))(updateState)
+ case Decrement => persist(CounterChanged(-1))(updateState)
+ case Get(_) => sender() ! count
+ case _: ReceiveTimeout => context.parent ! Passivate(stopMessage = Stop)
+ case Stop => context.stop(self)
}
}
// #counter-actor
@@ -499,7 +499,7 @@ abstract class ClusterShardingSpec(multiNodeConfig:
ClusterShardingSpecConfig)
runOn(second) {
region ! Get(2)
expectMsg(3)
- region ! EntityEnvelope(2, ReceiveTimeout)
+ region ! EntityEnvelope(2, ReceiveTimeout(120.seconds))
// let the Passivate-Stop roundtrip begin to trigger buffering of
subsequent messages
Thread.sleep(200)
region ! EntityEnvelope(2, Increment)
diff --git
a/cluster-tools/src/main/scala/org/apache/pekko/cluster/client/ClusterClient.scala
b/cluster-tools/src/main/scala/org/apache/pekko/cluster/client/ClusterClient.scala
index e953c151ff..1c087eac9d 100644
---
a/cluster-tools/src/main/scala/org/apache/pekko/cluster/client/ClusterClient.scala
+++
b/cluster-tools/src/main/scala/org/apache/pekko/cluster/client/ClusterClient.scala
@@ -924,8 +924,8 @@ object ClusterReceptionist {
}
def receive = {
- case Ping => // keep alive from client
- case ReceiveTimeout =>
+ case Ping => // keep alive from client
+ case _: ReceiveTimeout =>
log.debug("ClientResponseTunnel for client [{}] stopped due to
inactivity", client.path)
context.stop(self)
case msg =>
diff --git
a/cluster/src/main/scala/org/apache/pekko/cluster/SeedNodeProcess.scala
b/cluster/src/main/scala/org/apache/pekko/cluster/SeedNodeProcess.scala
index 9656b14020..36b905b95d 100644
--- a/cluster/src/main/scala/org/apache/pekko/cluster/SeedNodeProcess.scala
+++ b/cluster/src/main/scala/org/apache/pekko/cluster/SeedNodeProcess.scala
@@ -313,8 +313,8 @@ private[cluster] final class JoinSeedNodeProcess(
// first InitJoinAck reply, but incompatible
receiveInitJoinAckIncompatibleConfig(joinTo = address, origin =
sender(), behavior = Some(done))
- case InitJoinNack(_) => // that seed was uninitialized
- case ReceiveTimeout =>
+ case InitJoinNack(_) => // that seed was uninitialized
+ case _: ReceiveTimeout =>
if (attempt >= 2)
logWarning(
ClusterLogMarker.joinFailed,
@@ -332,6 +332,6 @@ private[cluster] final class JoinSeedNodeProcess(
def done: Actor.Receive = {
case InitJoinAck(_, _) => // already received one, skip rest
- case ReceiveTimeout => context.stop(self)
+ case _: ReceiveTimeout => context.stop(self)
}
}
diff --git
a/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/Replicator.scala
b/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/Replicator.scala
index 0546661029..957f5b108b 100644
---
a/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/Replicator.scala
+++
b/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/Replicator.scala
@@ -2639,7 +2639,7 @@ final class Replicator(settings: ReplicatorSettings)
extends Actor with ActorLog
import context.dispatcher
private val sendToSecondarySchedule =
context.system.scheduler.scheduleOnce(timeout / 5, self, SendToSecondary)
- private val timeoutSchedule = context.system.scheduler.scheduleOnce(timeout,
self, ReceiveTimeout)
+ private val timeoutSchedule = context.system.scheduler.scheduleOnce(timeout,
self, ReceiveTimeout(timeout))
var remaining = nodes.iterator.map(_.address).toSet
@@ -2801,7 +2801,7 @@ final class Replicator(settings: ReplicatorSettings)
extends Actor with ActorLog
}
}
secondaryNodes.foreach { replica(_) ! writeMsg }
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
reply(isTimeout = true)
}
@@ -2920,7 +2920,7 @@ final class Replicator(settings: ReplicatorSettings)
extends Actor with ActorLog
reply(ok = true)
case SendToSecondary =>
secondaryNodes.foreach { replica(_) ! readMsg }
- case ReceiveTimeout => reply(ok = false)
+ case _: ReceiveTimeout => reply(ok = false)
}
def reply(ok: Boolean): Unit =
@@ -2947,7 +2947,7 @@ final class Replicator(settings: ReplicatorSettings)
extends Actor with ActorLog
case _: ReadResult =>
// collect late replies
remaining -= sender().path.address
- case SendToSecondary =>
- case ReceiveTimeout =>
+ case SendToSecondary =>
+ case _: ReceiveTimeout =>
}
}
diff --git a/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
b/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
index 4b7ebd32a1..2cbb360a14 100644
--- a/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
+++ b/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
@@ -42,3 +42,9 @@ The Java DSL `SourceWithContext` graph shape now uses
`pekko.japi.Pair` instead
Java code that passes a `SourceWithContext` directly to `GraphDSL` must use
`Pair`-typed stages.
Code that converts it with `asSource()` already uses `Pair` and requires no
changes.
([PR3388](https://github.com/apache/pekko/pull/3388))
+
+`ReceiveTimeout` changed from a `case object` singleton to a `final case
class` that carries the configured
+timeout duration. Scala pattern matches must use a type pattern (`case
timeout: ReceiveTimeout =>`) instead of a
+stable identifier pattern, and the `ReceiveTimeout.getInstance()` method has
been removed from the Java API;
+Java users should match on `ReceiveTimeout.class` instead.
+([PR3399](https://github.com/apache/pekko/pull/3399))
diff --git a/docs/src/test/java/jdocs/actor/FaultHandlingDocSample.java
b/docs/src/test/java/jdocs/actor/FaultHandlingDocSample.java
index 48593c90d3..14daa885d0 100644
--- a/docs/src/test/java/jdocs/actor/FaultHandlingDocSample.java
+++ b/docs/src/test/java/jdocs/actor/FaultHandlingDocSample.java
@@ -84,8 +84,8 @@ public class FaultHandlingDocSample {
getContext().getSystem().terminate();
}
})
- .matchEquals(
- ReceiveTimeout.getInstance(),
+ .match(
+ ReceiveTimeout.class,
x -> {
// No progress within 15 seconds, ServiceUnavailable
log().error("Shutting down due to unavailable service");
diff --git a/docs/src/test/java/jdocs/sharding/ClusterShardingTest.java
b/docs/src/test/java/jdocs/sharding/ClusterShardingTest.java
index 7db1c69131..328cd0fbb9 100644
--- a/docs/src/test/java/jdocs/sharding/ClusterShardingTest.java
+++ b/docs/src/test/java/jdocs/sharding/ClusterShardingTest.java
@@ -218,7 +218,7 @@ public class ClusterShardingTest {
.match(Get.class, this::receiveGet)
.matchEquals(CounterOp.INCREMENT, msg -> receiveIncrement())
.matchEquals(CounterOp.DECREMENT, msg -> receiveDecrement())
- .matchEquals(ReceiveTimeout.getInstance(), msg -> passivate())
+ .match(ReceiveTimeout.class, msg -> passivate())
.build();
}
diff --git a/docs/src/test/scala/docs/actor/ActorDocSpec.scala
b/docs/src/test/scala/docs/actor/ActorDocSpec.scala
index 7cde553ee9..2f4f22d623 100644
--- a/docs/src/test/scala/docs/actor/ActorDocSpec.scala
+++ b/docs/src/test/scala/docs/actor/ActorDocSpec.scala
@@ -555,7 +555,7 @@ class ActorDocSpec extends PekkoSpec("""
case "Hello" =>
// To set in a response to a message
context.setReceiveTimeout(100.milliseconds)
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
// To turn it off
context.setReceiveTimeout(Duration.Undefined)
throw new RuntimeException("Receive timed out")
diff --git a/docs/src/test/scala/docs/actor/FaultHandlingDocSample.scala
b/docs/src/test/scala/docs/actor/FaultHandlingDocSample.scala
index 08db47715b..67014af3fd 100644
--- a/docs/src/test/scala/docs/actor/FaultHandlingDocSample.scala
+++ b/docs/src/test/scala/docs/actor/FaultHandlingDocSample.scala
@@ -67,7 +67,7 @@ class Listener extends Actor with ActorLogging {
context.system.terminate()
}
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
// No progress within 15 seconds, ServiceUnavailable
log.error("Shutting down due to unavailable service")
context.system.terminate()
diff --git
a/docs/src/test/scala/docs/circuitbreaker/CircuitBreakerDocSpec.scala
b/docs/src/test/scala/docs/circuitbreaker/CircuitBreakerDocSpec.scala
index 3cf1f37599..7d4e0d2932 100644
--- a/docs/src/test/scala/docs/circuitbreaker/CircuitBreakerDocSpec.scala
+++ b/docs/src/test/scala/docs/circuitbreaker/CircuitBreakerDocSpec.scala
@@ -75,7 +75,7 @@ class TellPatternActor(recipient: ActorRef) extends Actor
with ActorLogging {
case err: Throwable => {
breaker.fail()
}
- case ReceiveTimeout => {
+ case _: ReceiveTimeout => {
breaker.fail()
}
}
diff --git a/docs/src/test/scala/docs/cluster/FactorialFrontend.scala
b/docs/src/test/scala/docs/cluster/FactorialFrontend.scala
index f645efdd44..fe257c2727 100644
--- a/docs/src/test/scala/docs/cluster/FactorialFrontend.scala
+++ b/docs/src/test/scala/docs/cluster/FactorialFrontend.scala
@@ -46,7 +46,7 @@ class FactorialFrontend(upToN: Int, repeat: Boolean) extends
Actor with ActorLog
if (repeat) sendJobs()
else context.stop(self)
}
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
log.info("Timeout")
sendJobs()
}
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteProxy.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteProxy.scala
index 7e4b175340..139e42e599 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteProxy.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteProxy.scala
@@ -156,7 +156,7 @@ private class ReplayMediator(
case ReplayFailure(cause) =>
replayCompletionPromise.failure(cause)
context.stop(self)
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
replayCompletionPromise.failure(
new AsyncReplayTimeoutException(s"replay timed out after
${replayTimeout.toSeconds} seconds inactivity"))
context.stop(self)
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
index ba02fb9ad4..f9228dc55c 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
@@ -176,9 +176,9 @@ final class PersistencePluginProxy(config: Config) extends
Actor with Stash with
context.watch(target)
unstashAll()
context.become(active(target, address == selfAddress))
- case _: ActorIdentity => // will retry after ReceiveTimeout
- case Terminated(_) =>
- case ReceiveTimeout =>
+ case _: ActorIdentity => // will retry after ReceiveTimeout
+ case Terminated(_) =>
+ case _: ReceiveTimeout =>
sendIdentify(address)
}: Receive).orElse(init)
diff --git
a/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/artery/MaxThroughputSpec.scala
b/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/artery/MaxThroughputSpec.scala
index 5a788df1a4..c7712d3bdd 100644
---
a/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/artery/MaxThroughputSpec.scala
+++
b/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/artery/MaxThroughputSpec.scala
@@ -208,7 +208,7 @@ object MaxThroughputSpec extends MultiNodeConfig {
runWarmup()
} else
target.tell(Warmup(payload), self)
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
target.tell(Warmup(payload), self)
}
diff --git
a/testkit/src/test/scala/org/apache/pekko/testkit/TestActorRefSpec.scala
b/testkit/src/test/scala/org/apache/pekko/testkit/TestActorRefSpec.scala
index d448189423..fe5342b220 100644
--- a/testkit/src/test/scala/org/apache/pekko/testkit/TestActorRefSpec.scala
+++ b/testkit/src/test/scala/org/apache/pekko/testkit/TestActorRefSpec.scala
@@ -103,7 +103,7 @@ object TestActorRefSpec {
class ReceiveTimeoutActor(target: ActorRef) extends Actor {
context.setReceiveTimeout(1.second)
def receive = {
- case ReceiveTimeout =>
+ case _: ReceiveTimeout =>
target ! "timeout"
context.stop(self)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
