From 9f9bd354a4aa3c21f2ebf53cad5c67125d59eba5 Mon Sep 17 00:00:00 2001 From: Arnout Engelen Date: Mon, 21 Feb 2022 20:47:12 +0100 Subject: [PATCH] typo: resendFirsUnconfirmedIdleTimeout -> resendFirstUnconfirmedIdleTimeout (#31159) --- .../delivery/ShardingProducerController.scala | 28 +++++++++++++------ .../ShardingProducerControllerImpl.scala | 4 +-- .../ReliableDeliveryShardingSpec.scala | 2 +- 3 files changed, 23 insertions(+), 11 deletions(-) diff --git a/akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/delivery/ShardingProducerController.scala b/akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/delivery/ShardingProducerController.scala index 81c1b9b46b..4668e6335d 100644 --- a/akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/delivery/ShardingProducerController.scala +++ b/akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/delivery/ShardingProducerController.scala @@ -200,9 +200,13 @@ object ShardingProducerController { val bufferSize: Int, val internalAskTimeout: FiniteDuration, val cleanupUnusedAfter: FiniteDuration, - val resendFirsUnconfirmedIdleTimeout: FiniteDuration, + val resendFirstUnconfirmedIdleTimeout: FiniteDuration, val producerControllerSettings: ProducerController.Settings) { + @Deprecated + @deprecated("use resendFirstUnconfirmedIdleTimeout", "2.6.19") + def resendFirsUnconfirmedIdleTimeout: FiniteDuration = resendFirstUnconfirmedIdleTimeout + if (producerControllerSettings.chunkLargeMessagesBytes > 0) throw new IllegalArgumentException("Chunked messages not implemented for sharding yet.") @@ -221,11 +225,19 @@ object ShardingProducerController { def withCleanupUnusedAfter(newCleanupUnusedAfter: java.time.Duration): Settings = copy(cleanupUnusedAfter = newCleanupUnusedAfter.asScala) - def withResendFirsUnconfirmedIdleTimeout(newResendFirsUnconfirmedIdleTimeout: FiniteDuration): Settings = - copy(resendFirsUnconfirmedIdleTimeout = newResendFirsUnconfirmedIdleTimeout) + def withResendFirstUnconfirmedIdleTimeout(newResendFirstUnconfirmedIdleTimeout: FiniteDuration): Settings = + copy(resendFirstUnconfirmedIdleTimeout = newResendFirstUnconfirmedIdleTimeout) - def withResendFirsUnconfirmedIdleTimeout(newResendFirsUnconfirmedIdleTimeout: java.time.Duration): Settings = - copy(resendFirsUnconfirmedIdleTimeout = newResendFirsUnconfirmedIdleTimeout.asScala) + def withResendFirstUnconfirmedIdleTimeout(newResendFirstUnconfirmedIdleTimeout: java.time.Duration): Settings = + copy(resendFirstUnconfirmedIdleTimeout = newResendFirstUnconfirmedIdleTimeout.asScala) + + @deprecated("use resendFirstUnconfirmedIdleTimeout", "2.6.19") + def withResendFirsUnconfirmedIdleTimeout(newResendFirstUnconfirmedIdleTimeout: FiniteDuration): Settings = + copy(resendFirstUnconfirmedIdleTimeout = newResendFirstUnconfirmedIdleTimeout) + + @Deprecated + def withResendFirsUnconfirmedIdleTimeout(newResendFirstUnconfirmedIdleTimeout: java.time.Duration): Settings = + copy(resendFirstUnconfirmedIdleTimeout = newResendFirstUnconfirmedIdleTimeout.asScala) def withProducerControllerSettings(newProducerControllerSettings: ProducerController.Settings): Settings = copy(producerControllerSettings = newProducerControllerSettings) @@ -237,17 +249,17 @@ object ShardingProducerController { bufferSize: Int = bufferSize, internalAskTimeout: FiniteDuration = internalAskTimeout, cleanupUnusedAfter: FiniteDuration = cleanupUnusedAfter, - resendFirsUnconfirmedIdleTimeout: FiniteDuration = resendFirsUnconfirmedIdleTimeout, + resendFirstUnconfirmedIdleTimeout: FiniteDuration = resendFirstUnconfirmedIdleTimeout, producerControllerSettings: ProducerController.Settings = producerControllerSettings) = new Settings( bufferSize, internalAskTimeout, cleanupUnusedAfter, - resendFirsUnconfirmedIdleTimeout, + resendFirstUnconfirmedIdleTimeout, producerControllerSettings) override def toString: String = - s"Settings($bufferSize,$internalAskTimeout,$resendFirsUnconfirmedIdleTimeout,$producerControllerSettings)" + s"Settings($bufferSize,$internalAskTimeout,$resendFirstUnconfirmedIdleTimeout,$producerControllerSettings)" } def apply[A: ClassTag]( diff --git a/akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/delivery/internal/ShardingProducerControllerImpl.scala b/akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/delivery/internal/ShardingProducerControllerImpl.scala index 3415967cc2..f3c421cd6e 100644 --- a/akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/delivery/internal/ShardingProducerControllerImpl.scala +++ b/akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/delivery/internal/ShardingProducerControllerImpl.scala @@ -142,7 +142,7 @@ import akka.util.Timeout def becomeActive(p: ActorRef[RequestNext[A]], s: DurableProducerQueue.State[A]): Behavior[InternalCommand] = { Behaviors.withTimers { timers => timers.startTimerWithFixedDelay(CleanupUnused, settings.cleanupUnusedAfter / 2) - timers.startTimerWithFixedDelay(ResendFirstUnconfirmed, settings.resendFirsUnconfirmedIdleTimeout / 2) + timers.startTimerWithFixedDelay(ResendFirstUnconfirmed, settings.resendFirstUnconfirmedIdleTimeout / 2) // resend unconfirmed before other stashed messages Behaviors.withStash[InternalCommand](settings.bufferSize) { newStashBuffer => @@ -476,7 +476,7 @@ private class ShardingProducerControllerImpl[A: ClassTag]( s.out.foreach { case (outKey: OutKey, outState) => val idleDurationMillis = (now - outState.usedNanoTime) / 1000 / 1000 - if (outState.unconfirmed.nonEmpty && idleDurationMillis >= settings.resendFirsUnconfirmedIdleTimeout.toMillis) { + if (outState.unconfirmed.nonEmpty && idleDurationMillis >= settings.resendFirstUnconfirmedIdleTimeout.toMillis) { context.log.debug( "Resend first unconfirmed for [{}], because it was idle for [{} ms]", outKey, diff --git a/akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/delivery/ReliableDeliveryShardingSpec.scala b/akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/delivery/ReliableDeliveryShardingSpec.scala index cde61551f0..9746438a65 100644 --- a/akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/delivery/ReliableDeliveryShardingSpec.scala +++ b/akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/delivery/ReliableDeliveryShardingSpec.scala @@ -352,7 +352,7 @@ class ReliableDeliveryShardingSpec })) val shardingProducerSettings = - ShardingProducerController.Settings(system).withResendFirsUnconfirmedIdleTimeout(1500.millis) + ShardingProducerController.Settings(system).withResendFirstUnconfirmedIdleTimeout(1500.millis) val shardingProducerController = spawn( ShardingProducerController[TestConsumer.Job](producerId, region, None, shardingProducerSettings),