This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch feat/receive-timeout-with-duration in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 65d14fe6bd4ee2779ba4703cd03248ae336b6481 Author: 虎鸣 <[email protected]> AuthorDate: Thu Jul 30 02:03:44 2026 +0800 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) --- .../test/java/org/apache/pekko/actor/JavaAPI.java | 1 - .../apache/pekko/actor/ReceiveTimeoutSpec.scala | 44 +++++++++++++++------- .../actor/typed/CancelReceiveTimeoutSpec.scala | 2 +- .../typed/internal/adapter/ActorAdapter.scala | 2 +- .../receive-timeout-with-duration.excludes | 26 +++++++++++++ .../main/scala/org/apache/pekko/actor/Actor.scala | 12 +++--- .../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 | 4 +- .../pekko/cluster/client/ClusterClient.scala | 2 +- .../org/apache/pekko/cluster/SeedNodeProcess.scala | 4 +- .../apache/pekko/cluster/ddata/Replicator.scala | 8 ++-- .../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 | 2 +- .../pekko/remote/artery/MaxThroughputSpec.scala | 2 +- .../apache/pekko/testkit/TestActorRefSpec.scala | 2 +- 26 files changed, 93 insertions(+), 54 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..0242ad19b8 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,22 @@ 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) + system.stop(timeoutActor) + } + "reschedule timeout after regular receive" taggedAs TimingTest in { val timeoutLatch = TestLatch() val messageCount = 100 @@ -112,7 +128,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() { def receive = { case Tick => processedLatch.countDown() - case ReceiveTimeout => timeoutLatch.open() + case _: ReceiveTimeout => timeoutLatch.open() } })) @@ -133,7 +149,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() { def receive = { case Tick => () - case ReceiveTimeout => + case _: ReceiveTimeout => count.incrementAndGet timeoutLatch.open() context.setReceiveTimeout(Duration.Undefined) @@ -152,7 +168,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() { val timeoutActor = system.actorOf(Props(new Actor { def receive = { - case ReceiveTimeout => timeoutLatch.open() + case _: ReceiveTimeout => timeoutLatch.open() } })) @@ -167,7 +183,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() { context.setReceiveTimeout(1.second) def receive = { - case ReceiveTimeout => timeoutLatch.open() + case _: ReceiveTimeout => timeoutLatch.open() case TransparentTick => } })) @@ -189,7 +205,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() { context.setReceiveTimeout(1.second) def receive = { - case ReceiveTimeout => + case _: ReceiveTimeout => self ! TransparentTick timeoutLatch.countDown() case TransparentTick => @@ -210,7 +226,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() { context.setReceiveTimeout(1.second) def receive: Receive = { - case ReceiveTimeout => + case _: ReceiveTimeout => timeoutLatch.open() case TransparentTick => count.incrementAndGet() @@ -230,7 +246,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() { val timeoutActor = system.actorOf(Props(new Actor { def receive = { case TransparentTick => context.setReceiveTimeout(500.milliseconds) - case ReceiveTimeout => timeoutLatch.open() + case _: ReceiveTimeout => timeoutLatch.open() } })) @@ -248,7 +264,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() { def receive = { case TransparentTick => context.setReceiveTimeout(Duration.Inf) - case ReceiveTimeout => timeoutLatch.open() + case _: ReceiveTimeout => timeoutLatch.open() } })) @@ -267,7 +283,7 @@ class ReceiveTimeoutSpec extends PekkoSpec() { def receive: Receive = { case TransparentTick => context.setReceiveTimeout(Duration.Undefined) - case ReceiveTimeout => timeoutLatch.open() + case _: ReceiveTimeout => timeoutLatch.open() } })) @@ -292,7 +308,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 +336,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..9abb2e8d88 --- /dev/null +++ b/actor/src/main/mima-filters/2.0.x.backwards.excludes/receive-timeout-with-duration.excludes @@ -0,0 +1,26 @@ +# 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[MissingClassProblem]("org.apache.pekko.actor.ReceiveTimeout$") +ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.pekko.actor.ReceiveTimeout.copy") +ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.pekko.actor.ReceiveTimeout.productElement") +ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.pekko.actor.ReceiveTimeout.productIterator") +ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.actor.ReceiveTimeout.getInstance") +ProblemFilters.exclude[MissingTypesProblem]("org.apache.pekko.actor.ReceiveTimeout") +ProblemFilters.exclude[IncompatibleTemplateDefProblem]("org.apache.pekko.actor.ReceiveTimeout") 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..e535797c3c 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/Actor.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/Actor.scala @@ -144,19 +144,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 = java.time.Duration.ofNanos(timeout.toNanos) } /** 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..87070219c7 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..0727b90e7d 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..20e4ea196a 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 @@ -73,7 +73,7 @@ object ClusterShardingSpec { case Increment => persist(CounterChanged(+1))(updateState) case Decrement => persist(CounterChanged(-1))(updateState) case Get(_) => sender() ! count - case ReceiveTimeout => context.parent ! Passivate(stopMessage = Stop) + case _: ReceiveTimeout => context.parent ! Passivate(stopMessage = Stop) case Stop => context.stop(self) } } @@ -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..44efad19da 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 @@ -925,7 +925,7 @@ object ClusterReceptionist { def receive = { case Ping => // keep alive from client - case ReceiveTimeout => + 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..f7c83ae305 100644 --- a/cluster/src/main/scala/org/apache/pekko/cluster/SeedNodeProcess.scala +++ b/cluster/src/main/scala/org/apache/pekko/cluster/SeedNodeProcess.scala @@ -314,7 +314,7 @@ private[cluster] final class JoinSeedNodeProcess( receiveInitJoinAckIncompatibleConfig(joinTo = address, origin = sender(), behavior = Some(done)) case InitJoinNack(_) => // that seed was uninitialized - case ReceiveTimeout => + 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..1c4e3f92cb 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 = @@ -2948,6 +2948,6 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog // collect late replies remaining -= sender().path.address case SendToSecondary => - case ReceiveTimeout => + case _: ReceiveTimeout => } } 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..2f06cf3d5b 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 @@ -178,7 +178,7 @@ final class PersistencePluginProxy(config: Config) extends Actor with Stash with context.become(active(target, address == selfAddress)) case _: ActorIdentity => // will retry after ReceiveTimeout case Terminated(_) => - case ReceiveTimeout => + 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]
