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

typo: resendFirsUnconfirmedIdleTimeout -> resendFirstUnconfirmedIdleTimeout (#31159)

Browse source
This commit is contained in:
Arnout Engelen 2022-02-21 20:47:12 +01:00 • committed by GitHub
parent 61e44c3bad
commit 9f9bd354a4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 23 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

28
akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/delivery/ShardingProducerController.scala
View file

@ -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](

4
akka-cluster-sharding-typed/src/main/scala/akka/cluster/sharding/typed/delivery/internal/ShardingProducerControllerImpl.scala
View file

@ -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,

2
akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/delivery/ReliableDeliveryShardingSpec.scala
View file

@ -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),