You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2026/01/17 19:01:34 UTC
(pekko) 02/03: Reformat with scalafmt 3.10.4
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 63212bde59aab78011974d14f8ec4da3fc1a0fb3
Author: Scala Steward <sc...@virtuslab.com>
AuthorDate: Sat Jan 17 16:18:04 2026 +0000
Reformat with scalafmt 3.10.4
Executed command: scalafmt --non-interactive
---
.../org/apache/pekko/actor/ActorSystemSpec.scala | 11 +++--
.../pekko/actor/CoordinatedShutdownSpec.scala | 9 ++--
.../apache/pekko/actor/ReceiveTimeoutSpec.scala | 3 +-
.../org/apache/pekko/actor/SchedulerSpec.scala | 9 ++--
.../pekko/actor/SupervisorHierarchySpec.scala | 3 +-
.../actor/dispatch/BalancingDispatcherSpec.scala | 3 +-
.../scala/org/apache/pekko/pattern/AskSpec.scala | 4 +-
.../apache/pekko/actor/typed/SupervisionSpec.scala | 15 ++++--
.../actor/typed/internal/ActorSystemSpec.scala | 3 +-
.../typed/delivery/DurableProducerQueue.scala | 6 ++-
.../delivery/internal/ProducerControllerImpl.scala | 3 +-
.../pekko/actor/typed/internal/Supervision.scala | 3 +-
.../scala/org/apache/pekko/actor/ActorPath.scala | 9 +++-
.../scala/org/apache/pekko/actor/ActorSystem.scala | 9 ++--
.../pekko/actor/LightArrayRevolverScheduler.scala | 11 +++--
.../main/scala/org/apache/pekko/actor/Stash.scala | 5 +-
.../org/apache/pekko/dispatch/Mailboxes.scala | 11 ++---
.../apache/pekko/dispatch/ThreadPoolBuilder.scala | 17 ++-----
.../scala/org/apache/pekko/event/EventStream.scala | 3 +-
.../org/apache/pekko/event/LoggingReceive.scala | 5 +-
.../pekko/io/dns/internal/TcpDnsClient.scala | 3 +-
.../org/apache/pekko/pattern/CircuitBreaker.scala | 3 +-
.../pekko/pattern/CircuitBreakersRegistry.scala | 7 +--
.../scala/org/apache/pekko/pattern/Patterns.scala | 8 +--
.../internal/BackoffOnRestartSupervisor.scala | 8 +--
.../pattern/internal/BackoffOnStopSupervisor.scala | 8 +--
.../org/apache/pekko/routing/ConsistentHash.scala | 6 ++-
.../org/apache/pekko/util/FrequencySketch.scala | 9 ++--
.../scala/org/apache/pekko/util/SWARUtil.scala | 3 +-
.../typed/delivery/ReliableDeliveryBenchmark.scala | 5 +-
.../apache/pekko/io/dns/IdGeneratorBanchmark.scala | 11 +----
build.sbt | 8 +--
.../internal/ShardingProducerControllerImpl.scala | 9 ++--
.../cluster/sharding/typed/internal/Murmur2.scala | 5 +-
.../internal/ReplicatedShardingExtensionImpl.scala | 3 +-
.../typed/delivery/DeliveryThroughputSpec.scala | 10 ++--
.../cluster/sharding/ClusterShardingSettings.scala | 3 +-
.../pekko/cluster/sharding/ShardCoordinator.scala | 7 ++-
.../pekko/cluster/sharding/ShardRegion.scala | 12 ++---
.../internal/LeastShardAllocationStrategy.scala | 5 +-
.../ClusterShardingMessageSerializer.scala | 3 +-
.../RandomizedBrainResolverIntegrationSpec.scala | 13 +++--
.../sharding/MultiDcClusterShardingSpec.scala | 3 +-
.../sharding/MultiNodeClusterShardingConfig.scala | 3 +-
.../JoinConfigCompatCheckShardingSpec.scala | 3 +-
...eastShardAllocationStrategyRandomizedSpec.scala | 6 +--
.../cluster/pubsub/DistributedPubSubMediator.scala | 7 +--
.../ddata/typed/internal/ReplicatorBehavior.scala | 57 ++++++++++------------
.../pekko/cluster/typed/ClusterSingleton.scala | 3 +-
.../receptionist/ClusterReceptionist.scala | 4 +-
.../delivery/ReliableDeliverySerializerSpec.scala | 6 ++-
.../org/apache/pekko/cluster/ClusterDaemon.scala | 9 ++--
.../org/apache/pekko/cluster/ClusterEvent.scala | 3 +-
.../org/apache/pekko/cluster/MembershipState.scala | 3 +-
.../cluster/routing/ClusterRouterConfig.scala | 3 +-
.../ClusterAccrualFailureDetectorSpec.scala | 3 +-
.../pekko/cluster/ClusterDeathWatchSpec.scala | 3 +-
.../apache/pekko/cluster/JoinInProgressSpec.scala | 3 +-
.../apache/pekko/cluster/LeaderLeavingSpec.scala | 3 +-
.../cluster/MembershipChangeListenerUpSpec.scala | 6 ++-
.../pekko/cluster/MinMembersBeforeUpSpec.scala | 6 ++-
.../pekko/cluster/MultiDcSunnyWeatherSpec.scala | 5 +-
.../apache/pekko/cluster/NodeMembershipSpec.scala | 6 ++-
.../org/apache/pekko/cluster/SplitBrainSpec.scala | 3 +-
.../org/apache/pekko/cluster/StressSpec.scala | 14 +++---
.../cluster/ClusterHeartbeatSenderStateSpec.scala | 4 +-
.../cluster/JoinConfigCompatCheckerSpec.scala | 27 ++++++----
.../pekko/cluster/ReachabilityPerfSpec.scala | 15 +++---
.../pekko/cluster/StartupWithOneThreadSpec.scala | 6 ++-
.../apache/pekko/cluster/VectorClockPerfSpec.scala | 12 ++---
.../pekko/cluster/sbr/SplitBrainResolverSpec.scala | 21 +++++---
.../org/apache/pekko/cluster/ddata/ORMap.scala | 14 +++---
.../apache/pekko/cluster/ddata/Replicator.scala | 3 +-
.../docs/actor/io/dns/DnsCompileOnlyDocSpec.scala | 12 ++---
.../test/scala/typed/tutorial_5/DeviceGroup.scala | 6 +--
.../pekko/remote/testconductor/Conductor.scala | 13 +----
.../remote/testconductor/RemoteConnection.scala | 5 +-
.../leveldb/EventsByPersistenceIdStage.scala | 3 +-
.../leveldb/scaladsl/LeveldbReadJournal.scala | 3 +-
.../persistence/scalatest/OptionalTests.scala | 5 +-
.../javadsl/PersistenceTestKitReadJournal.scala | 6 +--
.../scaladsl/PersistenceTestKitReadJournal.scala | 5 +-
.../PersistenceTestKitDurableStateStore.scala | 7 +--
.../scaladsl/EventSourcedBehaviorTestKitSpec.scala | 3 +-
.../testkit/scaladsl/ScalaDslUtils.scala | 10 +---
.../typed/ReplicatedAuctionExampleSpec.scala | 6 ++-
.../persistence/typed/internal/BehaviorSetup.scala | 3 +-
.../pekko/persistence/typed/internal/Running.scala | 16 ++----
.../persistence/typed/ManyRecoveriesSpec.scala | 3 +-
.../journal/PersistencePluginProxy.scala | 11 +----
.../persistence/journal/inmem/InmemJournal.scala | 9 ++--
.../persistence/AtLeastOnceDeliverySpec.scala | 6 ++-
project/Doc.scala | 3 +-
project/Paradox.scala | 9 ++--
project/PekkoBuild.scala | 20 +++++---
project/PekkoDevelocityPlugin.scala | 5 +-
project/PekkoDisciplinePlugin.scala | 53 ++++++++++----------
project/SbtMultiJvmPlugin.scala | 3 +-
project/StreamOperatorsIndexGenerator.scala | 3 +-
project/TestExtras.scala | 9 ++--
.../pekko/remote/routing/RemoteRandomSpec.scala | 3 +-
.../remote/routing/RemoteScatterGatherSpec.scala | 3 +-
.../scala/org/apache/pekko/remote/Endpoint.scala | 8 ++-
.../org/apache/pekko/remote/RemoteDaemon.scala | 13 +----
.../pekko/remote/artery/ArteryTransport.scala | 4 +-
.../apache/pekko/remote/artery/Association.scala | 3 +-
.../pekko/remote/artery/EnvelopeBufferPool.scala | 6 +--
.../pekko/remote/artery/MessageDispatcher.scala | 3 +-
.../pekko/remote/artery/aeron/AeronSink.scala | 3 +-
.../artery/aeron/ArteryAeronUdpTransport.scala | 11 ++---
.../remote/artery/compress/CompressionTable.scala | 3 +-
.../remote/artery/tcp/ArteryTcpTransport.scala | 11 ++---
.../transport/AbstractTransportAdapter.scala | 3 +-
.../transport/ThrottlerTransportAdapter.scala | 6 +--
.../remote/transport/netty/NettyTransport.scala | 14 +-----
.../pekko/remote/AccrualFailureDetectorSpec.scala | 7 ++-
.../pekko/remote/DeadlineFailureDetectorSpec.scala | 3 +-
.../pekko/remote/artery/ImmutableLongMapSpec.scala | 9 ++--
.../remote/artery/OutboundIdleShutdownSpec.scala | 3 +-
.../remote/artery/RemoteActorSelectionSpec.scala | 11 +----
.../pekko/remote/artery/RemoteWatcherSpec.scala | 3 +-
.../pekko/remote/artery/tcp/TcpFramingSpec.scala | 4 +-
.../pekko/remote/classic/RemoteWatcherSpec.scala | 3 +-
.../classic/transport/PekkoProtocolSpec.scala | 5 +-
.../ArteryMessageSerializerSpec.scala | 4 +-
.../jackson/JacksonObjectMapperProvider.scala | 23 ++-------
.../jackson3/JacksonObjectMapperProvider.scala | 13 +----
.../jackson3/JacksonSerializerSpec.scala | 6 +--
.../pekko/stream/testkit/StreamTestKit.scala | 6 +--
.../impl/fusing/GraphInterpreterSpecKit.scala | 5 +-
.../apache/pekko/stream/DslConsistencySpec.scala | 6 ++-
.../pekko/stream/DslFactoriesConsistencySpec.scala | 9 ++--
.../stream/impl/LinearTraversalBuilderSpec.scala | 3 +-
.../pekko/stream/impl/fusing/InterpreterSpec.scala | 3 +-
.../impl/fusing/LifecycleInterpreterSpec.scala | 6 ++-
.../pekko/stream/io/ByteStringParserSpec.scala | 3 +-
.../pekko/stream/io/InputStreamSourceSpec.scala | 6 +--
.../stream/scaladsl/CancellationStrategySpec.scala | 3 +-
.../stream/scaladsl/FlowFlatMapPrefixSpec.scala | 11 +----
.../stream/scaladsl/FlowGroupedWithinSpec.scala | 3 +-
.../pekko/stream/scaladsl/FlowIteratorSpec.scala | 4 +-
.../pekko/stream/scaladsl/FlowOrElseSpec.scala | 3 +-
.../apache/pekko/stream/scaladsl/RestartSpec.scala | 6 ++-
.../pekko/stream/scaladsl/RetryFlowSpec.scala | 3 +-
.../pekko/stream/MapAsyncPartitionedSpec.scala | 11 +----
.../main/scala/org/apache/pekko/stream/Shape.scala | 3 +-
.../apache/pekko/stream/SystemMaterializer.scala | 9 ++--
.../scala/org/apache/pekko/stream/impl/FanIn.scala | 3 +-
.../pekko/stream/impl/fusing/FutureFlow.scala | 7 +--
.../org/apache/pekko/stream/impl/fusing/Ops.scala | 12 +----
.../pekko/stream/impl/fusing/SourceSink.scala | 6 +--
.../pekko/stream/impl/io/FileOutputStage.scala | 7 +--
.../pekko/stream/impl/io/InputStreamSource.scala | 7 +--
.../apache/pekko/stream/impl/io/TcpStages.scala | 3 +-
.../org/apache/pekko/stream/stage/GraphStage.scala | 3 +-
.../pekko/testkit/CallingThreadDispatcher.scala | 19 ++------
156 files changed, 520 insertions(+), 630 deletions(-)
diff --git a/actor-tests/src/test/scala/org/apache/pekko/actor/ActorSystemSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/actor/ActorSystemSpec.scala
index d791552dc5..2871d9ff39 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/ActorSystemSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/ActorSystemSpec.scala
@@ -43,11 +43,12 @@ object ActorSystemSpec {
def receive = {
case n: Int =>
master = sender()
- terminaters = Set() ++ (for (_ <- 1 to n) yield {
- val man = context.watch(context.system.actorOf(Props[Terminater]()))
- man ! "run"
- man
- })
+ terminaters = Set() ++
+ (for (_ <- 1 to n) yield {
+ val man = context.watch(context.system.actorOf(Props[Terminater]()))
+ man ! "run"
+ man
+ })
case Terminated(child) if terminaters contains child =>
terminaters -= child
if (terminaters.isEmpty) {
diff --git a/actor-tests/src/test/scala/org/apache/pekko/actor/CoordinatedShutdownSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/actor/CoordinatedShutdownSpec.scala
index f2dfe81ba6..9261674521 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/CoordinatedShutdownSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/CoordinatedShutdownSpec.scala
@@ -582,7 +582,8 @@ class CoordinatedShutdownSpec
}
}
- "add and remove user JVM hooks with run-by-jvm-shutdown-hook = off, terminate-actor-system = off" in new JvmHookTest {
+ "add and remove user JVM hooks with run-by-jvm-shutdown-hook = off, terminate-actor-system = off" in
+ new JvmHookTest {
lazy val systemName = s"CoordinatedShutdownSpec-JvmHooks-1-${System.currentTimeMillis()}"
lazy val systemConfig = ConfigFactory.parseString("""
pekko.coordinated-shutdown.run-by-jvm-shutdown-hook = off
@@ -598,7 +599,8 @@ class CoordinatedShutdownSpec
}
}
- "add and remove user JVM hooks with run-by-jvm-shutdown-hook = on, terminate-actor-system = off" in new JvmHookTest {
+ "add and remove user JVM hooks with run-by-jvm-shutdown-hook = on, terminate-actor-system = off" in
+ new JvmHookTest {
lazy val systemName = s"CoordinatedShutdownSpec-JvmHooks-2-${System.currentTimeMillis()}"
lazy val systemConfig = ConfigFactory.parseString("""
pekko.coordinated-shutdown.run-by-jvm-shutdown-hook = on
@@ -630,7 +632,8 @@ class CoordinatedShutdownSpec
}
}
- "add and remove user JVM hooks with run-by-jvm-shutdown-hook = on, pekko.jvm-shutdown-hooks = off" in new JvmHookTest {
+ "add and remove user JVM hooks with run-by-jvm-shutdown-hook = on, pekko.jvm-shutdown-hooks = off" in
+ new JvmHookTest {
lazy val systemName = s"CoordinatedShutdownSpec-JvmHooks-4-${System.currentTimeMillis()}"
lazy val systemConfig = ConfigFactory.parseString("""
pekko.jvm-shutdown-hooks = off
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 0a0b0ee0eb..4cf8978adf 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
@@ -265,7 +265,8 @@ class ReceiveTimeoutSpec extends PekkoSpec() {
system.stop(timeoutActor)
}
- "work correctly if the timeout is changed multiple times while handling a NotInfluenceReceiveTimeout message" taggedAs TimingTest in {
+ "work correctly if the timeout is changed multiple times while handling a NotInfluenceReceiveTimeout message" taggedAs
+ TimingTest in {
val probe = TestProbe()
val initialTimeout = 500.millis
diff --git a/actor-tests/src/test/scala/org/apache/pekko/actor/SchedulerSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/actor/SchedulerSpec.scala
index 14602cdd54..cca2a2c9e4 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/SchedulerSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/SchedulerSpec.scala
@@ -819,10 +819,11 @@ class LightArrayRevolverSchedulerSpec extends PekkoSpec(SchedulerSpec.testConfRe
override protected def waitNanos(ns: Long): Unit = {
// println(s"waiting $ns")
prb.ref ! ns
- try time += (lbq.get match {
- case q: LinkedBlockingQueue[Long] => q.take()
- case null => 0L
- })
+ try time +=
+ (lbq.get match {
+ case q: LinkedBlockingQueue[Long] => q.take()
+ case null => 0L
+ })
catch {
case _: InterruptedException => Thread.currentThread.interrupt()
}
diff --git a/actor-tests/src/test/scala/org/apache/pekko/actor/SupervisorHierarchySpec.scala b/actor-tests/src/test/scala/org/apache/pekko/actor/SupervisorHierarchySpec.scala
index a0c248516d..61d75642d7 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/SupervisorHierarchySpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/SupervisorHierarchySpec.scala
@@ -559,7 +559,8 @@ object SupervisorHierarchySpec {
case this.Event(StateTimeout, todo) =>
log.info("dumping state due to StateTimeout")
log.info(
- "children: " + children.size + " pinged: " + pingChildren.size + " idle: " + idleChildren.size + " work: " + todo)
+ "children: " + children.size + " pinged: " + pingChildren.size + " idle: " + idleChildren.size + " work: " +
+ todo)
pingChildren.foreach(println)
println(system.asInstanceOf[ActorSystemImpl].printTree)
pingChildren.foreach(getErrorsUp)
diff --git a/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/BalancingDispatcherSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/BalancingDispatcherSpec.scala
index 08e0bf2ab9..63e367b21a 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/BalancingDispatcherSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/BalancingDispatcherSpec.scala
@@ -95,7 +95,8 @@ class BalancingDispatcherSpec extends PekkoSpec(BalancingDispatcherSpec.config)
fast.underlying.asInstanceOf[ActorCell].mailbox.hasMessages should ===(false)
slow.underlying.asInstanceOf[ActorCell].mailbox.hasMessages should ===(false)
fast.underlying.asInstanceOf[ActorCell].actor.asInstanceOf[DelayableActor].invocationCount should be > sentToFast
- fast.underlying.asInstanceOf[ActorCell].actor.asInstanceOf[DelayableActor].invocationCount should be >
+ fast.underlying.asInstanceOf[ActorCell].actor.asInstanceOf[DelayableActor].invocationCount should
+ be >
slow.underlying.asInstanceOf[ActorCell].actor.asInstanceOf[DelayableActor].invocationCount
system.stop(slow)
system.stop(fast)
diff --git a/actor-tests/src/test/scala/org/apache/pekko/pattern/AskSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/pattern/AskSpec.scala
index b34e001f1c..57679f14d7 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/pattern/AskSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/pattern/AskSpec.scala
@@ -129,8 +129,8 @@ class AskSpec extends PekkoSpec("""
implicit val timeout: Timeout = Timeout(5.seconds)
import system.dispatcher
val echo = system.actorOf(Props(new Actor { def receive = { case x => sender() ! x } }), "select-echo")
- val identityFuture =
- (system.actorSelection("/user/select-echo") ? Identify(None)).mapTo[ActorIdentity].map(_.ref.get)
+ val identityFuture = (system.actorSelection("/user/select-echo") ? Identify(None)).mapTo[ActorIdentity].map(
+ _.ref.get)
Await.result(identityFuture, 5.seconds) should ===(echo)
}
diff --git a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala
index 76368df75f..30c9d61eee 100644
--- a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala
+++ b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/SupervisionSpec.scala
@@ -461,7 +461,8 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
.onFailure[Exception](SupervisorStrategy.restart)
}(ReceivedSignal(PostStop))
- "support nesting exceptions with outer restart and inner backoff strategies" in throwIOExceptionThenIllegalArgumentException {
+ "support nesting exceptions with outer restart and inner backoff strategies" in
+ throwIOExceptionThenIllegalArgumentException {
probeRef =>
supervise(
supervise(targetBehavior(probeRef))
@@ -469,14 +470,16 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
.onFailure[IOException](SupervisorStrategy.restart)
}()
- "support flattening exceptions with outer restart and inner backoff strategies" in throwIOExceptionThenIllegalArgumentException {
+ "support flattening exceptions with outer restart and inner backoff strategies" in
+ throwIOExceptionThenIllegalArgumentException {
probeRef =>
supervise(targetBehavior(probeRef))
.onFailure[IllegalArgumentException](SupervisorStrategy.restartWithBackoff(10.millis, 10.millis, 0.0))
.onFailure[IOException](SupervisorStrategy.restart)
}()
- "support nesting exceptions with inner restart and outer backoff strategies" in throwIOExceptionThenIllegalArgumentException {
+ "support nesting exceptions with inner restart and outer backoff strategies" in
+ throwIOExceptionThenIllegalArgumentException {
probeRef =>
supervise(
supervise(targetBehavior(probeRef))
@@ -484,7 +487,8 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
.onFailure[IOException](SupervisorStrategy.restartWithBackoff(10.millis, 10.millis, 0.0))
}()
- "support flattening exceptions with inner restart and outer backoff strategies" in throwIOExceptionThenIllegalArgumentException {
+ "support flattening exceptions with inner restart and outer backoff strategies" in
+ throwIOExceptionThenIllegalArgumentException {
probeRef =>
supervise(targetBehavior(probeRef))
.onFailure[IllegalArgumentException](SupervisorStrategy.restart)
@@ -1135,7 +1139,8 @@ class SupervisionSpec extends ScalaTestWithActorTestKit("""
}
}
- "fail instead of restart with exponential backoff when deferred factory throws unhandled" in new FailingUnhandledTestSetup(
+ "fail instead of restart with exponential backoff when deferred factory throws unhandled" in
+ new FailingUnhandledTestSetup(
strategy = SupervisorStrategy.restartWithBackoff(minBackoff = 100.millis.dilated, maxBackoff = 1.second, 0)) {
LoggingTestKit.error[ActorInitializationException].expect {
diff --git a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/internal/ActorSystemSpec.scala b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/internal/ActorSystemSpec.scala
index a7f603f845..6ea40fabf4 100644
--- a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/internal/ActorSystemSpec.scala
+++ b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/internal/ActorSystemSpec.scala
@@ -191,7 +191,8 @@ class ActorSystemSpec
"guardian-mailbox",
Behaviors.receive[WhatsYourMailbox] {
case (context, WhatsYourMailbox(replyTo)) =>
- replyTo ! context
+ replyTo !
+ context
.asInstanceOf[ActorContextImpl[_]]
.classicActorContext
.asInstanceOf[Dispatch]
diff --git a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/delivery/DurableProducerQueue.scala b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/delivery/DurableProducerQueue.scala
index cd69940cd8..cc51c43095 100644
--- a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/delivery/DurableProducerQueue.scala
+++ b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/delivery/DurableProducerQueue.scala
@@ -174,8 +174,10 @@ object DurableProducerQueue {
override def equals(obj: Any): Boolean = {
obj match {
case other: MessageSent[_] =>
- seqNr == other.seqNr && message == other.message && ack == other.ack && confirmationQualifier == other
- .confirmationQualifier && timestampMillis == other.timestampMillis
+ seqNr == other.seqNr && message == other.message && ack == other.ack &&
+ confirmationQualifier ==
+ other
+ .confirmationQualifier && timestampMillis == other.timestampMillis
case _ => false
}
}
diff --git a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/delivery/internal/ProducerControllerImpl.scala b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/delivery/internal/ProducerControllerImpl.scala
index 2b6a9cea01..143c0ecf28 100644
--- a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/delivery/internal/ProducerControllerImpl.scala
+++ b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/delivery/internal/ProducerControllerImpl.scala
@@ -713,7 +713,8 @@ private class ProducerControllerImpl[A: ClassTag](
}
def receiveSendChunk(): Behavior[InternalCommand] = {
- if (s.remainingChunks.nonEmpty && s.remainingChunks.head.seqNr <= s.requestedSeqNr && s
+ if (s.remainingChunks.nonEmpty && s.remainingChunks.head.seqNr <= s.requestedSeqNr &&
+ s
.storeMessageSentInProgress == 0) {
if (traceEnabled)
context.log.trace("Send next chunk seqNr [{}].", s.remainingChunks.head.seqNr)
diff --git a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/Supervision.scala b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/Supervision.scala
index ca0344ba84..d6f2e67de8 100644
--- a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/Supervision.scala
+++ b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/Supervision.scala
@@ -44,7 +44,8 @@ import org.slf4j.event.Level
*/
@InternalApi private[pekko] object Supervisor {
def apply[T, Thr <: Throwable: ClassTag](initialBehavior: Behavior[T], strategy: SupervisorStrategy): Behavior[T] = {
- if (initialBehavior.isInstanceOf[scaladsl.AbstractBehavior[_]] || initialBehavior
+ if (initialBehavior.isInstanceOf[scaladsl.AbstractBehavior[_]] ||
+ initialBehavior
.isInstanceOf[javadsl.AbstractBehavior[_]]) {
throw new IllegalArgumentException(
"The supervised Behavior must not be a AbstractBehavior instance directly," +
diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorPath.scala b/actor/src/main/scala/org/apache/pekko/actor/ActorPath.scala
index fe22af4a6c..91dcd6b424 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/ActorPath.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/ActorPath.scala
@@ -127,10 +127,15 @@ object ActorPath {
if (s.isEmpty) EmptyPathCode
else {
def isValidChar(c: Char): Boolean =
- (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || (ValidSymbols.indexOf(c) != -1)
+ (c >= 'a' && c <= 'z') ||
+ (c >= 'A' && c <= 'Z') ||
+ (c >= '0' && c <= '9') ||
+ (ValidSymbols.indexOf(c) != -1)
def isHexChar(c: Char): Boolean =
- (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F') || (c >= '0' && c <= '9')
+ (c >= 'a' && c <= 'f') ||
+ (c >= 'A' && c <= 'F') ||
+ (c >= '0' && c <= '9')
val len = s.length
def validate(pos: Int): Int =
diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala b/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
index 674ad6af99..3a6cfe2900 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
@@ -349,8 +349,10 @@ object ActorSystem {
val configuredLoggingFilter = config.getString(loggingFilterConfKey)
val loggingFilterAlreadyConfigured =
- configuredLoggingFilter == slf4jLoggingFilterClassName || configuredLoggingFilter != classOf[
- DefaultLoggingFilter].getName
+ configuredLoggingFilter == slf4jLoggingFilterClassName ||
+ configuredLoggingFilter !=
+ classOf[
+ DefaultLoggingFilter].getName
def newLoggingFilterConfStr = s"""$loggingFilterConfKey = "$slf4jLoggingFilterClassName""""
@@ -1276,7 +1278,8 @@ private[pekko] class ActorSystemImpl(
case real: ActorCell => " status=" + real.mailbox.currentStatus
case _ => ""
}) +
- " " + (cell.childrenRefs match {
+ " " +
+ (cell.childrenRefs match {
case ChildrenContainer.TerminatingChildrenContainer(_, toDie, reason) =>
"Terminating(" + reason + ")" +
(toDie.toSeq.sorted.mkString("\n" + indent + " | toDie: ", "\n" + indent + " | ", ""))
diff --git a/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala b/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
index a818dc5d50..71839865c5 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
@@ -271,11 +271,12 @@ class LightArrayRevolverScheduler(config: Config, log: LoggingAdapter, threadFac
node.value.ticks match {
case 0 => node.value.executeTask()
case ticks =>
- val futureTick = ((
- time - start + // calculate the nanos since timer start
- (ticks * tickNanos) + // adding the desired delay
- tickNanos - 1 // rounding up
- ) / tickNanos).toInt // and converting to slot number
+ val futureTick =
+ ((
+ time - start + // calculate the nanos since timer start
+ (ticks * tickNanos) + // adding the desired delay
+ tickNanos - 1 // rounding up
+ ) / tickNanos).toInt // and converting to slot number
// tick is an Int that will wrap around, but toInt of futureTick gives us modulo operations
// and the difference (offset) will be correct in any case
val offset = futureTick - tick
diff --git a/actor/src/main/scala/org/apache/pekko/actor/Stash.scala b/actor/src/main/scala/org/apache/pekko/actor/Stash.scala
index 823253f0ab..5d335d68b4 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/Stash.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/Stash.scala
@@ -20,10 +20,7 @@ import org.apache.pekko
import pekko.PekkoException
import pekko.annotation.InternalStableApi
import pekko.dispatch.{
- DequeBasedMessageQueueSemantics,
- Envelope,
- RequiresMessageQueue,
- UnboundedDequeBasedMessageQueueSemantics
+ DequeBasedMessageQueueSemantics, Envelope, RequiresMessageQueue, UnboundedDequeBasedMessageQueueSemantics
}
/**
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/Mailboxes.scala b/actor/src/main/scala/org/apache/pekko/dispatch/Mailboxes.scala
index e525f6f703..485c55f808 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/Mailboxes.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/Mailboxes.scala
@@ -26,10 +26,7 @@ import pekko.ConfigurationException
import pekko.actor.{ Actor, ActorRef, ActorSystem, DeadLetter, Deploy, DynamicAccess, Props }
import pekko.annotation.InternalStableApi
import pekko.dispatch.sysmsg.{
- EarliestFirstSystemMessageList,
- LatestFirstSystemMessageList,
- SystemMessage,
- SystemMessageList
+ EarliestFirstSystemMessageList, LatestFirstSystemMessageList, SystemMessage, SystemMessageList
}
import pekko.event.EventStream
import pekko.event.Logging.Warning
@@ -194,8 +191,10 @@ private[pekko] class Mailboxes(
if (deploy.mailbox != Deploy.NoMailboxGiven) {
verifyRequirements(lookup(deploy.mailbox))
- } else if (deploy.dispatcher != Deploy.NoDispatcherGiven && deploy.dispatcher != Deploy
- .DispatcherSameAsParent && hasMailboxType) {
+ } else if (deploy.dispatcher != Deploy.NoDispatcherGiven &&
+ deploy.dispatcher !=
+ Deploy
+ .DispatcherSameAsParent && hasMailboxType) {
verifyRequirements(lookup(dispatcherConfig.getString("id")))
} else if (hasRequiredType(actorClass)) {
try verifyRequirements(lookupByQueueType(getRequiredType(actorClass)))
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala b/actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala
index e1321c7cab..af37f95cd4 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala
@@ -15,20 +15,9 @@ package org.apache.pekko.dispatch
import java.util.Collection
import java.util.concurrent.{
- ArrayBlockingQueue,
- BlockingQueue,
- Callable,
- Executor,
- ExecutorService,
- ForkJoinPool,
- ForkJoinWorkerThread,
- LinkedBlockingQueue,
- RejectedExecutionException,
- RejectedExecutionHandler,
- SynchronousQueue,
- ThreadFactory,
- ThreadPoolExecutor,
- TimeUnit
+ ArrayBlockingQueue, BlockingQueue, Callable, Executor, ExecutorService, ForkJoinPool, ForkJoinWorkerThread,
+ LinkedBlockingQueue, RejectedExecutionException, RejectedExecutionHandler, SynchronousQueue, ThreadFactory,
+ ThreadPoolExecutor, TimeUnit
}
import java.util.concurrent.atomic.{ AtomicLong, AtomicReference }
diff --git a/actor/src/main/scala/org/apache/pekko/event/EventStream.scala b/actor/src/main/scala/org/apache/pekko/event/EventStream.scala
index 0c460f5af1..35bcac9684 100644
--- a/actor/src/main/scala/org/apache/pekko/event/EventStream.scala
+++ b/actor/src/main/scala/org/apache/pekko/event/EventStream.scala
@@ -103,7 +103,8 @@ class EventStream(sys: ActorSystem, private val debug: Boolean) extends LoggingB
Logging.Debug(
simpleName(this),
this.getClass,
- "initialized unsubscriber to: " + unsubscriber + ", registering " + subscribers
+ "initialized unsubscriber to: " + unsubscriber + ", registering " +
+ subscribers
.size + " initial subscribers with it"))
subscribers.foreach(registerWithUnsubscriber)
true
diff --git a/actor/src/main/scala/org/apache/pekko/event/LoggingReceive.scala b/actor/src/main/scala/org/apache/pekko/event/LoggingReceive.scala
index 2cddd32b1e..95a8f87119 100644
--- a/actor/src/main/scala/org/apache/pekko/event/LoggingReceive.scala
+++ b/actor/src/main/scala/org/apache/pekko/event/LoggingReceive.scala
@@ -89,8 +89,9 @@ class LoggingReceive(source: Option[AnyRef], r: Receive, label: Option[String],
if (context.system.eventStream.logLevel >= logLevel) {
val src = source.getOrElse(context.asInstanceOf[ActorCell].actor)
val (str, clazz) = LogSource.fromAnyRef(src)
- val message = "received " + (if (handled) "handled" else "unhandled") + " message " + o + " from " + context
- .sender() +
+ val message = "received " + (if (handled) "handled" else "unhandled") + " message " + o + " from " +
+ context
+ .sender() +
(label match {
case Some(l) => " in state " + l
case _ => ""
diff --git a/actor/src/main/scala/org/apache/pekko/io/dns/internal/TcpDnsClient.scala b/actor/src/main/scala/org/apache/pekko/io/dns/internal/TcpDnsClient.scala
index 6d32da28c8..a85f36d073 100644
--- a/actor/src/main/scala/org/apache/pekko/io/dns/internal/TcpDnsClient.scala
+++ b/actor/src/main/scala/org/apache/pekko/io/dns/internal/TcpDnsClient.scala
@@ -102,8 +102,7 @@ private[internal] object TcpDnsClient {
def encodeLength(length: Int): ByteString =
ByteString((length / 256).toByte, length.toByte)
- def decodeLength(data: ByteString): Int =
- ((data(0).toInt + 256) % 256) * 256 + ((data(1) + 256) % 256)
+ def decodeLength(data: ByteString): Int = ((data(0).toInt + 256) % 256) * 256 + ((data(1) + 256) % 256)
def throwFailure(message: String, cause: Option[Throwable]): Unit =
cause match {
diff --git a/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala b/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala
index f74a0c2ee4..75d4a0feca 100644
--- a/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala
+++ b/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala
@@ -758,7 +758,8 @@ class CircuitBreaker(
}
private def isIgnoredException(ex: Any): Boolean =
- allowExceptions.nonEmpty && (ex match {
+ allowExceptions.nonEmpty &&
+ (ex match {
case ce: CompletionException => (ce.getCause ne null) && allowExceptions.contains(ce.getCause.getClass.getName)
case _ => allowExceptions.contains(ex.getClass.getName)
})
diff --git a/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreakersRegistry.scala b/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreakersRegistry.scala
index c3694e0b0d..a2388c5817 100644
--- a/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreakersRegistry.scala
+++ b/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreakersRegistry.scala
@@ -20,12 +20,7 @@ import scala.jdk.CollectionConverters._
import org.apache.pekko
import pekko.actor.{
- ActorSystem,
- ClassicActorSystemProvider,
- ExtendedActorSystem,
- Extension,
- ExtensionId,
- ExtensionIdProvider
+ ActorSystem, ClassicActorSystemProvider, ExtendedActorSystem, Extension, ExtensionId, ExtensionIdProvider
}
import pekko.pattern.internal.CircuitBreakerTelemetryProvider
diff --git a/actor/src/main/scala/org/apache/pekko/pattern/Patterns.scala b/actor/src/main/scala/org/apache/pekko/pattern/Patterns.scala
index d39c5862b8..9a7d44c161 100644
--- a/actor/src/main/scala/org/apache/pekko/pattern/Patterns.scala
+++ b/actor/src/main/scala/org/apache/pekko/pattern/Patterns.scala
@@ -35,12 +35,8 @@ object Patterns {
import pekko.actor.ActorRef
import pekko.japi
import pekko.pattern.{
- after => scalaAfter,
- ask => scalaAsk,
- askWithStatus => scalaAskWithStatus,
- gracefulStop => scalaGracefulStop,
- pipe => scalaPipe,
- retry => scalaRetry
+ after => scalaAfter, ask => scalaAsk, askWithStatus => scalaAskWithStatus, gracefulStop => scalaGracefulStop,
+ pipe => scalaPipe, retry => scalaRetry
}
import pekko.util.Timeout
diff --git a/actor/src/main/scala/org/apache/pekko/pattern/internal/BackoffOnRestartSupervisor.scala b/actor/src/main/scala/org/apache/pekko/pattern/internal/BackoffOnRestartSupervisor.scala
index b9dbefb557..6140a28796 100644
--- a/actor/src/main/scala/org/apache/pekko/pattern/internal/BackoffOnRestartSupervisor.scala
+++ b/actor/src/main/scala/org/apache/pekko/pattern/internal/BackoffOnRestartSupervisor.scala
@@ -20,13 +20,7 @@ import pekko.actor.{ OneForOneStrategy, _ }
import pekko.actor.SupervisorStrategy._
import pekko.annotation.InternalApi
import pekko.pattern.{
- BackoffReset,
- BackoffSupervisor,
- ForwardDeathLetters,
- ForwardTo,
- HandleBackoff,
- HandlingWhileStopped,
- ReplyWith,
+ BackoffReset, BackoffSupervisor, ForwardDeathLetters, ForwardTo, HandleBackoff, HandlingWhileStopped, ReplyWith,
RetrySupport
}
diff --git a/actor/src/main/scala/org/apache/pekko/pattern/internal/BackoffOnStopSupervisor.scala b/actor/src/main/scala/org/apache/pekko/pattern/internal/BackoffOnStopSupervisor.scala
index 20f6ff6585..ae001b0467 100644
--- a/actor/src/main/scala/org/apache/pekko/pattern/internal/BackoffOnStopSupervisor.scala
+++ b/actor/src/main/scala/org/apache/pekko/pattern/internal/BackoffOnStopSupervisor.scala
@@ -20,13 +20,7 @@ import pekko.actor.{ Actor, ActorLogging, OneForOneStrategy, Props, SupervisorSt
import pekko.actor.SupervisorStrategy.{ Directive, Escalate }
import pekko.annotation.InternalApi
import pekko.pattern.{
- BackoffReset,
- BackoffSupervisor,
- ForwardDeathLetters,
- ForwardTo,
- HandleBackoff,
- HandlingWhileStopped,
- ReplyWith,
+ BackoffReset, BackoffSupervisor, ForwardDeathLetters, ForwardTo, HandleBackoff, HandlingWhileStopped, ReplyWith,
RetrySupport
}
diff --git a/actor/src/main/scala/org/apache/pekko/routing/ConsistentHash.scala b/actor/src/main/scala/org/apache/pekko/routing/ConsistentHash.scala
index 3d4e43ae5e..4020daae4a 100644
--- a/actor/src/main/scala/org/apache/pekko/routing/ConsistentHash.scala
+++ b/actor/src/main/scala/org/apache/pekko/routing/ConsistentHash.scala
@@ -48,7 +48,8 @@ class ConsistentHash[T: ClassTag] private (nodes: immutable.SortedMap[Int, T], v
*/
def :+(node: T): ConsistentHash[T] = {
val nodeHash = hashFor(node.toString)
- new ConsistentHash(nodes ++ ((1 to virtualNodesFactor).map { r =>
+ new ConsistentHash(nodes ++
+ ((1 to virtualNodesFactor).map { r =>
concatenateNodeHash(nodeHash, r) -> node
}), virtualNodesFactor)
}
@@ -67,7 +68,8 @@ class ConsistentHash[T: ClassTag] private (nodes: immutable.SortedMap[Int, T], v
*/
def :-(node: T): ConsistentHash[T] = {
val nodeHash = hashFor(node.toString)
- new ConsistentHash(nodes -- ((1 to virtualNodesFactor).map { r =>
+ new ConsistentHash(nodes --
+ ((1 to virtualNodesFactor).map { r =>
concatenateNodeHash(nodeHash, r)
}), virtualNodesFactor)
}
diff --git a/actor/src/main/scala/org/apache/pekko/util/FrequencySketch.scala b/actor/src/main/scala/org/apache/pekko/util/FrequencySketch.scala
index dea5071a1a..90d96df312 100644
--- a/actor/src/main/scala/org/apache/pekko/util/FrequencySketch.scala
+++ b/actor/src/main/scala/org/apache/pekko/util/FrequencySketch.scala
@@ -395,7 +395,9 @@ private[pekko] object FrequencySketchUtil {
val indexDigits = digits(rowWidth)
val counterDigits = math.max(2, digits(counterMax))
def divider(start: String, line: String, separator1: String, separator: String, end: String): String =
- start + (line * (indexDigits + 2)) + separator1 + (line * (counterDigits + 2)) +
+ start +
+ (line * (indexDigits + 2)) + separator1 +
+ (line * (counterDigits + 2)) +
((separator + (line * (counterDigits + 2))) * (slots - 1)) + end + "\n"
val builder = new StringBuilder
builder ++= divider("╔", "═", "╦", "╤", "╗")
@@ -407,8 +409,9 @@ private[pekko] object FrequencySketchUtil {
builder ++= "║\n"
for (row <- matrix.indices) {
for (column <- matrix(0).indices) {
- builder ++= (if (column == 0) divider("╠", "═", "╬", "╪", "╣")
- else divider("╟", "─", "╫", "┼", "╢"))
+ builder ++=
+ (if (column == 0) divider("╠", "═", "╬", "╪", "╣")
+ else divider("╟", "─", "╫", "┼", "╢"))
builder ++= s"║ %${indexDigits}d ".format(column)
var shift = 0
while (shift < 64) {
diff --git a/actor/src/main/scala/org/apache/pekko/util/SWARUtil.scala b/actor/src/main/scala/org/apache/pekko/util/SWARUtil.scala
index 8f7cc25ffd..cabbb52c1c 100644
--- a/actor/src/main/scala/org/apache/pekko/util/SWARUtil.scala
+++ b/actor/src/main/scala/org/apache/pekko/util/SWARUtil.scala
@@ -41,8 +41,7 @@ private[util] object SWARUtil {
/**
* Compiles given byte into a long pattern suitable for SWAR operations.
*/
- def compilePattern(byteToFind: Byte): Long =
- (byteToFind & 0xFFL) * 0x101010101010101L
+ def compilePattern(byteToFind: Byte): Long = (byteToFind & 0xFFL) * 0x101010101010101L
/**
* Applies a compiled pattern to given word.
diff --git a/bench-jmh/src/main/scala/org/apache/pekko/actor/typed/delivery/ReliableDeliveryBenchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/actor/typed/delivery/ReliableDeliveryBenchmark.scala
index 654bbaf6b4..da331671e3 100644
--- a/bench-jmh/src/main/scala/org/apache/pekko/actor/typed/delivery/ReliableDeliveryBenchmark.scala
+++ b/bench-jmh/src/main/scala/org/apache/pekko/actor/typed/delivery/ReliableDeliveryBenchmark.scala
@@ -122,8 +122,9 @@ object WorkPullingProducer {
Behaviors.setup { context =>
val requestNextAdapter =
context.messageAdapter[WorkPullingProducerController.RequestNext[Consumer.Command]](WrappedRequestNext(_))
- var remaining = numberOfMessages + context.system.settings.config
- .getInt("pekko.reliable-delivery.consumer-controller.flow-control-window")
+ var remaining = numberOfMessages +
+ context.system.settings.config
+ .getInt("pekko.reliable-delivery.consumer-controller.flow-control-window")
Behaviors.receiveMessagePartial {
case WrappedRequestNext(next) =>
diff --git a/bench-jmh/src/main/scala/org/apache/pekko/io/dns/IdGeneratorBanchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/io/dns/IdGeneratorBanchmark.scala
index a6fcf21457..dd4ac13f24 100644
--- a/bench-jmh/src/main/scala/org/apache/pekko/io/dns/IdGeneratorBanchmark.scala
+++ b/bench-jmh/src/main/scala/org/apache/pekko/io/dns/IdGeneratorBanchmark.scala
@@ -21,16 +21,7 @@ import java.security.SecureRandom
import java.util.concurrent.{ ThreadLocalRandom, TimeUnit }
import org.openjdk.jmh.annotations.{
- Benchmark,
- BenchmarkMode,
- Fork,
- Measurement,
- Mode,
- OutputTimeUnit,
- Scope,
- State,
- Threads,
- Warmup
+ Benchmark, BenchmarkMode, Fork, Measurement, Mode, OutputTimeUnit, Scope, State, Threads, Warmup
}
@BenchmarkMode(Array(Mode.Throughput))
diff --git a/build.sbt b/build.sbt
index 9027ac3e50..7cd56400e9 100644
--- a/build.sbt
+++ b/build.sbt
@@ -370,8 +370,9 @@ lazy val protobufV3 = pekkoModule("protobuf-v3")
assembly / assemblyOption := (assembly / assemblyOption).value.withIncludeScala(false).withIncludeBin(true),
autoScalaLibrary := false, // do not include scala dependency in pom
exportJars := true, // in dependent projects, use assembled and shaded jar
- makePomConfiguration := makePomConfiguration.value
- .withConfigurations(Vector(Compile)), // prevent original dependency to be added to pom as runtime dep
+ makePomConfiguration :=
+ makePomConfiguration.value
+ .withConfigurations(Vector(Compile)), // prevent original dependency to be added to pom as runtime dep
Compile / packageBin / packagedArtifact := Scoped.mkTuple2(
(Compile / packageBin / artifact).value,
ReproducibleBuildsPlugin.postProcessJar(OsgiKeys.bundle.value)),
@@ -392,7 +393,8 @@ lazy val protobufV3 = pekkoModule("protobuf-v3")
// redefining the fullClasspath with just what we need to avoid the cyclic task dependency
assembly / fullClasspath := (Runtime / managedClasspath).value ++ (Compile / products).value.map(Attributed.blank),
assembly / test := {}, // assembly runs tests for unknown reason which introduces another cyclic dependency to packageBin via exportedJars
- description := s"Apache Pekko Protobuf V3 is a shaded version of the protobuf runtime. Original POM: https://github.com/protocolbuffers/protobuf/blob/v${Dependencies.Protobuf.protobufJavaVersion}/java/pom.xml")
+ description :=
+ s"Apache Pekko Protobuf V3 is a shaded version of the protobuf runtime. Original POM: https://github.com/protocolbuffers/protobuf/blob/v${Dependencies.Protobuf.protobufJavaVersion}/java/pom.xml")
lazy val pki =
pekkoModule("pki")
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/delivery/internal/ShardingProducerControllerImpl.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/delivery/internal/ShardingProducerControllerImpl.scala
index 0ee5a06808..19ad5d4cf2 100644
--- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/delivery/internal/ShardingProducerControllerImpl.scala
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/delivery/internal/ShardingProducerControllerImpl.scala
@@ -488,7 +488,8 @@ 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.resendFirstUnconfirmedIdleTimeout.toMillis) {
+ if (outState.unconfirmed.nonEmpty &&
+ idleDurationMillis >= settings.resendFirstUnconfirmedIdleTimeout.toMillis) {
context.log.debug(
"Resend first unconfirmed for [{}], because it was idle for [{} ms]",
outKey,
@@ -506,8 +507,10 @@ private class ShardingProducerControllerImpl[A: ClassTag](
s.out.flatMap {
case (outKey: OutKey, outState) =>
val idleDurationMillis = (now - outState.usedNanoTime) / 1000 / 1000
- if (outState.unconfirmed.isEmpty && outState.buffered.isEmpty && idleDurationMillis >= settings
- .cleanupUnusedAfter.toMillis) {
+ if (outState.unconfirmed.isEmpty && outState.buffered.isEmpty &&
+ idleDurationMillis >=
+ settings
+ .cleanupUnusedAfter.toMillis) {
context.log.debug("Cleanup unused [{}], because it was idle for [{} ms]", outKey, idleDurationMillis)
context.stop(outState.producerController)
Some(outKey)
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/Murmur2.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/Murmur2.scala
index e5873d9a29..3f6be73c94 100644
--- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/Murmur2.scala
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/Murmur2.scala
@@ -42,8 +42,9 @@ private[sharding] object Murmur2 {
val length4 = length / 4
for (i <- 0 until length4) {
val i4 = i * 4
- var k = (data(i4 + 0) & 0xFF) + ((data(i4 + 1) & 0xFF) << 8) + ((data(i4 + 2) & 0xFF) << 16) + ((data(
- i4 + 3) & 0xFF) << 24)
+ var k = (data(i4 + 0) & 0xFF) + ((data(i4 + 1) & 0xFF) << 8) + ((data(i4 + 2) & 0xFF) << 16) +
+ ((data(
+ i4 + 3) & 0xFF) << 24)
k *= m
k ^= k >>> r
k *= m
diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ReplicatedShardingExtensionImpl.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ReplicatedShardingExtensionImpl.scala
index 4d792134d0..8f66a234d3 100644
--- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ReplicatedShardingExtensionImpl.scala
+++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ReplicatedShardingExtensionImpl.scala
@@ -101,7 +101,8 @@ private[pekko] final class ReplicatedShardingImpl[M](
override def entityRefsFor(entityId: String): Map[ReplicaId, EntityRef[M]] =
replicaTypeKeys.map {
case (replicaId, (typeKey, dc, typeName)) =>
- replicaId -> (dc match {
+ replicaId ->
+ (dc match {
case None => sharding.entityRefFor(typeKey, ReplicationId(typeName, entityId, replicaId).persistenceId.id)
case Some(dc) =>
sharding.entityRefFor(typeKey, ReplicationId(typeName, entityId, replicaId).persistenceId.id, dc)
diff --git a/cluster-sharding-typed/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/typed/delivery/DeliveryThroughputSpec.scala b/cluster-sharding-typed/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/typed/delivery/DeliveryThroughputSpec.scala
index 0fa92bd494..47a4711a7f 100644
--- a/cluster-sharding-typed/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/typed/delivery/DeliveryThroughputSpec.scala
+++ b/cluster-sharding-typed/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/typed/delivery/DeliveryThroughputSpec.scala
@@ -207,8 +207,9 @@ object DeliveryThroughputSpec extends MultiNodeConfig {
val requestNextAdapter =
context.messageAdapter[WorkPullingProducerController.RequestNext[Consumer.Command]](WrappedRequestNext(_))
var startTime = System.nanoTime()
- var remaining = numberOfMessages + context.system.settings.config
- .getInt("pekko.reliable-delivery.consumer-controller.flow-control-window")
+ var remaining = numberOfMessages +
+ context.system.settings.config
+ .getInt("pekko.reliable-delivery.consumer-controller.flow-control-window")
Behaviors.receiveMessage {
case WrappedRequestNext(next) =>
@@ -254,8 +255,9 @@ object DeliveryThroughputSpec extends MultiNodeConfig {
val requestNextAdapter =
context.messageAdapter[ShardingProducerController.RequestNext[Consumer.Command]](WrappedRequestNext(_))
var startTime = System.nanoTime()
- var remaining = numberOfMessages + context.system.settings.config
- .getInt("pekko.reliable-delivery.sharding.consumer-controller.flow-control-window")
+ var remaining = numberOfMessages +
+ context.system.settings.config
+ .getInt("pekko.reliable-delivery.sharding.consumer-controller.flow-control-window")
var latestDemand: ShardingProducerController.RequestNext[Consumer.Command] = null
var messagesSentToEachEntity: Map[String, Long] = Map.empty[String, Long].withDefaultValue(0L)
diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ClusterShardingSettings.scala b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ClusterShardingSettings.scala
index 2efbf1d551..f57df942b3 100644
--- a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ClusterShardingSettings.scala
+++ b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ClusterShardingSettings.scala
@@ -975,7 +975,8 @@ final class ClusterShardingSettings(
import ClusterShardingSettings.{ RememberEntitiesStoreCustom, StateStoreModeDData, StateStoreModePersistence }
require(
- stateStoreMode == StateStoreModePersistence || stateStoreMode == StateStoreModeDData || stateStoreMode == RememberEntitiesStoreCustom,
+ stateStoreMode == StateStoreModePersistence || stateStoreMode == StateStoreModeDData ||
+ stateStoreMode == RememberEntitiesStoreCustom,
s"Unknown 'state-store-mode' [$stateStoreMode], valid values are '$StateStoreModeDData' or '$StateStoreModePersistence'")
/** If true, this node should run the shard region, otherwise just a shard proxy should started on this node. */
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 1af3ca7d3a..357e3bb4ed 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
@@ -34,9 +34,7 @@ import pekko.cluster.ddata.Replicator._
import pekko.cluster.ddata.SelfUniqueAddress
import pekko.cluster.sharding.ShardRegion.ShardId
import pekko.cluster.sharding.internal.{
- EventSourcedRememberEntitiesCoordinatorStore,
- RememberEntitiesCoordinatorStore,
- RememberEntitiesProvider
+ EventSourcedRememberEntitiesCoordinatorStore, RememberEntitiesCoordinatorStore, RememberEntitiesProvider
}
import pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy
import pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy.RegionEntry
@@ -1205,7 +1203,8 @@ abstract class ShardCoordinator(
state.shards.get(shard) match {
case Some(ref) => getShardHomeSender ! ShardHome(shard, ref)
case None =>
- if (state.regions.contains(region) && !gracefulShutdownInProgress(region) && !regionTerminationInProgress
+ if (state.regions.contains(region) && !gracefulShutdownInProgress(region) &&
+ !regionTerminationInProgress
.contains(region)) {
update(ShardHomeAllocated(shard, region)) { evt =>
state = state.updated(evt)
diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala
index 63eb189ca8..eb5f263aec 100644
--- a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala
+++ b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardRegion.scala
@@ -1005,7 +1005,8 @@ private[pekko] class ShardRegion(
shardsByRef = shardsByRef - ref
shards = shards - shardId
startingShards -= shardId
- if (settings.passivationStrategy != ClusterShardingSettings.NoPassivationStrategy && !gracefulShutdownInProgress) {
+ if (settings.passivationStrategy != ClusterShardingSettings.NoPassivationStrategy &&
+ !gracefulShutdownInProgress) {
val update = ShardsUpdated(shards.size)
shards.values.foreach(_ ! update)
}
@@ -1086,11 +1087,10 @@ private[pekko] class ShardRegion(
}
private def askOne[T: ClassTag](shard: ActorRef, msg: Any, shardId: ShardId)(
- implicit timeout: Timeout): Future[Either[ShardId, T]] =
- (shard ? msg).mapTo[T].transform {
- case Success(t) => Success(Right(t))
- case Failure(_) => Success(Left(shardId))
- }
+ implicit timeout: Timeout): Future[Either[ShardId, T]] = (shard ? msg).mapTo[T].transform {
+ case Success(t) => Success(Right(t))
+ case Failure(_) => Success(Left(shardId))
+ }
private def tryCompleteGracefulShutdownIfInProgress(): Unit =
if (gracefulShutdownInProgress && shards.isEmpty && shardBuffers.isEmpty) {
diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/LeastShardAllocationStrategy.scala b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/LeastShardAllocationStrategy.scala
index c2d2a54df4..86be1f4481 100644
--- a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/LeastShardAllocationStrategy.scala
+++ b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/internal/LeastShardAllocationStrategy.scala
@@ -115,8 +115,9 @@ import pekko.cluster.sharding.internal.AbstractLeastShardAllocationStrategy.Shar
if (numberOfRegions == 0 || numberOfShards == 0) {
emptyRebalanceResult
} else {
- val optimalPerRegion = numberOfShards / numberOfRegions + (if (numberOfShards % numberOfRegions == 0) 0
- else 1)
+ val optimalPerRegion = numberOfShards / numberOfRegions +
+ (if (numberOfShards % numberOfRegions == 0) 0
+ else 1)
val result1 = rebalancePhase1(numberOfShards, optimalPerRegion, sortedRegionEntries)
diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala
index 6951a109a8..de9ad97b0a 100644
--- a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala
+++ b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/protobuf/ClusterShardingMessageSerializer.scala
@@ -31,8 +31,7 @@ import pekko.cluster.sharding.Shard
import pekko.cluster.sharding.ShardCoordinator
import pekko.cluster.sharding.ShardRegion._
import pekko.cluster.sharding.internal.EventSourcedRememberEntitiesCoordinatorStore.{
- MigrationMarker,
- State => RememberShardsState
+ MigrationMarker, State => RememberShardsState
}
import pekko.cluster.sharding.internal.EventSourcedRememberEntitiesShardStore.{ EntitiesStarted, EntitiesStopped }
import pekko.cluster.sharding.internal.EventSourcedRememberEntitiesShardStore.{ State => EntityState }
diff --git a/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sbr/RandomizedBrainResolverIntegrationSpec.scala b/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sbr/RandomizedBrainResolverIntegrationSpec.scala
index 395ec70356..a0380d157f 100644
--- a/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sbr/RandomizedBrainResolverIntegrationSpec.scala
+++ b/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sbr/RandomizedBrainResolverIntegrationSpec.scala
@@ -297,13 +297,12 @@ class RandomizedSplitBrainResolverIntegrationSpec
// The test is limited to one flaky step, see issue #29185.
val numberOfFlaky = if (cleanSplit) 0 else 1
val healLastFlaky = numberOfFlaky > 0 && random.nextBoolean()
- val flaky: Map[Int, (RoleName, List[RoleName])] =
- (0 until numberOfFlaky).map { i =>
- val from = nodes(random.nextInt(nodes.size))
- val targets = nodes.filterNot(_ == from)
- val to = (0 to random.nextInt(math.min(5, targets.size))).map(j => targets(j)).toList
- i -> (from -> to)
- }.toMap
+ val flaky: Map[Int, (RoleName, List[RoleName])] = (0 until numberOfFlaky).map { i =>
+ val from = nodes(random.nextInt(nodes.size))
+ val targets = nodes.filterNot(_ == from)
+ val to = (0 to random.nextInt(math.min(5, targets.size))).map(j => targets(j)).toList
+ i -> (from -> to)
+ }.toMap
val delays = (0 until 10).map(_ => 2 + random.nextInt(13))
diff --git a/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/MultiDcClusterShardingSpec.scala b/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/MultiDcClusterShardingSpec.scala
index fca967bbcb..c7a9c62da3 100644
--- a/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/MultiDcClusterShardingSpec.scala
+++ b/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/MultiDcClusterShardingSpec.scala
@@ -96,7 +96,8 @@ abstract class MultiDcClusterShardingSpec
to, {
startSharding()
withClue(
- s"Failed waiting for ${cluster.selfUniqueAddress} to be up. Current state: ${cluster.state}" + cluster.state) {
+ s"Failed waiting for ${cluster.selfUniqueAddress} to be up. Current state: ${cluster.state}" +
+ cluster.state) {
within(15.seconds) {
awaitAssert(cluster.state.members.exists { m =>
m.uniqueAddress == cluster.selfUniqueAddress && m.status == MemberStatus.Up
diff --git a/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/MultiNodeClusterShardingConfig.scala b/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/MultiNodeClusterShardingConfig.scala
index 2abb0202a5..16ec015e77 100644
--- a/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/MultiNodeClusterShardingConfig.scala
+++ b/cluster-sharding/src/multi-jvm/scala/org/apache/pekko/cluster/sharding/MultiNodeClusterShardingConfig.scala
@@ -108,7 +108,8 @@ abstract class MultiNodeClusterShardingConfig(
"").replace("_", "")}"
val persistenceConfig: Config =
- if (mode == ClusterShardingSettings.StateStoreModeDData && rememberEntitiesStore != ClusterShardingSettings.RememberEntitiesStoreEventsourced)
+ if (mode == ClusterShardingSettings.StateStoreModeDData &&
+ rememberEntitiesStore != ClusterShardingSettings.RememberEntitiesStoreEventsourced)
ConfigFactory.empty
else MultiNodeClusterShardingConfig.persistenceConfig(targetDir)
diff --git a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/JoinConfigCompatCheckShardingSpec.scala b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/JoinConfigCompatCheckShardingSpec.scala
index d0f22bac41..6ca482663b 100644
--- a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/JoinConfigCompatCheckShardingSpec.scala
+++ b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/JoinConfigCompatCheckShardingSpec.scala
@@ -48,7 +48,8 @@ class JoinConfigCompatCheckShardingSpec extends PekkoSpec() with WithLogCapturin
"A Joining Node" must {
/** This test verifies the built-in JoinConfigCompatCheckerSharding */
- "NOT be allowed to join a cluster using a different value for pekko.cluster.sharding.state-store-mode" taggedAs LongRunningTest in {
+ "NOT be allowed to join a cluster using a different value for pekko.cluster.sharding.state-store-mode" taggedAs
+ LongRunningTest in {
val joinNodeConfig =
ConfigFactory.parseString("""
diff --git a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/LeastShardAllocationStrategyRandomizedSpec.scala b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/LeastShardAllocationStrategyRandomizedSpec.scala
index 68e398318a..d8437d823b 100644
--- a/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/LeastShardAllocationStrategyRandomizedSpec.scala
+++ b/cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/LeastShardAllocationStrategyRandomizedSpec.scala
@@ -30,11 +30,7 @@ import pekko.testkit.PekkoSpec
class LeastShardAllocationStrategyRandomizedSpec extends PekkoSpec("pekko.loglevel = INFO") {
import LeastShardAllocationStrategySpec.{
- afterRebalance,
- countShards,
- countShardsPerRegion,
- newFakeRegion,
- newUpMember
+ afterRebalance, countShards, countShardsPerRegion, newFakeRegion, newUpMember
}
@volatile var clusterMembers: SortedSet[Member] = SortedSet.empty
diff --git a/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/DistributedPubSubMediator.scala b/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/DistributedPubSubMediator.scala
index 47339e5e28..09dfef508c 100644
--- a/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/DistributedPubSubMediator.scala
+++ b/cluster-tools/src/main/scala/org/apache/pekko/cluster/pubsub/DistributedPubSubMediator.scala
@@ -831,9 +831,10 @@ class DistributedPubSubMediator(settings: DistributedPubSubSettings)
def put(key: String, valueOption: Option[ActorRef]): Unit = {
val bucket = registry(selfAddress)
val v = nextVersion()
- registry += (selfAddress -> bucket.copy(
- version = v,
- content = bucket.content + (key -> ValueHolder(v, valueOption))))
+ registry +=
+ (selfAddress -> bucket.copy(
+ version = v,
+ content = bucket.content + (key -> ValueHolder(v, valueOption))))
}
def getCurrentTopics(): Set[String] = {
diff --git a/cluster-typed/src/main/scala/org/apache/pekko/cluster/ddata/typed/internal/ReplicatorBehavior.scala b/cluster-typed/src/main/scala/org/apache/pekko/cluster/ddata/typed/internal/ReplicatorBehavior.scala
index a363add871..9988fb3076 100644
--- a/cluster-typed/src/main/scala/org/apache/pekko/cluster/ddata/typed/internal/ReplicatorBehavior.scala
+++ b/cluster-typed/src/main/scala/org/apache/pekko/cluster/ddata/typed/internal/ReplicatorBehavior.scala
@@ -87,19 +87,18 @@ import pekko.util.Timeout
case t => t.toScala + additionalAskTimeout
})
import ctx.executionContext
- val reply =
- (classicReplicator ? dd.Replicator.Get(cmd.key, cmd.consistency.toClassic))
- .mapTo[dd.Replicator.GetResponse[d]]
- .map {
- case rsp: dd.Replicator.GetSuccess[d] =>
- JReplicator.GetSuccess(rsp.key)(rsp.dataValue)
- case rsp: dd.Replicator.NotFound[d] => JReplicator.NotFound(rsp.key)
- case rsp: dd.Replicator.GetFailure[d] => JReplicator.GetFailure(rsp.key)
- case rsp: dd.Replicator.GetDataDeleted[d] => JReplicator.GetDataDeleted(rsp.key)
- }
- .recover {
- case _ => JReplicator.GetFailure(cmd.key)
- }
+ val reply = (classicReplicator ? dd.Replicator.Get(cmd.key, cmd.consistency.toClassic))
+ .mapTo[dd.Replicator.GetResponse[d]]
+ .map {
+ case rsp: dd.Replicator.GetSuccess[d] =>
+ JReplicator.GetSuccess(rsp.key)(rsp.dataValue)
+ case rsp: dd.Replicator.NotFound[d] => JReplicator.NotFound(rsp.key)
+ case rsp: dd.Replicator.GetFailure[d] => JReplicator.GetFailure(rsp.key)
+ case rsp: dd.Replicator.GetDataDeleted[d] => JReplicator.GetDataDeleted(rsp.key)
+ }
+ .recover {
+ case _ => JReplicator.GetFailure(cmd.key)
+ }
reply.foreach { cmd.replyTo ! _ }
Behaviors.same
@@ -182,19 +181,18 @@ import pekko.util.Timeout
case t => t.toScala + additionalAskTimeout
})
import ctx.executionContext
- val reply =
- (classicReplicator ? dd.Replicator.Delete(cmd.key, cmd.consistency.toClassic))
- .mapTo[dd.Replicator.DeleteResponse[d]]
- .map {
- case rsp: dd.Replicator.DeleteSuccess[d] => JReplicator.DeleteSuccess(rsp.key)
- case rsp: dd.Replicator.ReplicationDeleteFailure[d] =>
- JReplicator.DeleteFailure(rsp.key)
- case rsp: dd.Replicator.DataDeleted[d] => JReplicator.DataDeleted(rsp.key)
- case rsp: dd.Replicator.StoreFailure[d] => JReplicator.StoreFailure(rsp.key)
- }
- .recover {
- case _ => JReplicator.DeleteFailure(cmd.key)
- }
+ val reply = (classicReplicator ? dd.Replicator.Delete(cmd.key, cmd.consistency.toClassic))
+ .mapTo[dd.Replicator.DeleteResponse[d]]
+ .map {
+ case rsp: dd.Replicator.DeleteSuccess[d] => JReplicator.DeleteSuccess(rsp.key)
+ case rsp: dd.Replicator.ReplicationDeleteFailure[d] =>
+ JReplicator.DeleteFailure(rsp.key)
+ case rsp: dd.Replicator.DataDeleted[d] => JReplicator.DataDeleted(rsp.key)
+ case rsp: dd.Replicator.StoreFailure[d] => JReplicator.StoreFailure(rsp.key)
+ }
+ .recover {
+ case _ => JReplicator.DeleteFailure(cmd.key)
+ }
reply.foreach { cmd.replyTo ! _ }
Behaviors.same
@@ -205,10 +203,9 @@ import pekko.util.Timeout
case JReplicator.GetReplicaCount(replyTo) =>
implicit val timeout = Timeout(localAskTimeout)
import ctx.executionContext
- val reply =
- (classicReplicator ? dd.Replicator.GetReplicaCount)
- .mapTo[dd.Replicator.ReplicaCount]
- .map(rsp => JReplicator.ReplicaCount(rsp.n))
+ val reply = (classicReplicator ? dd.Replicator.GetReplicaCount)
+ .mapTo[dd.Replicator.ReplicaCount]
+ .map(rsp => JReplicator.ReplicaCount(rsp.n))
reply.foreach { replyTo ! _ }
Behaviors.same
diff --git a/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/ClusterSingleton.scala b/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/ClusterSingleton.scala
index a4bd03fb27..f09e5736e2 100644
--- a/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/ClusterSingleton.scala
+++ b/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/ClusterSingleton.scala
@@ -21,8 +21,7 @@ import pekko.actor.typed._
import pekko.annotation.{ DoNotInherit, InternalApi }
import pekko.cluster.ClusterSettings.DataCenter
import pekko.cluster.singleton.{
- ClusterSingletonManagerSettings => ClassicClusterSingletonManagerSettings,
- ClusterSingletonProxySettings
+ ClusterSingletonManagerSettings => ClassicClusterSingletonManagerSettings, ClusterSingletonProxySettings
}
import pekko.cluster.typed.internal.AdaptedClusterSingletonImpl
import pekko.coordination.lease.LeaseUsageSettings
diff --git a/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/internal/receptionist/ClusterReceptionist.scala b/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/internal/receptionist/ClusterReceptionist.scala
index 102f3abd00..bf4490bae5 100644
--- a/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/internal/receptionist/ClusterReceptionist.scala
+++ b/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/internal/receptionist/ClusterReceptionist.scala
@@ -19,9 +19,7 @@ import org.apache.pekko
import pekko.actor.Address
import pekko.actor.typed.{ ActorRef, Behavior }
import pekko.actor.typed.internal.receptionist.{
- AbstractServiceKey,
- ReceptionistBehaviorProvider,
- ReceptionistMessages
+ AbstractServiceKey, ReceptionistBehaviorProvider, ReceptionistMessages
}
import pekko.actor.typed.receptionist.Receptionist.Command
import pekko.actor.typed.receptionist.ServiceKey
diff --git a/cluster-typed/src/test/scala/org/apache/pekko/cluster/typed/internal/delivery/ReliableDeliverySerializerSpec.scala b/cluster-typed/src/test/scala/org/apache/pekko/cluster/typed/internal/delivery/ReliableDeliverySerializerSpec.scala
index 341f993f68..87d0573f25 100644
--- a/cluster-typed/src/test/scala/org/apache/pekko/cluster/typed/internal/delivery/ReliableDeliverySerializerSpec.scala
+++ b/cluster-typed/src/test/scala/org/apache/pekko/cluster/typed/internal/delivery/ReliableDeliverySerializerSpec.scala
@@ -66,9 +66,11 @@ class ReliableDeliverySerializerSpec extends ScalaTestWithActorTestKit with AnyW
DurableProducerQueue.MessageSent(15L, "msg15", true, "q4", timestamp),
DurableProducerQueue.MessageSent(16L, "msg16", true, "q4", timestamp))),
"DurableProducerQueue.Cleanup" -> DurableProducerQueue.Cleanup(Set("q1", "q2", "q3")),
- "SequencedMessage-chunked-1" -> ConsumerController.SequencedMessage
+ "SequencedMessage-chunked-1" ->
+ ConsumerController.SequencedMessage
.fromChunked("prod-1", 1L, ChunkedMessage(ByteString.fromString("abc"), true, true, 20, ""), true, true, ref),
- "SequencedMessage-chunked-2" -> ConsumerController.SequencedMessage
+ "SequencedMessage-chunked-2" ->
+ ConsumerController.SequencedMessage
.fromChunked("prod-1", 1L, ChunkedMessage(ByteString(1, 2, 3), true, false, 123456, "A"), false, false, ref),
"DurableProducerQueue.MessageSent-chunked" -> DurableProducerQueue.MessageSent.fromChunked(
3L,
diff --git a/cluster/src/main/scala/org/apache/pekko/cluster/ClusterDaemon.scala b/cluster/src/main/scala/org/apache/pekko/cluster/ClusterDaemon.scala
index da0a73d8bd..1b085cafa6 100644
--- a/cluster/src/main/scala/org/apache/pekko/cluster/ClusterDaemon.scala
+++ b/cluster/src/main/scala/org/apache/pekko/cluster/ClusterDaemon.scala
@@ -911,7 +911,8 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef, joinConfigCompatCh
def leaving(address: Address): Unit = {
// only try to update if the node is available (in the member ring)
latestGossip.members.find(_.address == address).foreach { existingMember =>
- if (existingMember.status == Joining || existingMember.status == WeaklyUp || existingMember
+ if (existingMember.status == Joining || existingMember.status == WeaklyUp ||
+ existingMember
.status == Up || existingMember.status == PreparingForShutdown || existingMember.status == ReadyForShutdown) {
// mark node as LEAVING
val newMembers = latestGossip.members - existingMember + existingMember.copy(status = Leaving)
@@ -1290,7 +1291,8 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef, joinConfigCompatCh
} else {
leaderActionCounter += 1
import cluster.settings.{ AllowWeaklyUpMembers, LeaderActionsInterval, WeaklyUpAfter }
- if (AllowWeaklyUpMembers && LeaderActionsInterval * leaderActionCounter >= WeaklyUpAfter && !preparingForShutdown)
+ if (AllowWeaklyUpMembers && LeaderActionsInterval * leaderActionCounter >= WeaklyUpAfter &&
+ !preparingForShutdown)
moveJoiningToWeaklyUp()
if (leaderActionCounter == firstNotice || leaderActionCounter % periodicNotice == 0)
@@ -1316,7 +1318,8 @@ private[cluster] class ClusterCoreDaemon(publisher: ActorRef, joinConfigCompatCh
def checkForPrepareForShutdown(): Unit = {
if (MembershipState.allowedToPrepareToShutdown(
- latestGossip.member(selfUniqueAddress).status) && latestGossip.members
+ latestGossip.member(selfUniqueAddress).status) &&
+ latestGossip.members
.exists(m => MembershipState.prepareForShutdownStates(m.status))) {
logDebug("Detected full cluster shutdown")
self ! ClusterUserAction.PrepareForShutdown
diff --git a/cluster/src/main/scala/org/apache/pekko/cluster/ClusterEvent.scala b/cluster/src/main/scala/org/apache/pekko/cluster/ClusterEvent.scala
index 5387845abb..c8daba849f 100644
--- a/cluster/src/main/scala/org/apache/pekko/cluster/ClusterEvent.scala
+++ b/cluster/src/main/scala/org/apache/pekko/cluster/ClusterEvent.scala
@@ -227,7 +227,8 @@ object ClusterEvent {
override def equals(other: Any): Boolean = other match {
case that: CurrentClusterState =>
- (this eq that) || (members == that.members &&
+ (this eq that) ||
+ (members == that.members &&
unreachable == that.unreachable &&
seenBy == that.seenBy &&
leader == that.leader &&
diff --git a/cluster/src/main/scala/org/apache/pekko/cluster/MembershipState.scala b/cluster/src/main/scala/org/apache/pekko/cluster/MembershipState.scala
index 136826cfb6..2a56bbeb3d 100644
--- a/cluster/src/main/scala/org/apache/pekko/cluster/MembershipState.scala
+++ b/cluster/src/main/scala/org/apache/pekko/cluster/MembershipState.scala
@@ -306,7 +306,8 @@ import pekko.cluster.MemberStatus._
latestGossip.members.iterator
.collect {
case m
- if m.dataCenter == state.selfDc && !latestGossip.seenByNode(m.uniqueAddress) && state
+ if m.dataCenter == state.selfDc && !latestGossip.seenByNode(m.uniqueAddress) &&
+ state
.validNodeForGossip(m.uniqueAddress) =>
m.uniqueAddress
}
diff --git a/cluster/src/main/scala/org/apache/pekko/cluster/routing/ClusterRouterConfig.scala b/cluster/src/main/scala/org/apache/pekko/cluster/routing/ClusterRouterConfig.scala
index 3a32055a05..f499c8801a 100644
--- a/cluster/src/main/scala/org/apache/pekko/cluster/routing/ClusterRouterConfig.scala
+++ b/cluster/src/main/scala/org/apache/pekko/cluster/routing/ClusterRouterConfig.scala
@@ -547,8 +547,7 @@ private[pekko] trait ClusterRouterActor { this: RouterActor =>
}
}
- def isAvailable(m: Member): Boolean =
- (m.status == MemberStatus.Up || m.status == MemberStatus.WeaklyUp) &&
+ def isAvailable(m: Member): Boolean = (m.status == MemberStatus.Up || m.status == MemberStatus.WeaklyUp) &&
satisfiesRoles(m.roles) &&
(settings.allowLocalRoutees || m.address != cluster.selfAddress)
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/ClusterAccrualFailureDetectorSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/ClusterAccrualFailureDetectorSpec.scala
index 695ee51342..a3fc9d844a 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/ClusterAccrualFailureDetectorSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/ClusterAccrualFailureDetectorSpec.scala
@@ -100,7 +100,8 @@ abstract class ClusterAccrualFailureDetectorSpec
enterBarrier("after-2")
}
- "mark node as 'unavailable' if a node in the cluster is shut down (and its heartbeats stops)" taggedAs LongRunningTest in {
+ "mark node as 'unavailable' if a node in the cluster is shut down (and its heartbeats stops)" taggedAs
+ LongRunningTest in {
runOn(first) {
testConductor.exit(third, 0).await
}
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/ClusterDeathWatchSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/ClusterDeathWatchSpec.scala
index c69e026b6a..f02f7e7cb9 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/ClusterDeathWatchSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/ClusterDeathWatchSpec.scala
@@ -162,7 +162,8 @@ abstract class ClusterDeathWatchSpec
}
- "not be able to watch an actor before node joins cluster, ClusterRemoteWatcher takes over from RemoteWatcher" in within(
+ "not be able to watch an actor before node joins cluster, ClusterRemoteWatcher takes over from RemoteWatcher" in
+ within(
20.seconds) {
runOn(fifth) {
system.actorOf(
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/JoinInProgressSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/JoinInProgressSpec.scala
index f3fe52a6b3..3b84163653 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/JoinInProgressSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/JoinInProgressSpec.scala
@@ -45,7 +45,8 @@ abstract class JoinInProgressSpec extends MultiNodeClusterSpec(JoinInProgressMul
import JoinInProgressMultiJvmSpec._
"A cluster node" must {
- "send heartbeats immediately when joining to avoid false failure detection due to delayed gossip" taggedAs LongRunningTest in {
+ "send heartbeats immediately when joining to avoid false failure detection due to delayed gossip" taggedAs
+ LongRunningTest in {
runOn(first) {
startClusterNode()
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/LeaderLeavingSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/LeaderLeavingSpec.scala
index 73cccc0afb..836657b451 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/LeaderLeavingSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/LeaderLeavingSpec.scala
@@ -49,7 +49,8 @@ abstract class LeaderLeavingSpec extends MultiNodeClusterSpec(LeaderLeavingMulti
"A LEADER that is LEAVING" must {
- "be moved to LEAVING, then to EXITING, then to REMOVED, then be shut down and then a new LEADER should be elected" taggedAs LongRunningTest in {
+ "be moved to LEAVING, then to EXITING, then to REMOVED, then be shut down and then a new LEADER should be elected" taggedAs
+ LongRunningTest in {
awaitClusterUp(first, second, third)
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MembershipChangeListenerUpSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MembershipChangeListenerUpSpec.scala
index 49d55cd6f0..11d4e033c4 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MembershipChangeListenerUpSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MembershipChangeListenerUpSpec.scala
@@ -39,7 +39,8 @@ abstract class MembershipChangeListenerUpSpec extends MultiNodeClusterSpec(Membe
"A set of connected cluster systems" must {
- "(when two nodes) after cluster convergence updates the membership table then all MembershipChangeListeners should be triggered" taggedAs LongRunningTest in {
+ "(when two nodes) after cluster convergence updates the membership table then all MembershipChangeListeners should be triggered" taggedAs
+ LongRunningTest in {
awaitClusterUp(first)
@@ -70,7 +71,8 @@ abstract class MembershipChangeListenerUpSpec extends MultiNodeClusterSpec(Membe
enterBarrier("after-1")
}
- "(when three nodes) after cluster convergence updates the membership table then all MembershipChangeListeners should be triggered" taggedAs LongRunningTest in {
+ "(when three nodes) after cluster convergence updates the membership table then all MembershipChangeListeners should be triggered" taggedAs
+ LongRunningTest in {
val latch = TestLatch()
val expectedAddresses = Set(first, second, third).map(address)
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MinMembersBeforeUpSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MinMembersBeforeUpSpec.scala
index 9a57467f42..160ee6eb0c 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MinMembersBeforeUpSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MinMembersBeforeUpSpec.scala
@@ -92,7 +92,8 @@ abstract class MinMembersBeforeUpWithWeaklyUpSpec extends MinMembersBeforeUpBase
override def third: RoleName = MinMembersBeforeUpWithWeaklyUpMultiJvmSpec.third
"Cluster leader" must {
- "wait with moving members to UP until minimum number of members have joined with weakly up enabled" taggedAs LongRunningTest in {
+ "wait with moving members to UP until minimum number of members have joined with weakly up enabled" taggedAs
+ LongRunningTest in {
testWaitMovingMembersToUp()
}
}
@@ -105,7 +106,8 @@ abstract class MinMembersOfRoleBeforeUpSpec extends MinMembersBeforeUpBase(MinMe
override def third: RoleName = MinMembersOfRoleBeforeUpMultiJvmSpec.third
"Cluster leader" must {
- "wait with moving members to UP until minimum number of members with specific role have joined" taggedAs LongRunningTest in {
+ "wait with moving members to UP until minimum number of members with specific role have joined" taggedAs
+ LongRunningTest in {
testWaitMovingMembersToUp()
}
}
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MultiDcSunnyWeatherSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MultiDcSunnyWeatherSpec.scala
index 223bc66a7a..dece03ecdd 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MultiDcSunnyWeatherSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/MultiDcSunnyWeatherSpec.scala
@@ -91,8 +91,9 @@ abstract class MultiDcSunnyWeatherSpec extends MultiNodeClusterSpec(MultiDcSunny
val expectedBetaHeartbeaterNodes = takeNOldestMembers(dataCenter = "beta", 2)
val expectedBetaHeartbeaterRoles = membersAsRoles(expectedBetaHeartbeaterNodes)
- val expectedNoActiveHeartbeatSenderRoles = roles.toSet -- (expectedAlphaHeartbeaterRoles.union(
- expectedBetaHeartbeaterRoles))
+ val expectedNoActiveHeartbeatSenderRoles = roles.toSet --
+ (expectedAlphaHeartbeaterRoles.union(
+ expectedBetaHeartbeaterRoles))
enterBarrier("found-expectations")
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/NodeMembershipSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/NodeMembershipSpec.scala
index d613f1a5ae..fbca908773 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/NodeMembershipSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/NodeMembershipSpec.scala
@@ -35,7 +35,8 @@ abstract class NodeMembershipSpec extends MultiNodeClusterSpec(NodeMembershipMul
"A set of connected cluster systems" must {
- "(when two nodes) start gossiping to each other so that both nodes gets the same gossip info" taggedAs LongRunningTest in {
+ "(when two nodes) start gossiping to each other so that both nodes gets the same gossip info" taggedAs
+ LongRunningTest in {
// make sure that the node-to-join is started before other join
runOn(first) {
@@ -53,7 +54,8 @@ abstract class NodeMembershipSpec extends MultiNodeClusterSpec(NodeMembershipMul
enterBarrier("after-1")
}
- "(when three nodes) start gossiping to each other so that all nodes gets the same gossip info" taggedAs LongRunningTest in {
+ "(when three nodes) start gossiping to each other so that all nodes gets the same gossip info" taggedAs
+ LongRunningTest in {
runOn(third) {
cluster.join(first)
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/SplitBrainSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/SplitBrainSpec.scala
index 8d2ea41cd9..23daa5fe81 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/SplitBrainSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/SplitBrainSpec.scala
@@ -75,7 +75,8 @@ abstract class SplitBrainSpec(multiNodeConfig: SplitBrainMultiNodeConfig)
enterBarrier("after-1")
}
- "detect network partition and mark nodes on other side as unreachable and form new cluster" taggedAs LongRunningTest in within(
+ "detect network partition and mark nodes on other side as unreachable and form new cluster" taggedAs
+ LongRunningTest in within(
30.seconds) {
enterBarrier("before-split")
diff --git a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/StressSpec.scala b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/StressSpec.scala
index 3dd7555e8e..85c787cfa3 100644
--- a/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/StressSpec.scala
+++ b/cluster/src/multi-jvm/scala/org/apache/pekko/cluster/StressSpec.scala
@@ -145,10 +145,11 @@ private[cluster] object StressMultiJvmSpec extends MultiNodeConfig {
val numberOfNodesJoiningOneByOneLarge = getInt("nr-of-nodes-joining-one-by-one-large") * nFactor
val numberOfNodesJoiningToOneNode = getInt("nr-of-nodes-joining-to-one") * nFactor
// remaining will join to seed nodes
- val numberOfNodesJoiningToSeedNodes = (totalNumberOfNodes - numberOfSeedNodes -
+ val numberOfNodesJoiningToSeedNodes =
+ (totalNumberOfNodes - numberOfSeedNodes -
numberOfNodesJoiningToSeedNodesInitially - numberOfNodesJoiningOneByOneSmall -
numberOfNodesJoiningOneByOneLarge - numberOfNodesJoiningToOneNode)
- .requiring(_ >= 0, s"too many configured nr-of-nodes-joining-*, total should be <= $totalNumberOfNodes")
+ .requiring(_ >= 0, s"too many configured nr-of-nodes-joining-*, total should be <= $totalNumberOfNodes")
val numberOfNodesLeavingOneByOneSmall = getInt("nr-of-nodes-leaving-one-by-one-small") * nFactor
val numberOfNodesLeavingOneByOneLarge = getInt("nr-of-nodes-leaving-one-by-one-large") * nFactor
val numberOfNodesLeaving = getInt("nr-of-nodes-leaving") * nFactor
@@ -166,13 +167,15 @@ private[cluster] object StressMultiJvmSpec extends MultiNodeConfig {
require(
numberOfSeedNodes + numberOfNodesJoiningToSeedNodesInitially + numberOfNodesJoiningOneByOneSmall +
- numberOfNodesJoiningOneByOneLarge + numberOfNodesJoiningToOneNode + numberOfNodesJoiningToSeedNodes <= totalNumberOfNodes,
+ numberOfNodesJoiningOneByOneLarge + numberOfNodesJoiningToOneNode + numberOfNodesJoiningToSeedNodes <=
+ totalNumberOfNodes,
s"specified number of joining nodes <= $totalNumberOfNodes")
// don't shutdown the 3 nodes hosting the master actors
require(
numberOfNodesLeavingOneByOneSmall + numberOfNodesLeavingOneByOneLarge + numberOfNodesLeaving +
- numberOfNodesShutdownOneByOneSmall + numberOfNodesShutdownOneByOneLarge + numberOfNodesShutdown <= totalNumberOfNodes - 3,
+ numberOfNodesShutdownOneByOneSmall + numberOfNodesShutdownOneByOneLarge + numberOfNodesShutdown <=
+ totalNumberOfNodes - 3,
s"""specified number of leaving/shutdown nodes <= ${totalNumberOfNodes - 3}""")
require(
@@ -287,8 +290,7 @@ private[cluster] object StressMultiJvmSpec extends MultiNodeConfig {
log.info("Cluster result history\n" + formatHistory)
}
- def formatHistory: String =
- (formatHistoryHeader +: (history.map(formatHistoryLine))).mkString("\n")
+ def formatHistory: String = (formatHistoryHeader +: (history.map(formatHistoryLine))).mkString("\n")
def formatHistoryHeader: String = "[Title]\t[Duration (ms)]\t[GossipStats(gossip, merge, same, newer, older)]"
diff --git a/cluster/src/test/scala/org/apache/pekko/cluster/ClusterHeartbeatSenderStateSpec.scala b/cluster/src/test/scala/org/apache/pekko/cluster/ClusterHeartbeatSenderStateSpec.scala
index d949181f79..91b0fc1b7a 100644
--- a/cluster/src/test/scala/org/apache/pekko/cluster/ClusterHeartbeatSenderStateSpec.scala
+++ b/cluster/src/test/scala/org/apache/pekko/cluster/ClusterHeartbeatSenderStateSpec.scala
@@ -163,8 +163,8 @@ class ClusterHeartbeatSenderStateSpec extends AnyWordSpec with Matchers {
"behave correctly for random operations" in {
val rnd = ThreadLocalRandom.current
- val nodes =
- (1 to rnd.nextInt(10, 200)).map(n => UniqueAddress(Address("pekko", "sys", "n" + n, 7355), n.toLong)).toVector
+ val nodes = (1 to rnd.nextInt(10, 200)).map(n =>
+ UniqueAddress(Address("pekko", "sys", "n" + n, 7355), n.toLong)).toVector
def rndNode() = nodes(rnd.nextInt(0, nodes.size))
val selfUniqueAddress = rndNode()
var state = emptyState(selfUniqueAddress)
diff --git a/cluster/src/test/scala/org/apache/pekko/cluster/JoinConfigCompatCheckerSpec.scala b/cluster/src/test/scala/org/apache/pekko/cluster/JoinConfigCompatCheckerSpec.scala
index d141074590..086a42f1e7 100644
--- a/cluster/src/test/scala/org/apache/pekko/cluster/JoinConfigCompatCheckerSpec.scala
+++ b/cluster/src/test/scala/org/apache/pekko/cluster/JoinConfigCompatCheckerSpec.scala
@@ -112,7 +112,8 @@ class JoinConfigCompatCheckerSpec extends PekkoSpec with ClusterTestKit {
}
- "NOT be allowed to join a cluster when one of its required properties are not available on cluster side" taggedAs LongRunningTest in {
+ "NOT be allowed to join a cluster when one of its required properties are not available on cluster side" taggedAs
+ LongRunningTest in {
// this config is NOT compatible with the cluster config
// because there is one missing required configuration property.
@@ -153,7 +154,8 @@ class JoinConfigCompatCheckerSpec extends PekkoSpec with ClusterTestKit {
}
- "NOT be allowed to join a cluster when one of the cluster required properties are not available on the joining side" taggedAs LongRunningTest in {
+ "NOT be allowed to join a cluster when one of the cluster required properties are not available on the joining side" taggedAs
+ LongRunningTest in {
// this config is NOT compatible with the cluster config
// because there is one missing required configuration property.
@@ -190,7 +192,8 @@ class JoinConfigCompatCheckerSpec extends PekkoSpec with ClusterTestKit {
}
- "be allowed to join a cluster when one of its required properties are not available on cluster side but it's configured to NOT enforce it" taggedAs LongRunningTest in {
+ "be allowed to join a cluster when one of its required properties are not available on cluster side but it's configured to NOT enforce it" taggedAs
+ LongRunningTest in {
// this config is NOT compatible with the cluster config
// because there is one missing required configuration property.
@@ -228,7 +231,8 @@ class JoinConfigCompatCheckerSpec extends PekkoSpec with ClusterTestKit {
}
- "be allowed to join a cluster when its configuration is incompatible but it's configured to NOT enforce it" taggedAs LongRunningTest in {
+ "be allowed to join a cluster when its configuration is incompatible but it's configured to NOT enforce it" taggedAs
+ LongRunningTest in {
// this config is NOT compatible with the cluster config,
// but node will ignore the the config check and join anyway
val joinNodeConfig =
@@ -263,7 +267,8 @@ class JoinConfigCompatCheckerSpec extends PekkoSpec with ClusterTestKit {
}
/** This test verifies the built-in JoinConfigCompatCheckerPekkoCluster */
- "NOT be allowed to join a cluster using a different value for pekko.cluster.downing-provider-class" taggedAs LongRunningTest in {
+ "NOT be allowed to join a cluster using a different value for pekko.cluster.downing-provider-class" taggedAs
+ LongRunningTest in {
val joinNodeConfig =
ConfigFactory.parseString("""
@@ -372,7 +377,8 @@ class JoinConfigCompatCheckerSpec extends PekkoSpec with ClusterTestKit {
}
- "NOT be allowed to re-join a cluster when one of its required properties are not available on cluster side" taggedAs LongRunningTest in {
+ "NOT be allowed to re-join a cluster when one of its required properties are not available on cluster side" taggedAs
+ LongRunningTest in {
// this config is NOT compatible with the cluster config
// because there is one missing required configuration property.
@@ -420,7 +426,8 @@ class JoinConfigCompatCheckerSpec extends PekkoSpec with ClusterTestKit {
}
- "NOT be allowed to re-join a cluster when one of the cluster required properties are not available on the joining side" taggedAs LongRunningTest in {
+ "NOT be allowed to re-join a cluster when one of the cluster required properties are not available on the joining side" taggedAs
+ LongRunningTest in {
// this config is NOT compatible with the cluster config
// because there is one missing required configuration property.
@@ -464,7 +471,8 @@ class JoinConfigCompatCheckerSpec extends PekkoSpec with ClusterTestKit {
}
- "be allowed to re-join a cluster when one of its required properties are not available on cluster side but it's configured to NOT enforce it" taggedAs LongRunningTest in {
+ "be allowed to re-join a cluster when one of its required properties are not available on cluster side but it's configured to NOT enforce it" taggedAs
+ LongRunningTest in {
// this config is NOT compatible with the cluster config
// because there is one missing required configuration property.
@@ -512,7 +520,8 @@ class JoinConfigCompatCheckerSpec extends PekkoSpec with ClusterTestKit {
}
- "be allowed to re-join a cluster when its configuration is incompatible but it's configured to NOT enforce it" taggedAs LongRunningTest in {
+ "be allowed to re-join a cluster when its configuration is incompatible but it's configured to NOT enforce it" taggedAs
+ LongRunningTest in {
// this config is NOT compatible with the cluster config,
// but node will ignore the the config check and join anyway
val joinNodeConfig =
diff --git a/cluster/src/test/scala/org/apache/pekko/cluster/ReachabilityPerfSpec.scala b/cluster/src/test/scala/org/apache/pekko/cluster/ReachabilityPerfSpec.scala
index 85d6b1d37a..471fb7f8a5 100644
--- a/cluster/src/test/scala/org/apache/pekko/cluster/ReachabilityPerfSpec.scala
+++ b/cluster/src/test/scala/org/apache/pekko/cluster/ReachabilityPerfSpec.scala
@@ -29,14 +29,13 @@ class ReachabilityPerfSpec extends AnyWordSpec with Matchers {
val address = Address("pekko", "sys", "a", 7355)
val node = Address("pekko", "sys", "a", 7355)
- private def createReachabilityOfSize(base: Reachability, size: Int): Reachability =
- (1 to size).foldLeft(base) {
- case (r, i) =>
- val observer = UniqueAddress(address.copy(host = Some("node-" + i)), i.toLong)
- val j = if (i == size) 1 else i + 1
- val subject = UniqueAddress(address.copy(host = Some("node-" + j)), j.toLong)
- r.unreachable(observer, subject).reachable(observer, subject)
- }
+ private def createReachabilityOfSize(base: Reachability, size: Int): Reachability = (1 to size).foldLeft(base) {
+ case (r, i) =>
+ val observer = UniqueAddress(address.copy(host = Some("node-" + i)), i.toLong)
+ val j = if (i == size) 1 else i + 1
+ val subject = UniqueAddress(address.copy(host = Some("node-" + j)), j.toLong)
+ r.unreachable(observer, subject).reachable(observer, subject)
+ }
@nowarn
private def addUnreachable(base: Reachability, count: Int): Reachability = {
diff --git a/cluster/src/test/scala/org/apache/pekko/cluster/StartupWithOneThreadSpec.scala b/cluster/src/test/scala/org/apache/pekko/cluster/StartupWithOneThreadSpec.scala
index 398f0c6d6c..752ae0aad4 100644
--- a/cluster/src/test/scala/org/apache/pekko/cluster/StartupWithOneThreadSpec.scala
+++ b/cluster/src/test/scala/org/apache/pekko/cluster/StartupWithOneThreadSpec.scala
@@ -66,14 +66,16 @@ class StartupWithOneThreadSpec(startTime: Long) extends PekkoSpec(StartupWithOne
// dispatcher threads.
// Note that the Cluster extension is started via ClusterActorRefProvider
// before ActorSystem.apply returns, i.e. in the constructor of PekkoSpec.
- (System.nanoTime - startTime).nanos.toMillis should be <
+ (System.nanoTime - startTime).nanos.toMillis should
+ be <
(system.settings.CreationTimeout.duration - 2.second).toMillis
system.actorOf(testProps) ! "hello"
system.actorOf(testProps) ! "hello"
system.actorOf(testProps) ! "hello"
Cluster(system)
- (System.nanoTime - startTime).nanos.toMillis should be <
+ (System.nanoTime - startTime).nanos.toMillis should
+ be <
(system.settings.CreationTimeout.duration - 2.second).toMillis
expectMsg("hello")
diff --git a/cluster/src/test/scala/org/apache/pekko/cluster/VectorClockPerfSpec.scala b/cluster/src/test/scala/org/apache/pekko/cluster/VectorClockPerfSpec.scala
index 350aacb33f..e160056610 100644
--- a/cluster/src/test/scala/org/apache/pekko/cluster/VectorClockPerfSpec.scala
+++ b/cluster/src/test/scala/org/apache/pekko/cluster/VectorClockPerfSpec.scala
@@ -21,12 +21,12 @@ import org.scalatest.wordspec.AnyWordSpec
object VectorClockPerfSpec {
import VectorClock._
- def createVectorClockOfSize(size: Int): (VectorClock, SortedSet[Node]) =
- (1 to size).foldLeft((VectorClock(), SortedSet.empty[Node])) {
- case ((vc, nodes), i) =>
- val node = Node(i.toString)
- (vc :+ node, nodes + node)
- }
+ def createVectorClockOfSize(
+ size: Int): (VectorClock, SortedSet[Node]) = (1 to size).foldLeft((VectorClock(), SortedSet.empty[Node])) {
+ case ((vc, nodes), i) =>
+ val node = Node(i.toString)
+ (vc :+ node, nodes + node)
+ }
def copyVectorClock(vc: VectorClock): VectorClock = {
val versions = vc.versions.foldLeft(TreeMap.empty[Node, Long]) {
diff --git a/cluster/src/test/scala/org/apache/pekko/cluster/sbr/SplitBrainResolverSpec.scala b/cluster/src/test/scala/org/apache/pekko/cluster/sbr/SplitBrainResolverSpec.scala
index 360bfb3028..605a63c0ac 100644
--- a/cluster/src/test/scala/org/apache/pekko/cluster/sbr/SplitBrainResolverSpec.scala
+++ b/cluster/src/test/scala/org/apache/pekko/cluster/sbr/SplitBrainResolverSpec.scala
@@ -515,7 +515,8 @@ class SplitBrainResolverSpec
assertDowningSide(side2, Set(memberB, memberC, memberD, memberE))
}
- "down indirectly connected on minority side, when combined with clean partition: {A, (B, C)} | {D, E, F, G} => {D, E, F, G}" in new Setup2(
+ "down indirectly connected on minority side, when combined with clean partition: {A, (B, C)} | {D, E, F, G} => {D, E, F, G}" in
+ new Setup2(
role = None) {
side1 = Set(memberA, memberB, memberC)
side2 = Set(memberD, memberE, memberF, memberG)
@@ -535,7 +536,8 @@ class SplitBrainResolverSpec
assertDowningSide(side2, Set(memberA, memberB, memberC))
}
- "down indirectly connected on majority side, when combined with clean partition: {A, B, C} | {(D, E), F, G} => {F, G}" in new Setup2(
+ "down indirectly connected on majority side, when combined with clean partition: {A, B, C} | {(D, E), F, G} => {F, G}" in
+ new Setup2(
role = None) {
side1 = Set(memberA, memberB, memberC)
side2 = Set(memberD, memberE, memberF, memberG)
@@ -569,7 +571,8 @@ class SplitBrainResolverSpec
assertDowningSide(side2, Set(memberA, memberB, memberC, memberE, memberF))
}
- "down indirectly connected, detected via seen, combined with clean partition: {A, B, C} | {(D, E), (F, G)} => {}" in new Setup2(
+ "down indirectly connected, detected via seen, combined with clean partition: {A, B, C} | {(D, E), (F, G)} => {}" in
+ new Setup2(
role = None) {
side1 = Set(memberA, memberB, memberC)
side2 = Set(memberD, memberE, memberF, memberG)
@@ -582,7 +585,8 @@ class SplitBrainResolverSpec
assertDowningSide(side2, Set(memberA, memberB, memberC, memberD, memberE, memberF, memberG))
}
- "double DownIndirectlyConnected when indirectly connected happens before clean partition: {A, B, C} | {(D, E), (F, G)} => {}" in new Setup2(
+ "double DownIndirectlyConnected when indirectly connected happens before clean partition: {A, B, C} | {(D, E), (F, G)} => {}" in
+ new Setup2(
role = None) {
side1 = Set(memberA, memberB, memberC)
side2 = Set(memberD, memberE, memberF, memberG)
@@ -775,7 +779,8 @@ class SplitBrainResolverSpec
assertDowning(Set(memberA, memberB))
}
- "down indirectly connected on younger side, when combined with clean partition: {A, (B, C)} | {D, E, F, G} => {D, E, F, G}" in new Setup2 {
+ "down indirectly connected on younger side, when combined with clean partition: {A, (B, C)} | {D, E, F, G} => {D, E, F, G}" in
+ new Setup2 {
side1 = Set(memberA, memberB, memberC)
side2 = Set(memberD, memberE, memberF, memberG)
indirectlyConnected = List(memberB -> memberC, memberC -> memberB)
@@ -794,7 +799,8 @@ class SplitBrainResolverSpec
assertDowningSide(side2, Set(memberA, memberB, memberC))
}
- "down indirectly connected on oldest side, when combined with clean partition: {A, B, C} | {(D, E), F, G} => {F, G}" in new Setup2 {
+ "down indirectly connected on oldest side, when combined with clean partition: {A, B, C} | {(D, E), F, G} => {F, G}" in
+ new Setup2 {
side1 = Set(memberA, memberB, memberC)
side2 = Set(memberD, memberE, memberF, memberG)
@@ -951,7 +957,8 @@ class SplitBrainResolverSpec
strategy3.nodesToDown(reverseDecision3) should ===(Set(memberB, memberC, memberD, memberE).map(_.uniqueAddress))
}
- "down indirectly connected to already downed node during partition: {A, B, C, D} | {(E, F)} => {A, B, C, D}" in new Setup2(
+ "down indirectly connected to already downed node during partition: {A, B, C, D} | {(E, F)} => {A, B, C, D}" in
+ new Setup2(
role = None) {
val memberELeaving = leaving(memberE)
val memberFDown = downed(memberF)
diff --git a/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/ORMap.scala b/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/ORMap.scala
index ec5fbf8819..3e3cffadd8 100644
--- a/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/ORMap.scala
+++ b/distributed-data/src/main/scala/org/apache/pekko/cluster/ddata/ORMap.scala
@@ -418,9 +418,10 @@ final class ORMap[A, B <: ReplicatedData] private[pekko] (
case putOp: PutDeltaOp[_, _] =>
val keyDelta = putOp.underlying
mergedKeys = mergedKeys.mergeDelta(keyDelta)
- mergedValues = mergedValues + putOp
- .asInstanceOf[PutDeltaOp[A, B]]
- .value // put is destructive and propagates only full values of B!
+ mergedValues = mergedValues +
+ putOp
+ .asInstanceOf[PutDeltaOp[A, B]]
+ .value // put is destructive and propagates only full values of B!
case removeOp: RemoveDeltaOp[_, _] =>
val removedKey = removeOp.underlying match {
// if op is RemoveDeltaOp then it must have exactly one element in the elements
@@ -434,9 +435,10 @@ final class ORMap[A, B <: ReplicatedData] private[pekko] (
case removeKeyOp: RemoveKeyDeltaOp[_, _] =>
// removeKeyOp tombstones values for later use
if (mergedValues.contains(removeKeyOp.asInstanceOf[RemoveKeyDeltaOp[A, B]].removedKey)) {
- tombstonedVals = tombstonedVals + (removeKeyOp
- .asInstanceOf[RemoveKeyDeltaOp[A, B]]
- .removedKey -> mergedValues(removeKeyOp.asInstanceOf[RemoveKeyDeltaOp[A, B]].removedKey))
+ tombstonedVals = tombstonedVals +
+ (removeKeyOp
+ .asInstanceOf[RemoveKeyDeltaOp[A, B]]
+ .removedKey -> mergedValues(removeKeyOp.asInstanceOf[RemoveKeyDeltaOp[A, B]].removedKey))
}
mergedValues = mergedValues - removeKeyOp.asInstanceOf[RemoveKeyDeltaOp[A, B]].removedKey
mergedKeys = mergedKeys.mergeDelta(removeKeyOp.underlying)
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 dfc50fd04f..b4b69cab6b 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
@@ -1839,7 +1839,8 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
}
def isNodeRemoved(node: UniqueAddress, keys: Iterable[KeyId]): Boolean = {
- removedNodes.contains(node) || (keys.exists(key =>
+ removedNodes.contains(node) ||
+ (keys.exists(key =>
dataEntries.get(key) match {
case Some((DataEnvelope(_, pruning, _), _)) => pruning.contains(node)
case None => false
diff --git a/docs/src/test/scala/docs/actor/io/dns/DnsCompileOnlyDocSpec.scala b/docs/src/test/scala/docs/actor/io/dns/DnsCompileOnlyDocSpec.scala
index c418091058..56e14ca2e8 100644
--- a/docs/src/test/scala/docs/actor/io/dns/DnsCompileOnlyDocSpec.scala
+++ b/docs/src/test/scala/docs/actor/io/dns/DnsCompileOnlyDocSpec.scala
@@ -41,22 +41,22 @@ object DnsCompileOnlyDocSpec {
{
// #actor-api-inet-address
- val resolved: Future[DnsProtocol.Resolved] =
- (IO(Dns) ? DnsProtocol.Resolve("google.com")).mapTo[DnsProtocol.Resolved]
+ val resolved
+ : Future[DnsProtocol.Resolved] = (IO(Dns) ? DnsProtocol.Resolve("google.com")).mapTo[DnsProtocol.Resolved]
// #actor-api-inet-address
}
{
// #actor-api-async
- val resolved: Future[DnsProtocol.Resolved] =
- (IO(Dns) ? DnsProtocol.Resolve("google.com")).mapTo[DnsProtocol.Resolved]
+ val resolved
+ : Future[DnsProtocol.Resolved] = (IO(Dns) ? DnsProtocol.Resolve("google.com")).mapTo[DnsProtocol.Resolved]
// #actor-api-async
}
{
// #srv
- val resolved: Future[DnsProtocol.Resolved] =
- (IO(Dns) ? DnsProtocol.Resolve("your-service", Srv)).mapTo[DnsProtocol.Resolved]
+ val resolved: Future[DnsProtocol.Resolved] = (IO(Dns) ? DnsProtocol.Resolve("your-service", Srv)).mapTo[
+ DnsProtocol.Resolved]
// #srv
}
diff --git a/docs/src/test/scala/typed/tutorial_5/DeviceGroup.scala b/docs/src/test/scala/typed/tutorial_5/DeviceGroup.scala
index 9e2da39da6..0b7d8aa230 100644
--- a/docs/src/test/scala/typed/tutorial_5/DeviceGroup.scala
+++ b/docs/src/test/scala/typed/tutorial_5/DeviceGroup.scala
@@ -40,11 +40,7 @@ class DeviceGroup(context: ActorContext[DeviceGroup.Command], groupId: String)
extends AbstractBehavior[DeviceGroup.Command](context) {
import DeviceGroup._
import DeviceManager.{
- DeviceRegistered,
- ReplyDeviceList,
- RequestAllTemperatures,
- RequestDeviceList,
- RequestTrackDevice
+ DeviceRegistered, ReplyDeviceList, RequestAllTemperatures, RequestDeviceList, RequestTrackDevice
}
private var deviceIdToActor = Map.empty[String, ActorRef[Device.Command]]
diff --git a/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/Conductor.scala b/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/Conductor.scala
index 303e09b8c9..36e5231bfd 100644
--- a/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/Conductor.scala
+++ b/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/Conductor.scala
@@ -29,17 +29,8 @@ import org.apache.pekko
import pekko.ConfigurationException
import pekko.PekkoException
import pekko.actor.{
- Actor,
- ActorRef,
- Address,
- DeadLetterSuppression,
- Deploy,
- LoggingFSM,
- NoSerializationVerificationNeeded,
- OneForOneStrategy,
- Props,
- Status,
- SupervisorStrategy
+ Actor, ActorRef, Address, DeadLetterSuppression, Deploy, LoggingFSM, NoSerializationVerificationNeeded,
+ OneForOneStrategy, Props, Status, SupervisorStrategy
}
import pekko.event.{ Logging, LoggingAdapter }
import pekko.event.LoggingReceive
diff --git a/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala b/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala
index ae9504934b..026d0dfe01 100644
--- a/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala
+++ b/multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala
@@ -30,10 +30,7 @@ import io.netty.channel.nio.NioIoHandler
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.{ NioServerSocketChannel, NioSocketChannel }
import io.netty.handler.codec.{
- LengthFieldBasedFrameDecoder,
- LengthFieldPrepender,
- MessageToMessageDecoder,
- MessageToMessageEncoder
+ LengthFieldBasedFrameDecoder, LengthFieldPrepender, MessageToMessageDecoder, MessageToMessageEncoder
}
/**
diff --git a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/journal/leveldb/EventsByPersistenceIdStage.scala b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/journal/leveldb/EventsByPersistenceIdStage.scala
index cfd30c388d..e351830a1d 100644
--- a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/journal/leveldb/EventsByPersistenceIdStage.scala
+++ b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/journal/leveldb/EventsByPersistenceIdStage.scala
@@ -132,7 +132,8 @@ final private[pekko] class EventsByPersistenceIdStage(
nextSequenceNr,
toSequenceNr,
bufferSize)
- if (bufferEmpty && (nextSequenceNr > toSequenceNr || (nextSequenceNr == fromSequenceNr && isCurrentQuery()))) {
+ if (bufferEmpty &&
+ (nextSequenceNr > toSequenceNr || (nextSequenceNr == fromSequenceNr && isCurrentQuery()))) {
completeStage()
} else if (nextSequenceNr < toSequenceNr) {
// need further requests to the journal
diff --git a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/journal/leveldb/scaladsl/LeveldbReadJournal.scala b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/journal/leveldb/scaladsl/LeveldbReadJournal.scala
index 496f90cd02..75cad65a4f 100644
--- a/persistence-query/src/main/scala/org/apache/pekko/persistence/query/journal/leveldb/scaladsl/LeveldbReadJournal.scala
+++ b/persistence-query/src/main/scala/org/apache/pekko/persistence/query/journal/leveldb/scaladsl/LeveldbReadJournal.scala
@@ -69,7 +69,8 @@ class LeveldbReadJournal(system: ExtendedActorSystem, config: Config)
else
writeJournalPluginId
require(
- resolvedWriteJournalPluginId.nonEmpty && system.settings.config
+ resolvedWriteJournalPluginId.nonEmpty &&
+ system.settings.config
.getConfig(resolvedWriteJournalPluginId)
.getString("class") == "org.apache.pekko.persistence.journal.leveldb.LeveldbJournal",
s"Leveldb read journal can only work with a Leveldb write journal. Current plugin [$resolvedWriteJournalPluginId] is not a LeveldbJournal")
diff --git a/persistence-tck/src/main/scala/org/apache/pekko/persistence/scalatest/OptionalTests.scala b/persistence-tck/src/main/scala/org/apache/pekko/persistence/scalatest/OptionalTests.scala
index 67b5bce226..ac2b8e8328 100644
--- a/persistence-tck/src/main/scala/org/apache/pekko/persistence/scalatest/OptionalTests.scala
+++ b/persistence-tck/src/main/scala/org/apache/pekko/persistence/scalatest/OptionalTests.scala
@@ -23,8 +23,9 @@ trait OptionalTests {
def optional(flag: CapabilityFlag)(test: => Unit) = {
val msg = s"CapabilityFlag `${flag.name}` was turned `" +
(if (flag.value) "on" else "off") +
- "`. " + (if (!flag.value) "To enable the related tests override it with `CapabilityFlag.on` (or `true` in Scala)."
- else "")
+ "`. " +
+ (if (!flag.value) "To enable the related tests override it with `CapabilityFlag.on` (or `true` in Scala)."
+ else "")
info(msg)
if (flag.value)
try test
diff --git a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/javadsl/PersistenceTestKitReadJournal.scala b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/javadsl/PersistenceTestKitReadJournal.scala
index 49f63684bb..af738f8e0f 100644
--- a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/javadsl/PersistenceTestKitReadJournal.scala
+++ b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/javadsl/PersistenceTestKitReadJournal.scala
@@ -19,11 +19,7 @@ import pekko.japi.Pair
import pekko.persistence.query.EventEnvelope
import pekko.persistence.query.Offset
import pekko.persistence.query.javadsl.{
- CurrentEventsByPersistenceIdQuery,
- CurrentEventsByTagQuery,
- EventsByPersistenceIdQuery,
- EventsByTagQuery,
- ReadJournal
+ CurrentEventsByPersistenceIdQuery, CurrentEventsByTagQuery, EventsByPersistenceIdQuery, EventsByTagQuery, ReadJournal
}
import pekko.persistence.query.typed
import pekko.persistence.query.typed.javadsl.{ CurrentEventsBySliceQuery, EventsBySliceQuery }
diff --git a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/scaladsl/PersistenceTestKitReadJournal.scala b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/scaladsl/PersistenceTestKitReadJournal.scala
index 73c7945160..510cb2cb97 100644
--- a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/scaladsl/PersistenceTestKitReadJournal.scala
+++ b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/scaladsl/PersistenceTestKitReadJournal.scala
@@ -24,10 +24,7 @@ import pekko.persistence.query.{ EventEnvelope, Sequence }
import pekko.persistence.query.NoOffset
import pekko.persistence.query.Offset
import pekko.persistence.query.scaladsl.{
- CurrentEventsByPersistenceIdQuery,
- CurrentEventsByTagQuery,
- EventsByPersistenceIdQuery,
- PagedPersistenceIdsQuery,
+ CurrentEventsByPersistenceIdQuery, CurrentEventsByTagQuery, EventsByPersistenceIdQuery, PagedPersistenceIdsQuery,
ReadJournal
}
import pekko.persistence.query.scaladsl.EventsByTagQuery
diff --git a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/state/scaladsl/PersistenceTestKitDurableStateStore.scala b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/state/scaladsl/PersistenceTestKitDurableStateStore.scala
index 414a9f1962..fb31657241 100644
--- a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/state/scaladsl/PersistenceTestKitDurableStateStore.scala
+++ b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/state/scaladsl/PersistenceTestKitDurableStateStore.scala
@@ -23,12 +23,7 @@ import pekko.{ Done, NotUsed }
import pekko.actor.ExtendedActorSystem
import pekko.persistence.Persistence
import pekko.persistence.query.{
- DeletedDurableState,
- DurableStateChange,
- NoOffset,
- Offset,
- Sequence,
- UpdatedDurableState
+ DeletedDurableState, DurableStateChange, NoOffset, Offset, Sequence, UpdatedDurableState
}
import pekko.persistence.query.scaladsl.{ DurableStateStorePagedPersistenceIdsQuery, DurableStateStoreQuery }
import pekko.persistence.query.typed.scaladsl.DurableStateStoreBySliceQuery
diff --git a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/EventSourcedBehaviorTestKitSpec.scala b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/EventSourcedBehaviorTestKitSpec.scala
index 5c8c4781bf..dd75fbbb2d 100644
--- a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/EventSourcedBehaviorTestKitSpec.scala
+++ b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/EventSourcedBehaviorTestKitSpec.scala
@@ -26,8 +26,7 @@ import pekko.actor.typed.Behavior
import pekko.actor.typed.scaladsl.ActorContext
import pekko.actor.typed.scaladsl.Behaviors
import pekko.persistence.testkit.scaladsl.EventSourcedBehaviorTestKitSpec.TestCounter.{
- NotSerializableState,
- NullState
+ NotSerializableState, NullState
}
import pekko.persistence.typed.PersistenceId
import pekko.persistence.typed.internal.JournalFailureException
diff --git a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/ScalaDslUtils.scala b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/ScalaDslUtils.scala
index 54e44371c6..c67f560039 100644
--- a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/ScalaDslUtils.scala
+++ b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/ScalaDslUtils.scala
@@ -16,15 +16,7 @@ package org.apache.pekko.persistence.testkit.scaladsl
import org.apache.pekko
import pekko.actor.typed.ActorRef
import pekko.persistence.testkit.{
- Cmd,
- CommonUtils,
- EmptyState,
- Evt,
- NonEmptyState,
- Passivate,
- Recovered,
- Stopped,
- TestCommand
+ Cmd, CommonUtils, EmptyState, Evt, NonEmptyState, Passivate, Recovered, Stopped, TestCommand
}
import pekko.persistence.typed.PersistenceId
import pekko.persistence.typed.RecoveryCompleted
diff --git a/persistence-typed-tests/src/test/scala/docs/org/apache/pekko/persistence/typed/ReplicatedAuctionExampleSpec.scala b/persistence-typed-tests/src/test/scala/docs/org/apache/pekko/persistence/typed/ReplicatedAuctionExampleSpec.scala
index 920905327b..9a2bc55e1b 100644
--- a/persistence-typed-tests/src/test/scala/docs/org/apache/pekko/persistence/typed/ReplicatedAuctionExampleSpec.scala
+++ b/persistence-typed-tests/src/test/scala/docs/org/apache/pekko/persistence/typed/ReplicatedAuctionExampleSpec.scala
@@ -138,7 +138,8 @@ object ReplicatedAuctionExampleSpec {
// If timestamps are equal, choose by dc where the offer was submitted
// In real auctions, this last comparison should be deterministic but unpredictable, so that submitting to a
// particular DC would not be an advantage.
- (first.offer == second.offer && first.timestamp.equals(second.timestamp) && first.originReplica.id
+ (first.offer == second.offer && first.timestamp.equals(second.timestamp) &&
+ first.originReplica.id
.compareTo(second.originReplica.id) < 0)
}
// #state
@@ -279,7 +280,8 @@ object ReplicatedAuctionExampleSpec {
}
private def shouldClose(state: AuctionState): Boolean = {
- responsibleForClosing && (state.phase match {
+ responsibleForClosing &&
+ (state.phase match {
case Closing(alreadyFinishedAtDc) =>
val allDone = allReplicas.diff(alreadyFinishedAtDc).isEmpty
if (!allDone) {
diff --git a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/BehaviorSetup.scala b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/BehaviorSetup.scala
index 54bc204356..356db67709 100644
--- a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/BehaviorSetup.scala
+++ b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/BehaviorSetup.scala
@@ -88,7 +88,8 @@ private[pekko] final class BehaviorSetup[C, E, S](
val isSnapshotOptional: Boolean =
Persistence(context.system.classicSystem).configFor(snapshotStore).getBoolean("snapshot-is-optional")
- if (isSnapshotOptional && (retention match {
+ if (isSnapshotOptional &&
+ (retention match {
case SnapshotCountRetentionCriteriaImpl(_, _, true) => true
case _ => false
})) {
diff --git a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/Running.scala b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/Running.scala
index e5f38625e7..35068f4fbd 100644
--- a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/Running.scala
+++ b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/Running.scala
@@ -47,19 +47,9 @@ import pekko.persistence.journal.Tagged
import pekko.persistence.query.{ EventEnvelope, PersistenceQuery }
import pekko.persistence.query.scaladsl.EventsByPersistenceIdQuery
import pekko.persistence.typed.{
- DeleteEventsCompleted,
- DeleteEventsFailed,
- DeleteSnapshotsCompleted,
- DeleteSnapshotsFailed,
- DeletionTarget,
- EventRejectedException,
- JournalPersistFailed,
- JournalPersistRejected,
- PersistenceId,
- SnapshotCompleted,
- SnapshotFailed,
- SnapshotMetadata,
- SnapshotSelectionCriteria
+ DeleteEventsCompleted, DeleteEventsFailed, DeleteSnapshotsCompleted, DeleteSnapshotsFailed, DeletionTarget,
+ EventRejectedException, JournalPersistFailed, JournalPersistRejected, PersistenceId, SnapshotCompleted,
+ SnapshotFailed, SnapshotMetadata, SnapshotSelectionCriteria
}
import pekko.persistence.typed.ReplicaId
import pekko.persistence.typed.ReplicationId
diff --git a/persistence-typed/src/test/scala/org/apache/pekko/persistence/typed/ManyRecoveriesSpec.scala b/persistence-typed/src/test/scala/org/apache/pekko/persistence/typed/ManyRecoveriesSpec.scala
index 21c868594d..2da271e8ac 100644
--- a/persistence-typed/src/test/scala/org/apache/pekko/persistence/typed/ManyRecoveriesSpec.scala
+++ b/persistence-typed/src/test/scala/org/apache/pekko/persistence/typed/ManyRecoveriesSpec.scala
@@ -55,8 +55,7 @@ object ManyRecoveriesSpec {
Behaviors.same
}
- def forN(n: Int)(mapper: Int => String): Set[String] =
- (1 to n).map(mapper).toSet
+ def forN(n: Int)(mapper: Int => String): Set[String] = (1 to n).map(mapper).toSet
}
class ManyRecoveriesSpec extends ScalaTestWithActorTestKit(s"""
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 01b2fe8b90..a947e851fa 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
@@ -21,15 +21,8 @@ import scala.concurrent.duration._
import org.apache.pekko
import pekko.actor._
import pekko.persistence.{
- AtomicWrite,
- DeleteMessagesFailure,
- DeleteSnapshotFailure,
- DeleteSnapshotsFailure,
- JournalProtocol,
- NonPersistentRepr,
- Persistence,
- SaveSnapshotFailure,
- SnapshotProtocol
+ AtomicWrite, DeleteMessagesFailure, DeleteSnapshotFailure, DeleteSnapshotsFailure, JournalProtocol, NonPersistentRepr,
+ Persistence, SaveSnapshotFailure, SnapshotProtocol
}
import pekko.util.Helpers.Requiring
diff --git a/persistence/src/main/scala/org/apache/pekko/persistence/journal/inmem/InmemJournal.scala b/persistence/src/main/scala/org/apache/pekko/persistence/journal/inmem/InmemJournal.scala
index 4888a6f789..235a003010 100644
--- a/persistence/src/main/scala/org/apache/pekko/persistence/journal/inmem/InmemJournal.scala
+++ b/persistence/src/main/scala/org/apache/pekko/persistence/journal/inmem/InmemJournal.scala
@@ -164,10 +164,11 @@ object InmemJournal {
case _ => (p.withTimestamp(System.currentTimeMillis()), OptionVal.None)
}
- messages = messages + (messages.get(p.persistenceId) match {
- case Some(ms) => p.persistenceId -> (ms :+ pr)
- case None => p.persistenceId -> Vector(pr)
- })
+ messages = messages +
+ (messages.get(p.persistenceId) match {
+ case Some(ms) => p.persistenceId -> (ms :+ pr)
+ case None => p.persistenceId -> Vector(pr)
+ })
highestSequenceNumbers =
highestSequenceNumbers.updated(p.persistenceId, math.max(highestSequenceNr(p.persistenceId), p.sequenceNr))
}
diff --git a/persistence/src/test/scala/org/apache/pekko/persistence/AtLeastOnceDeliverySpec.scala b/persistence/src/test/scala/org/apache/pekko/persistence/AtLeastOnceDeliverySpec.scala
index ad2f108c81..9753c9e08e 100644
--- a/persistence/src/test/scala/org/apache/pekko/persistence/AtLeastOnceDeliverySpec.scala
+++ b/persistence/src/test/scala/org/apache/pekko/persistence/AtLeastOnceDeliverySpec.scala
@@ -207,7 +207,8 @@ class AtLeastOnceDeliverySpec
"AtLeastOnceDelivery" must {
List(true, false).foreach { deliverUsingActorSelection =>
- s"deliver messages in order when nothing is lost (using actorSelection: $deliverUsingActorSelection)" taggedAs TimingTest in {
+ s"deliver messages in order when nothing is lost (using actorSelection: $deliverUsingActorSelection)" taggedAs
+ TimingTest in {
val probe = TestProbe()
val probeA = TestProbe()
val destinations = Map("A" -> system.actorOf(destinationProps(probeA.ref)).path)
@@ -293,7 +294,8 @@ class AtLeastOnceDeliverySpec
probeA.expectNoMessage(1.second)
}
- "re-send replayed deliveries with an 'initially in-order' strategy, before delivering fresh messages" taggedAs TimingTest in {
+ "re-send replayed deliveries with an 'initially in-order' strategy, before delivering fresh messages" taggedAs
+ TimingTest in {
val probe = TestProbe()
val probeA = TestProbe()
val dst = system.actorOf(destinationProps(probeA.ref))
diff --git a/project/Doc.scala b/project/Doc.scala
index a446a58aeb..8ba8b57c14 100644
--- a/project/Doc.scala
+++ b/project/Doc.scala
@@ -152,8 +152,7 @@ object UnidocRoot extends AutoPlugin {
ScalaUnidoc / unidocProjectFilter := unidocRootProjectFilter(unidocRootIgnoreProjects.value),
JavaUnidoc / unidocProjectFilter := unidocRootProjectFilter(unidocRootIgnoreProjects.value),
Compile / doc / apiMappings ++= {
- val entries: Seq[Attributed[File]] =
- (LocalProject("slf4j") / Compile / fullClasspath).value ++
+ val entries: Seq[Attributed[File]] = (LocalProject("slf4j") / Compile / fullClasspath).value ++
(LocalProject("persistence") / Compile / fullClasspath).value ++
(LocalProject("remote") / Compile / fullClasspath).value ++
(LocalProject("stream") / Compile / fullClasspath).value
diff --git a/project/Paradox.scala b/project/Paradox.scala
index 5e8515ad59..b9b39562f9 100644
--- a/project/Paradox.scala
+++ b/project/Paradox.scala
@@ -27,7 +27,8 @@ object Paradox {
lazy val propertiesSettings = Seq(
Compile / paradoxProperties ++= Map(
"canonical.base_url" -> s"$pekkoBaseURL/docs/pekko/current",
- "github.base_url" -> GitHub
+ "github.base_url" ->
+ GitHub
.url(version.value), // for links like this: @github[#1](#1) or @github[83986f9](83986f9)
"extref.pekko.http.base_url" -> s"$pekkoBaseURL/docs/pekko-http/current/%s",
"extref.pekko-management.base_url" -> s"$pekkoBaseURL/docs/pekko-management/current/%s",
@@ -50,9 +51,11 @@ object Paradox {
"javadoc.org.apache.pekko.link_style" -> "direct",
"javadoc.org.apache.pekko.http.base_url" -> s"$pekkoBaseURL/japi/pekko-http/current",
"javadoc.org.apache.pekko.http.link_style" -> "frames",
- "javadoc.com.fasterxml.jackson.annotation.base_url" -> "https://javadoc.io/doc/com.fasterxml.jackson.core/jackson-annotations/latest/",
+ "javadoc.com.fasterxml.jackson.annotation.base_url" ->
+ "https://javadoc.io/doc/com.fasterxml.jackson.core/jackson-annotations/latest/",
"javadoc.com.fasterxml.jackson.annotation.link_style" -> "direct",
- "javadoc.com.fasterxml.jackson.databind.base_url" -> "https://javadoc.io/doc/com.fasterxml.jackson.core/jackson-databind/latest/",
+ "javadoc.com.fasterxml.jackson.databind.base_url" ->
+ "https://javadoc.io/doc/com.fasterxml.jackson.core/jackson-databind/latest/",
"javadoc.com.fasterxml.jackson.databind.link_style" -> "direct",
"javadoc.com.google.protobuf.base_url" -> "https://javadoc.io/doc/com.google.protobuf/protobuf-java/latest/",
"javadoc.com.google.protobuf.link_style" -> "direct",
diff --git a/project/PekkoBuild.scala b/project/PekkoBuild.scala
index 2b69b3f914..bff7ab34bf 100644
--- a/project/PekkoBuild.scala
+++ b/project/PekkoBuild.scala
@@ -40,9 +40,10 @@ object PekkoBuild {
lazy val rootSettings = Def.settings(
UnidocRoot.pekkoSettings,
Protobuf.settings,
- GlobalScope / parallelExecution := System
- .getProperty("pekko.parallelExecution", parallelExecutionByDefault.toString)
- .toBoolean,
+ GlobalScope / parallelExecution :=
+ System
+ .getProperty("pekko.parallelExecution", parallelExecutionByDefault.toString)
+ .toBoolean,
// used for linking to API docs (overwrites `project-info.version`)
ThisBuild / projectInfoVersion := { if (isSnapshot.value) "snapshot" else version.value })
@@ -152,7 +153,8 @@ object PekkoBuild {
ThisBuild / ivyLoggingLevel := UpdateLogging.Quiet,
licenses := Seq(("Apache-2.0", url("https://www.apache.org/licenses/LICENSE-2.0.html"))),
homepage := Some(url("https://pekko.apache.org/")),
- description := "Apache Pekko is a toolkit for building highly concurrent, distributed, and resilient message-driven applications for Java and Scala.",
+ description :=
+ "Apache Pekko is a toolkit for building highly concurrent, distributed, and resilient message-driven applications for Java and Scala.",
scmInfo := Some(
ScmInfo(
url("https://github.com/apache/pekko"),
@@ -200,7 +202,8 @@ object PekkoBuild {
// faster random source
"-Djava.security.egd=file:/dev/./urandom")
- defaults ++ CliOptions.runningOnCi
+ defaults ++
+ CliOptions.runningOnCi
.ifTrue(jvmGCLogOptions)
.getOrElse(Nil) ++
JdkOptions.versionSpecificJavaOptions
@@ -232,9 +235,10 @@ object PekkoBuild {
}
}
},
- Test / parallelExecution := System
- .getProperty("pekko.parallelExecution", parallelExecutionByDefault.toString)
- .toBoolean,
+ Test / parallelExecution :=
+ System
+ .getProperty("pekko.parallelExecution", parallelExecutionByDefault.toString)
+ .toBoolean,
Test / logBuffered := System.getProperty("pekko.logBufferedTests", "false").toBoolean,
// show full stack traces and test case durations
Test / testOptions += Tests.Argument("-oDF"),
diff --git a/project/PekkoDevelocityPlugin.scala b/project/PekkoDevelocityPlugin.scala
index 01a0602847..0092a15d65 100644
--- a/project/PekkoDevelocityPlugin.scala
+++ b/project/PekkoDevelocityPlugin.scala
@@ -17,10 +17,7 @@
import com.gradle.develocity.agent.sbt.DevelocityPlugin
import com.gradle.develocity.agent.sbt.DevelocityPlugin.autoImport.{
- develocityConfiguration,
- FlakyTestPolicy,
- ProjectId,
- Publishing
+ develocityConfiguration, FlakyTestPolicy, ProjectId, Publishing
}
import sbt.{ inConfig, url, AutoPlugin, Def, PluginTrigger, Plugins, Setting }
import sbt.Keys.insideCI
diff --git a/project/PekkoDisciplinePlugin.scala b/project/PekkoDisciplinePlugin.scala
index ef4a66b2f1..48212954c2 100644
--- a/project/PekkoDisciplinePlugin.scala
+++ b/project/PekkoDisciplinePlugin.scala
@@ -102,18 +102,20 @@ object PekkoDisciplinePlugin extends AutoPlugin {
lazy val docs =
Seq(
Compile / scalacOptions -= defaultScalaOptions.value,
- Compile / scalacOptions ++= (CrossVersion.partialVersion(scalaVersion.value).get match {
- case (3, _) => Nil
- case (2, 13) => Seq("-Wconf:any:e,cat=unused:s,cat=deprecation:s,cat=unchecked:s")
- case (2, 12) => Seq("-Wconf:cat=unused:s,cat=deprecation:s,cat=unchecked:s,any:e")
- }),
+ Compile / scalacOptions ++=
+ (CrossVersion.partialVersion(scalaVersion.value).get match {
+ case (3, _) => Nil
+ case (2, 13) => Seq("-Wconf:any:e,cat=unused:s,cat=deprecation:s,cat=unchecked:s")
+ case (2, 12) => Seq("-Wconf:cat=unused:s,cat=deprecation:s,cat=unchecked:s,any:e")
+ }),
Test / scalacOptions --= Seq("-Xlint", "-unchecked", "-deprecation"),
Test / scalacOptions -= defaultScalaOptions.value,
- Test / scalacOptions ++= (CrossVersion.partialVersion(scalaVersion.value).get match {
- case (3, _) => Nil
- case (2, 13) => Seq("-Wconf:any:e,cat=unused:s,cat=deprecation:s,cat=unchecked:s")
- case (2, 12) => Seq("-Wconf:cat=unused:s,cat=deprecation:s,cat=unchecked:s,any:e")
- }),
+ Test / scalacOptions ++=
+ (CrossVersion.partialVersion(scalaVersion.value).get match {
+ case (3, _) => Nil
+ case (2, 13) => Seq("-Wconf:any:e,cat=unused:s,cat=deprecation:s,cat=unchecked:s")
+ case (2, 12) => Seq("-Wconf:cat=unused:s,cat=deprecation:s,cat=unchecked:s,any:e")
+ }),
Compile / doc / scalacOptions := Seq())
lazy val disciplineSettings =
@@ -129,21 +131,22 @@ object PekkoDisciplinePlugin extends AutoPlugin {
}
),
Compile / doc / javacOptions := Seq("-Xdoclint:none"),
- Compile / scalacOptions ++= (CrossVersion.partialVersion(scalaVersion.value) match {
- case Some((2, 13)) =>
- disciplineScalacOptions -- Set(
- "-Ywarn-inaccessible",
- "-Ywarn-infer-any",
- "-Ywarn-nullary-override",
- "-Ywarn-nullary-unit",
- "-Ypartial-unification",
- "-Yno-adapted-args") ++ Set(
- "-Xlint:-strict-unsealed-patmat")
- case Some((2, 12)) =>
- disciplineScalacOptions
- case _ =>
- Seq("-Wconf:cat=deprecation:s")
- }).toSeq,
+ Compile / scalacOptions ++=
+ (CrossVersion.partialVersion(scalaVersion.value) match {
+ case Some((2, 13)) =>
+ disciplineScalacOptions -- Set(
+ "-Ywarn-inaccessible",
+ "-Ywarn-infer-any",
+ "-Ywarn-nullary-override",
+ "-Ywarn-nullary-unit",
+ "-Ypartial-unification",
+ "-Yno-adapted-args") ++ Set(
+ "-Xlint:-strict-unsealed-patmat")
+ case Some((2, 12)) =>
+ disciplineScalacOptions
+ case _ =>
+ Seq("-Wconf:cat=deprecation:s")
+ }).toSeq,
Compile / scalacOptions --=
(if (looseProjects.contains(name.value)) undisciplineScalacOptions.toSeq
else Seq.empty),
diff --git a/project/SbtMultiJvmPlugin.scala b/project/SbtMultiJvmPlugin.scala
index 726dfd14d3..c07b27ac99 100644
--- a/project/SbtMultiJvmPlugin.scala
+++ b/project/SbtMultiJvmPlugin.scala
@@ -248,7 +248,8 @@ object MultiJvmPlugin extends AutoPlugin {
.foreach(classpathFile =>
IO.copyFile(classpathFile, new File(multiRunCopiedClassDir, classpathFile.getName), true))
val cp =
- directoryBasedClasspathEntries.absString + File.pathSeparator + multiRunCopiedClassDir.getAbsolutePath + File
+ directoryBasedClasspathEntries.absString + File.pathSeparator + multiRunCopiedClassDir.getAbsolutePath +
+ File
.separator + "*"
(testClass: String) => { Seq("-cp", cp, runner, "-s", testClass) ++ options }
}
diff --git a/project/StreamOperatorsIndexGenerator.scala b/project/StreamOperatorsIndexGenerator.scala
index 93b0d57d25..4da0cf2b3c 100644
--- a/project/StreamOperatorsIndexGenerator.scala
+++ b/project/StreamOperatorsIndexGenerator.scala
@@ -89,7 +89,8 @@ object StreamOperatorsIndexGenerator extends AutoPlugin {
// FIXME document these methods as well
val pendingTestCases = Map(
"Source" -> pendingSourceOrFlow,
- "Flow" -> (pendingSourceOrFlow ++ Seq(
+ "Flow" ->
+ (pendingSourceOrFlow ++ Seq(
"fromProcessorMat",
"toProcessor",
"fromProcessor",
diff --git a/project/TestExtras.scala b/project/TestExtras.scala
index 106724994a..c53ed776ca 100644
--- a/project/TestExtras.scala
+++ b/project/TestExtras.scala
@@ -62,10 +62,13 @@ object TestExtras {
val baseList =
List(
- "The java JavaExtension.java" -> "actor-tests/target/test-reports/TEST-org.apache.pekko.actor.JavaExtension.xml")
+ "The java JavaExtension.java" ->
+ "actor-tests/target/test-reports/TEST-org.apache.pekko.actor.JavaExtension.xml")
val jdk9Only = List(
- "The jdk9-only FlowPublisherSinkSpec.scala" -> "stream-tests/target/test-reports/TEST-org.apache.pekko.stream.scaladsl.FlowPublisherSinkSpec.xml",
- "The jdk9-only JavaFlowSupportCompileTest.java" -> "stream-tests/target/test-reports/TEST-org.apache.pekko.stream.javadsl.JavaFlowSupportCompileTest.xml")
+ "The jdk9-only FlowPublisherSinkSpec.scala" ->
+ "stream-tests/target/test-reports/TEST-org.apache.pekko.stream.scaladsl.FlowPublisherSinkSpec.xml",
+ "The jdk9-only JavaFlowSupportCompileTest.java" ->
+ "stream-tests/target/test-reports/TEST-org.apache.pekko.stream.javadsl.JavaFlowSupportCompileTest.xml")
val testsToCheck =
baseList ::: jdk9Only
diff --git a/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/routing/RemoteRandomSpec.scala b/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/routing/RemoteRandomSpec.scala
index 54d8ab3c00..f0a48e4a81 100644
--- a/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/routing/RemoteRandomSpec.scala
+++ b/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/routing/RemoteRandomSpec.scala
@@ -78,7 +78,8 @@ class RemoteRandomSpec(multiNodeConfig: RemoteRandomConfig)
def initialParticipants = roles.size
"A remote random pool" must {
- "be locally instantiated on a remote node and be able to communicate through its RemoteActorRef" taggedAs LongRunningTest in {
+ "be locally instantiated on a remote node and be able to communicate through its RemoteActorRef" taggedAs
+ LongRunningTest in {
runOn(first, second, third) {
enterBarrier("start", "broadcast-end", "end", "done")
diff --git a/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/routing/RemoteScatterGatherSpec.scala b/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/routing/RemoteScatterGatherSpec.scala
index 4fb030145f..fb4d607afe 100644
--- a/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/routing/RemoteScatterGatherSpec.scala
+++ b/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/routing/RemoteScatterGatherSpec.scala
@@ -82,7 +82,8 @@ class RemoteScatterGatherSpec(multiNodeConfig: RemoteScatterGatherConfig)
def initialParticipants = roles.size
"A remote ScatterGatherFirstCompleted pool" must {
- "be locally instantiated on a remote node and be able to communicate through its RemoteActorRef" taggedAs LongRunningTest in {
+ "be locally instantiated on a remote node and be able to communicate through its RemoteActorRef" taggedAs
+ LongRunningTest in {
system.eventStream.publish(Mute(EventFilter.warning(pattern = ".*received dead letter from.*")))
diff --git a/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala b/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
index 3b3facfce6..be63b5c6dd 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/Endpoint.scala
@@ -38,10 +38,7 @@ import pekko.remote.EndpointWriter.{ FlushAndStop, StoppedReading }
import pekko.remote.WireFormats.SerializedMessage
import pekko.remote.transport._
import pekko.remote.transport.AssociationHandle.{
- ActorHandleEventListener,
- DisassociateInfo,
- Disassociated,
- InboundPayload
+ ActorHandleEventListener, DisassociateInfo, Disassociated, InboundPayload
}
import pekko.remote.transport.PekkoPduCodec.Message
import pekko.remote.transport.Transport.InvalidAssociationException
@@ -108,7 +105,8 @@ private[remote] class DefaultMessageDispatcher(
logMessageReceived("local message")
payload match {
case sel: ActorSelectionMessage =>
- if (UntrustedMode && (!TrustedSelectionPaths.contains(sel.elements.mkString("/", "/", "")) ||
+ if (UntrustedMode &&
+ (!TrustedSelectionPaths.contains(sel.elements.mkString("/", "/", "")) ||
sel.msg.isInstanceOf[PossiblyHarmful] || l != provider.rootGuardian))
log.debug(
LogMarker.Security,
diff --git a/remote/src/main/scala/org/apache/pekko/remote/RemoteDaemon.scala b/remote/src/main/scala/org/apache/pekko/remote/RemoteDaemon.scala
index 01509ec4b3..e7c195669c 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/RemoteDaemon.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/RemoteDaemon.scala
@@ -21,17 +21,8 @@ import scala.util.control.NonFatal
import org.apache.pekko
import pekko.actor.{
- Actor,
- ActorPath,
- ActorPathExtractor,
- ActorRef,
- ActorSystemImpl,
- AddressTerminated,
- Deploy,
- InternalActorRef,
- Nobody,
- Props,
- VirtualPathContainer
+ Actor, ActorPath, ActorPathExtractor, ActorRef, ActorSystemImpl, AddressTerminated, Deploy, InternalActorRef, Nobody,
+ Props, VirtualPathContainer
}
import pekko.actor.ActorIdentity
import pekko.actor.ActorRefScope
diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
index 4d4aff0d50..f453a14044 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
@@ -805,8 +805,8 @@ private[remote] abstract class ArteryTransport(_system: ExtendedActorSystem, _pr
def outboundControl(
outboundContext: OutboundContext): Sink[OutboundEnvelope, (OutboundControlIngress, Future[Done])] = {
- val livenessProbeInterval =
- (settings.Advanced.QuarantineIdleOutboundAfter / 10).max(settings.Advanced.HandshakeRetryInterval)
+ val livenessProbeInterval = (settings.Advanced.QuarantineIdleOutboundAfter / 10).max(
+ settings.Advanced.HandshakeRetryInterval)
Flow
.fromGraph(killSwitch.flow[OutboundEnvelope])
.via(
diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
index eda39552d4..0df2d90350 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala
@@ -1024,7 +1024,8 @@ private[remote] class Association(
}
def isConnectException: Boolean =
- cause.isInstanceOf[StreamTcpException] && (cause.getCause ne null) && cause.getCause
+ cause.isInstanceOf[StreamTcpException] && (cause.getCause ne null) &&
+ cause.getCause
.isInstanceOf[ConnectException]
if (stoppedIdle) {
diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/EnvelopeBufferPool.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/EnvelopeBufferPool.scala
index 7cd602715a..5da684ebe0 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/EnvelopeBufferPool.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/EnvelopeBufferPool.scala
@@ -324,8 +324,7 @@ private[remote] final class HeaderBuilderImpl(
_senderActorRef = null
_senderActorRefIdx = DeadLettersCode
}
- override def isNoSender: Boolean =
- (_senderActorRef eq null) && _senderActorRefIdx == DeadLettersCode
+ override def isNoSender: Boolean = (_senderActorRef eq null) && _senderActorRefIdx == DeadLettersCode
override def senderActorRef(originUid: Long): OptionVal[ActorRef] = {
// we treat deadLetters as always present, but not included in table
if ((_senderActorRef eq null) && !isNoSender)
@@ -340,8 +339,7 @@ private[remote] final class HeaderBuilderImpl(
_recipientActorRef = null
_recipientActorRefIdx = DeadLettersCode
}
- def isNoRecipient: Boolean =
- (_recipientActorRef eq null) && _recipientActorRefIdx == DeadLettersCode
+ def isNoRecipient: Boolean = (_recipientActorRef eq null) && _recipientActorRefIdx == DeadLettersCode
/**
* Note that Serialization.currentTransportInformation must be set when calling this method,
diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/MessageDispatcher.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/MessageDispatcher.scala
index f66b507073..ba9ae0241c 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/MessageDispatcher.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/MessageDispatcher.scala
@@ -70,7 +70,8 @@ private[remote] class MessageDispatcher(system: ExtendedActorSystem, provider: R
log.debug("received message [{}] to [{}] from [{}]", message, recipient, senderOption.getOrElse(""))
message match {
case sel: ActorSelectionMessage =>
- if (UntrustedMode && (!TrustedSelectionPaths.contains(sel.elements.mkString("/", "/", "")) ||
+ if (UntrustedMode &&
+ (!TrustedSelectionPaths.contains(sel.elements.mkString("/", "/", "")) ||
sel.msg.isInstanceOf[PossiblyHarmful] || l != provider.rootGuardian)) {
if (debugLogEnabled)
log.debug(
diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/AeronSink.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/AeronSink.scala
index 679585ca47..adeced9d48 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/AeronSink.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/AeronSink.scala
@@ -80,7 +80,8 @@ private[remote] object AeronSink {
} else if (result == Publication.CLOSED) {
onPublicationClosed.invoke(())
true
- } else if (giveUpAfterNanos >= 0 && (n & TimerCheckMask) == 0 && (System
+ } else if (giveUpAfterNanos >= 0 && (n & TimerCheckMask) == 0 &&
+ (System
.nanoTime() - startTime) > giveUpAfterNanos) {
// the task is invoked by the spinning thread, only check nanoTime each 8192th invocation
n = 0L
diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/ArteryAeronUdpTransport.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/ArteryAeronUdpTransport.scala
index bdfe30123d..e18953e0ea 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/ArteryAeronUdpTransport.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/ArteryAeronUdpTransport.scala
@@ -405,12 +405,11 @@ private[remote] class ArteryAeronUdpTransport(_system: ExtendedActorSystem, _pro
.run()(materializer)
val lane = inboundSink(envelopeBufferPool)
- val completedValues: Vector[Future[Done]] =
- (0 until inboundLanes).iterator
- .map { _ =>
- laneHub.toMat(lane)(Keep.right).run()(materializer)
- }
- .to(immutable.Vector)
+ val completedValues: Vector[Future[Done]] = (0 until inboundLanes).iterator
+ .map { _ =>
+ laneHub.toMat(lane)(Keep.right).run()(materializer)
+ }
+ .to(immutable.Vector)
implicit val ec = system.dispatchers.internalDispatcher
diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/CompressionTable.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/CompressionTable.scala
index 96866fd673..7da34aa2ae 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/CompressionTable.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/CompressionTable.scala
@@ -49,7 +49,8 @@ private[remote] final class CompressionTable[T](
import scala.jdk.CollectionConverters._
require(
_dictionary.values.iterator().asScala.min == 0,
- "Compression table should start allocating from 0, yet lowest allocated id was " + _dictionary.values
+ "Compression table should start allocating from 0, yet lowest allocated id was " +
+ _dictionary.values
.iterator()
.asScala
.min)
diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/ArteryTcpTransport.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/ArteryTcpTransport.scala
index 5f1879bcd2..10069b2168 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/ArteryTcpTransport.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/tcp/ArteryTcpTransport.scala
@@ -441,12 +441,11 @@ private[remote] class ArteryTcpTransport(
.run()(materializer)
val lane = inboundSink(envelopeBufferPool)
- val completedValues: Vector[Future[Done]] =
- (0 until inboundLanes).iterator
- .map { _ =>
- laneHub.toMat(lane)(Keep.right).run()(materializer)
- }
- .to(immutable.Vector)
+ val completedValues: Vector[Future[Done]] = (0 until inboundLanes).iterator
+ .map { _ =>
+ laneHub.toMat(lane)(Keep.right).run()(materializer)
+ }
+ .to(immutable.Vector)
implicit val ec = system.dispatchers.internalDispatcher
diff --git a/remote/src/main/scala/org/apache/pekko/remote/transport/AbstractTransportAdapter.scala b/remote/src/main/scala/org/apache/pekko/remote/transport/AbstractTransportAdapter.scala
index ba0d60d571..97dd60a335 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/transport/AbstractTransportAdapter.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/transport/AbstractTransportAdapter.scala
@@ -42,7 +42,8 @@ class TransportAdapters(system: ExtendedActorSystem) extends Extension {
val settings = RARP(system).provider.remoteSettings
private val adaptersTable: Map[String, TransportAdapterProvider] = for ((name, fqn) <- settings.Adapters) yield {
- name -> system.dynamicAccess
+ name ->
+ system.dynamicAccess
.createInstanceFor[TransportAdapterProvider](fqn, immutable.Seq.empty)
.recover {
case e => throw new IllegalArgumentException(s"Cannot instantiate transport adapter [$fqn]", e)
diff --git a/remote/src/main/scala/org/apache/pekko/remote/transport/ThrottlerTransportAdapter.scala b/remote/src/main/scala/org/apache/pekko/remote/transport/ThrottlerTransportAdapter.scala
index 2660b21b19..daec54343d 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/transport/ThrottlerTransportAdapter.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/transport/ThrottlerTransportAdapter.scala
@@ -35,11 +35,7 @@ import pekko.pattern.{ ask, pipe, PromiseActorRef }
import pekko.remote.RARP
import pekko.remote.transport.ActorTransportAdapter.AssociateUnderlying
import pekko.remote.transport.AssociationHandle.{
- ActorHandleEventListener,
- DisassociateInfo,
- Disassociated,
- HandleEventListener,
- InboundPayload
+ ActorHandleEventListener, DisassociateInfo, Disassociated, HandleEventListener, InboundPayload
}
import pekko.remote.transport.PekkoPduCodec.Associate
import pekko.remote.transport.ThrottlerManager.{ Checkin, Handle, Listener, ListenerAndMode }
diff --git a/remote/src/main/scala/org/apache/pekko/remote/transport/netty/NettyTransport.scala b/remote/src/main/scala/org/apache/pekko/remote/transport/netty/NettyTransport.scala
index 6cad95e682..56ab768c3d 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/transport/netty/NettyTransport.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/transport/netty/NettyTransport.scala
@@ -38,20 +38,10 @@ import pekko.util.Helpers.Requiring
import io.netty.bootstrap.{ Bootstrap => ClientBootstrap, ServerBootstrap }
import io.netty.buffer.{
- AdaptiveByteBufAllocator,
- ByteBufAllocator,
- PooledByteBufAllocator,
- Unpooled,
- UnpooledByteBufAllocator
+ AdaptiveByteBufAllocator, ByteBufAllocator, PooledByteBufAllocator, Unpooled, UnpooledByteBufAllocator
}
import io.netty.channel.{
- Channel,
- ChannelFuture,
- ChannelHandlerContext,
- ChannelInitializer,
- ChannelOption,
- ChannelPipeline,
- EventLoopGroup,
+ Channel, ChannelFuture, ChannelHandlerContext, ChannelInitializer, ChannelOption, ChannelPipeline, EventLoopGroup,
MultiThreadIoEventLoopGroup
}
import io.netty.channel.group.{ ChannelGroup, ChannelGroupFuture, ChannelMatchers, DefaultChannelGroup }
diff --git a/remote/src/test/scala/org/apache/pekko/remote/AccrualFailureDetectorSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/AccrualFailureDetectorSpec.scala
index f9b1cdd118..d547203ff8 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/AccrualFailureDetectorSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/AccrualFailureDetectorSpec.scala
@@ -83,7 +83,9 @@ class AccrualFailureDetectorSpec extends PekkoSpec("pekko.loglevel = INFO") {
}
// larger stdDeviation results => lower phi
- fd.phi(timeDiff = 1100, mean = 1000.0, stdDeviation = 500.0) should be < (fd
+ fd.phi(timeDiff = 1100, mean = 1000.0, stdDeviation = 500.0) should
+ be <
+ (fd
.phi(timeDiff = 1100, mean = 1000.0, stdDeviation = 100.0))
}
@@ -142,7 +144,8 @@ class AccrualFailureDetectorSpec extends PekkoSpec("pekko.loglevel = INFO") {
// 1000 regular intervals, 5 minute pause, and then a short pause again that should trigger unreachable again
val regularIntervals = 0L +: Vector.fill(999)(1000L)
val timeIntervals =
- regularIntervals :+ (5 * 60 * 1000L) :+ 100L :+ 900L :+ 100L :+ 7000L :+ 100L :+ 900L :+ 100L :+ 900L
+ regularIntervals :+
+ (5 * 60 * 1000L) :+ 100L :+ 900L :+ 100L :+ 7000L :+ 100L :+ 900L :+ 100L :+ 900L
val fd = createFailureDetector(
threshold = 8,
acceptableLostDuration = 3.seconds,
diff --git a/remote/src/test/scala/org/apache/pekko/remote/DeadlineFailureDetectorSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/DeadlineFailureDetectorSpec.scala
index de75fb8afa..5477a76e39 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/DeadlineFailureDetectorSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/DeadlineFailureDetectorSpec.scala
@@ -65,7 +65,8 @@ class DeadlineFailureDetectorSpec extends PekkoSpec {
// 1000 regular intervals, 5 minute pause, and then a short pause again that should trigger unreachable again
val regularIntervals = 0L +: Vector.fill(999)(1000L)
val timeIntervals =
- regularIntervals :+ (5 * 60 * 1000L) :+ 100L :+ 900L :+ 100L :+ 7000L :+ 100L :+ 900L :+ 100L :+ 900L
+ regularIntervals :+
+ (5 * 60 * 1000L) :+ 100L :+ 900L :+ 100L :+ 7000L :+ 100L :+ 900L :+ 100L :+ 900L
val fd = createFailureDetector(acceptableLostDuration = 4.seconds, clock = fakeTimeGenerator(timeIntervals))
for (_ <- 0 until 1000) fd.heartbeat()
diff --git a/remote/src/test/scala/org/apache/pekko/remote/artery/ImmutableLongMapSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/artery/ImmutableLongMapSpec.scala
index dd862885f2..309aa99af5 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/artery/ImmutableLongMapSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/artery/ImmutableLongMapSpec.scala
@@ -78,7 +78,8 @@ class ImmutableLongMapSpec extends AnyWordSpec with Matchers {
ImmutableLongMap.empty[String].updated(10L, "10").updated(20, "20").updated(30, "30").hashCode should be(
ImmutableLongMap.empty[String].updated(10L, "10").updated(20, "20").updated(30, "30").hashCode)
- ImmutableLongMap.empty[String].updated(10L, "10").updated(20, "20") should not be (ImmutableLongMap
+ ImmutableLongMap.empty[String].updated(10L, "10").updated(20, "20") should not be
+ (ImmutableLongMap
.empty[String]
.updated(10L, "10"))
@@ -86,7 +87,8 @@ class ImmutableLongMapSpec extends AnyWordSpec with Matchers {
.empty[String]
.updated(10L, "10")
.updated(20, "20")
- .updated(30, "30") should not be (ImmutableLongMap
+ .updated(30, "30") should not be
+ (ImmutableLongMap
.empty[String]
.updated(10L, "10")
.updated(20, "20b")
@@ -96,7 +98,8 @@ class ImmutableLongMapSpec extends AnyWordSpec with Matchers {
.empty[String]
.updated(10L, "10")
.updated(20, "20")
- .updated(30, "30") should not be (ImmutableLongMap
+ .updated(30, "30") should not be
+ (ImmutableLongMap
.empty[String]
.updated(10L, "10")
.updated(20, "20b")
diff --git a/remote/src/test/scala/org/apache/pekko/remote/artery/OutboundIdleShutdownSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/artery/OutboundIdleShutdownSpec.scala
index 8569262c35..088d0da555 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/artery/OutboundIdleShutdownSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/artery/OutboundIdleShutdownSpec.scala
@@ -262,7 +262,8 @@ class OutboundIdleShutdownSpec extends ArteryMultiNodeSpec("""
localArtery.inboundCompressionAccess.get.currentCompressionOriginUids.futureValue should contain(remoteUid2)
}
eventually {
- localArtery.inboundCompressionAccess.get.currentCompressionOriginUids.futureValue should not contain remoteUid
+ localArtery.inboundCompressionAccess.get.currentCompressionOriginUids.futureValue should not contain
+ remoteUid
}
} finally {
shutdown(remoteSystem2)
diff --git a/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteActorSelectionSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteActorSelectionSpec.scala
index 30461e2bd8..96d83b4542 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteActorSelectionSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteActorSelectionSpec.scala
@@ -17,16 +17,7 @@ import scala.concurrent.duration._
import org.apache.pekko
import pekko.actor.{
- Actor,
- ActorIdentity,
- ActorLogging,
- ActorRef,
- ActorRefScope,
- ActorSelection,
- Identify,
- PoisonPill,
- Props,
- Terminated
+ Actor, ActorIdentity, ActorLogging, ActorRef, ActorRefScope, ActorSelection, Identify, PoisonPill, Props, Terminated
}
import pekko.testkit.{ ImplicitSender, TestActors }
import pekko.testkit.JavaSerializable
diff --git a/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteWatcherSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteWatcherSpec.scala
index a51eafdf87..1a5402a5be 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteWatcherSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteWatcherSpec.scala
@@ -239,7 +239,8 @@ class RemoteWatcherSpec
expectNoMessage(2.seconds)
}
- "generate AddressTerminated for new watch after broken connection that was re-established and broken again" taggedAs LongRunningTest in {
+ "generate AddressTerminated for new watch after broken connection that was re-established and broken again" taggedAs
+ LongRunningTest in {
val p = TestProbe()
val q = TestProbe()
system.eventStream.subscribe(p.ref, classOf[TestRemoteWatcher.AddressTerm])
diff --git a/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
index 2f845f96b2..c927853dfb 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
@@ -34,8 +34,8 @@ class TcpFramingSpec extends PekkoSpec("""
private val payload5 = ByteString((1 to 5).map(_.toByte).toArray)
- private def frameBytes(numberOfFrames: Int): ByteString =
- (1 to numberOfFrames).foldLeft(ByteString.empty)((acc, _) => acc ++ encodeFrameHeader(payload5.size) ++ payload5)
+ private def frameBytes(numberOfFrames: Int): ByteString = (1 to numberOfFrames).foldLeft(ByteString.empty)((acc, _) =>
+ acc ++ encodeFrameHeader(payload5.size) ++ payload5)
private val rndSeed = System.currentTimeMillis()
private val rnd = new Random(rndSeed)
diff --git a/remote/src/test/scala/org/apache/pekko/remote/classic/RemoteWatcherSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/classic/RemoteWatcherSpec.scala
index c913711e79..7dbc07acac 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/classic/RemoteWatcherSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/classic/RemoteWatcherSpec.scala
@@ -254,7 +254,8 @@ class RemoteWatcherSpec extends PekkoSpec("""
expectNoMessage(2.seconds)
}
- "generate AddressTerminated for new watch after broken connection that was re-established and broken again" taggedAs LongRunningTest in {
+ "generate AddressTerminated for new watch after broken connection that was re-established and broken again" taggedAs
+ LongRunningTest in {
val p = TestProbe()
val q = TestProbe()
system.eventStream.subscribe(p.ref, classOf[TestRemoteWatcher.AddressTerm])
diff --git a/remote/src/test/scala/org/apache/pekko/remote/classic/transport/PekkoProtocolSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/classic/transport/PekkoProtocolSpec.scala
index f869337eed..c591400f74 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/classic/transport/PekkoProtocolSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/classic/transport/PekkoProtocolSpec.scala
@@ -26,10 +26,7 @@ import pekko.remote.{ FailureDetector, WireFormats }
import pekko.remote.classic.transport.PekkoProtocolSpec.TestFailureDetector
import pekko.remote.transport.{ AssociationRegistry => _, _ }
import pekko.remote.transport.AssociationHandle.{
- ActorHandleEventListener,
- DisassociateInfo,
- Disassociated,
- InboundPayload
+ ActorHandleEventListener, DisassociateInfo, Disassociated, InboundPayload
}
import pekko.remote.transport.PekkoPduCodec.{ Associate, Disassociate, Heartbeat }
import pekko.remote.transport.ProtocolStateActor
diff --git a/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala
index 7810e6acc9..e42b45d050 100644
--- a/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala
+++ b/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala
@@ -23,9 +23,7 @@ import pekko.remote.artery.Flush
import pekko.remote.artery.FlushAck
import pekko.remote.artery.OutboundHandshake.{ HandshakeReq, HandshakeRsp }
import pekko.remote.artery.compress.CompressionProtocol.{
- ActorRefCompressionAdvertisement,
- ActorRefCompressionAdvertisementAck,
- ClassManifestCompressionAdvertisement,
+ ActorRefCompressionAdvertisement, ActorRefCompressionAdvertisementAck, ClassManifestCompressionAdvertisement,
ClassManifestCompressionAdvertisementAck
}
import pekko.remote.artery.compress.CompressionTable
diff --git a/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonObjectMapperProvider.scala b/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonObjectMapperProvider.scala
index 567307f7fc..860eaffa63 100644
--- a/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonObjectMapperProvider.scala
+++ b/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonObjectMapperProvider.scala
@@ -23,23 +23,13 @@ import scala.util.{ Failure, Success }
import com.fasterxml.jackson.annotation.{ JsonAutoDetect, JsonCreator, PropertyAccessor }
import com.fasterxml.jackson.core.{
- JsonFactory,
- JsonFactoryBuilder,
- JsonGenerator,
- JsonParser,
- StreamReadConstraints,
- StreamReadFeature,
- StreamWriteConstraints,
- StreamWriteFeature
+ JsonFactory, JsonFactoryBuilder, JsonGenerator, JsonParser, StreamReadConstraints, StreamReadFeature,
+ StreamWriteConstraints, StreamWriteFeature
}
import com.fasterxml.jackson.core.json.{ JsonReadFeature, JsonWriteFeature }
import com.fasterxml.jackson.core.util.{ BufferRecycler, JsonRecyclerPools, RecyclerPool }
import com.fasterxml.jackson.databind.{
- DeserializationFeature,
- MapperFeature,
- Module,
- ObjectMapper,
- SerializationFeature
+ DeserializationFeature, MapperFeature, Module, ObjectMapper, SerializationFeature
}
import com.fasterxml.jackson.databind.cfg.EnumFeature
import com.fasterxml.jackson.databind.json.JsonMapper
@@ -47,12 +37,7 @@ import com.fasterxml.jackson.module.paramnames.ParameterNamesModule
import org.apache.pekko
import pekko.actor.{
- ActorSystem,
- ClassicActorSystemProvider,
- DynamicAccess,
- ExtendedActorSystem,
- Extension,
- ExtensionId,
+ ActorSystem, ClassicActorSystemProvider, DynamicAccess, ExtendedActorSystem, Extension, ExtensionId,
ExtensionIdProvider
}
import pekko.actor.setup.Setup
diff --git a/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonObjectMapperProvider.scala b/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonObjectMapperProvider.scala
index ae83185ba2..2897048b05 100644
--- a/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonObjectMapperProvider.scala
+++ b/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonObjectMapperProvider.scala
@@ -27,11 +27,7 @@ import tools.jackson.core.{ StreamReadConstraints, StreamReadFeature, StreamWrit
import tools.jackson.core.json.{ JsonFactory, JsonReadFeature, JsonWriteFeature }
import tools.jackson.core.util.{ BufferRecycler, JsonRecyclerPools, RecyclerPool }
import tools.jackson.databind.{
- DeserializationFeature,
- JacksonModule,
- MapperFeature,
- ObjectMapper,
- SerializationFeature
+ DeserializationFeature, JacksonModule, MapperFeature, ObjectMapper, SerializationFeature
}
import tools.jackson.databind.cfg.{ DateTimeFeature, EnumFeature, MapperBuilder }
import tools.jackson.databind.introspect.VisibilityChecker
@@ -40,12 +36,7 @@ import tools.jackson.dataformat.cbor.{ CBORFactory, CBORMapper }
import org.apache.pekko
import pekko.actor.{
- ActorSystem,
- ClassicActorSystemProvider,
- DynamicAccess,
- ExtendedActorSystem,
- Extension,
- ExtensionId,
+ ActorSystem, ClassicActorSystemProvider, DynamicAccess, ExtendedActorSystem, Extension, ExtensionId,
ExtensionIdProvider
}
import pekko.actor.setup.Setup
diff --git a/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala b/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
index d0a8084d9c..b1e1128c36 100644
--- a/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
+++ b/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
@@ -29,11 +29,7 @@ import com.fasterxml.jackson.annotation.{ JsonIgnore, JsonSubTypes, JsonTypeInfo
import tools.jackson.core.{ JsonGenerator, StreamReadFeature, StreamWriteFeature }
import tools.jackson.core.`type`.TypeReference
import tools.jackson.databind.{
- DeserializationFeature,
- JacksonModule,
- MapperFeature,
- SerializationContext,
- SerializationFeature
+ DeserializationFeature, JacksonModule, MapperFeature, SerializationContext, SerializationFeature
}
import tools.jackson.databind.annotation.{ JsonDeserialize, JsonSerialize }
import tools.jackson.databind.cfg.DateTimeFeature
diff --git a/stream-testkit/src/main/scala/org/apache/pekko/stream/testkit/StreamTestKit.scala b/stream-testkit/src/main/scala/org/apache/pekko/stream/testkit/StreamTestKit.scala
index 36abb43fc8..361dc9cd0d 100644
--- a/stream-testkit/src/main/scala/org/apache/pekko/stream/testkit/StreamTestKit.scala
+++ b/stream-testkit/src/main/scala/org/apache/pekko/stream/testkit/StreamTestKit.scala
@@ -25,11 +25,7 @@ import scala.reflect.ClassTag
import org.apache.pekko
import pekko.actor.{
- ActorRef,
- ActorSystem,
- ClassicActorSystemProvider,
- DeadLetterSuppression,
- NoSerializationVerificationNeeded
+ ActorRef, ActorSystem, ClassicActorSystemProvider, DeadLetterSuppression, NoSerializationVerificationNeeded
}
import pekko.japi._
import pekko.stream._
diff --git a/stream-testkit/src/test/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreterSpecKit.scala b/stream-testkit/src/test/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreterSpecKit.scala
index 46bf05c5ac..02a419d106 100644
--- a/stream-testkit/src/test/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreterSpecKit.scala
+++ b/stream-testkit/src/test/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreterSpecKit.scala
@@ -27,10 +27,7 @@ import pekko.event.Logging
import pekko.stream._
import pekko.stream.Supervision.Decider
import pekko.stream.impl.fusing.GraphInterpreter.{
- Connection,
- DownstreamBoundaryStageLogic,
- Failed,
- UpstreamBoundaryStageLogic
+ Connection, DownstreamBoundaryStageLogic, Failed, UpstreamBoundaryStageLogic
}
import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler, _ }
import pekko.stream.testkit.StreamSpec
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala
index b576641952..04b8bbe3fe 100755
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala
@@ -102,9 +102,11 @@ class DslConsistencySpec extends AnyWordSpec with Matchers {
jFlowClass -> (graphHelpers ++ forComprehensions),
jSourceClass -> (graphHelpers ++ forComprehensions ++ Set("watch", "ask")),
// Java subflows can only be nested using .via and .to (due to type system restrictions)
- jSubFlowClass -> (graphHelpers ++ forComprehensions ++ Set("groupBy", "splitAfter", "splitWhen", "subFlow", "watch",
+ jSubFlowClass ->
+ (graphHelpers ++ forComprehensions ++ Set("groupBy", "splitAfter", "splitWhen", "subFlow", "watch",
"ask")),
- jSubSourceClass -> (graphHelpers ++ forComprehensions ++ Set("groupBy", "splitAfter", "splitWhen", "subFlow",
+ jSubSourceClass ->
+ (graphHelpers ++ forComprehensions ++ Set("groupBy", "splitAfter", "splitWhen", "subFlow",
"watch", "ask")),
sFlowClass -> Set("of"),
sSourceClass -> Set("adapt", "from", "watch"),
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala
index ff480505ec..a2c30e7347 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala
@@ -188,7 +188,8 @@ class DslFactoriesConsistencySpec extends AnyWordSpec with Matchers {
ignores.foldLeft(false) {
case (acc, i) =>
- acc || (i.cls(m.declaringClass) && i.name(m.name) && i.parameters(m.parameterTypes.length) && i.paramTypes(
+ acc ||
+ (i.cls(m.declaringClass) && i.name(m.name) && i.parameters(m.parameterTypes.length) && i.paramTypes(
m.parameterTypes))
}
}
@@ -234,11 +235,13 @@ class DslFactoriesConsistencySpec extends AnyWordSpec with Matchers {
}
} else if (matches.length == 1) {
info(
- "Matched: Scala:" + row._1.name + "(" + row._1.parameterTypes
+ "Matched: Scala:" + row._1.name + "(" +
+ row._1.parameterTypes
.map(_.getName)
.mkString(",") + "): " + returnTypeString(row._1) +
" == " +
- "Java:" + matches.head.j.name + "(" + matches.head.j.parameterTypes
+ "Java:" + matches.head.j.name + "(" +
+ matches.head.j.parameterTypes
.map(_.getName)
.mkString(",") + "): " + returnTypeString(matches.head.j))
} else {
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/LinearTraversalBuilderSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/LinearTraversalBuilderSpec.scala
index cccb0868d9..8c2075972d 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/LinearTraversalBuilderSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/LinearTraversalBuilderSpec.scala
@@ -686,7 +686,8 @@ class LinearTraversalBuilderSpec extends PekkoSpec {
List(
sink -> (Attributes.name("test") and Attributes.name("sink")),
flow1 -> (Attributes.name("test") and Attributes.name("flow")),
- compositeSource -> (Attributes.name("test") and Attributes.name("source-outer") and Attributes.name(
+ compositeSource ->
+ (Attributes.name("test") and Attributes.name("source-outer") and Attributes.name(
"source"))))
}
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/InterpreterSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/InterpreterSpec.scala
index 1422b5fb7c..8ba3ced7e8 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/InterpreterSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/InterpreterSpec.scala
@@ -485,7 +485,8 @@ class InterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
lastEvents() should be(Set(OnNext(1), OnComplete))
}
- "work with pushAndFinish if upstream completes with pushAndFinish and downstream immediately pulls" in new OneBoundedSetup[
+ "work with pushAndFinish if upstream completes with pushAndFinish and downstream immediately pulls" in
+ new OneBoundedSetup[
Int](new PushFinishStage, Fold(0, (x: Int, y: Int) => x + y)) {
lastEvents() should be(Set.empty)
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/LifecycleInterpreterSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/LifecycleInterpreterSpec.scala
index deec0f31cc..6419fdc90c 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/LifecycleInterpreterSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/LifecycleInterpreterSpec.scala
@@ -127,7 +127,8 @@ class LifecycleInterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
expectMsg("stop")
}
- "postStop when pushAndFinish called with pushAndFinish if indirect upstream completes with pushAndFinish" in new OneBoundedSetup[
+ "postStop when pushAndFinish called with pushAndFinish if indirect upstream completes with pushAndFinish" in
+ new OneBoundedSetup[
String](Map((x: Any) => x), new PushFinishStage(onPostStop = () => testActor ! "stop"), Map((x: Any) => x)) {
lastEvents() should be(Set.empty)
@@ -140,7 +141,8 @@ class LifecycleInterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
expectMsg("stop")
}
- "postStop when pushAndFinish called with pushAndFinish if upstream completes with pushAndFinish and downstream immediately pulls" in new OneBoundedSetup[
+ "postStop when pushAndFinish called with pushAndFinish if upstream completes with pushAndFinish and downstream immediately pulls" in
+ new OneBoundedSetup[
String](new PushFinishStage(onPostStop = () => testActor ! "stop"), Fold("", (x: String, y: String) => x + y)) {
lastEvents() should be(Set.empty)
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/io/ByteStringParserSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/io/ByteStringParserSpec.scala
index 69a2eb9dfe..142e96e513 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/io/ByteStringParserSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/io/ByteStringParserSpec.scala
@@ -118,7 +118,8 @@ class ByteStringParserSpec extends StreamSpec {
(the[IllegalStateException] thrownBy Await.result(
Source.single(ByteString("abc")).via(SpinningLogic).runWith(Sink.ignore),
- 3.seconds)).getMessage shouldBe "Parsing logic didn't produce result after 10 steps. " +
+ 3.seconds)).getMessage shouldBe
+ "Parsing logic didn't produce result after 10 steps. " +
"Aborting processing to avoid infinite cycles. In the unlikely case that the parsing logic needs more recursion, " +
"override ParsingLogic.recursionLimit."
}
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/io/InputStreamSourceSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/io/InputStreamSourceSpec.scala
index fa6de2b7ed..67622ebdfd 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/io/InputStreamSourceSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/io/InputStreamSourceSpec.scala
@@ -22,11 +22,7 @@ import scala.util.Success
import org.apache.pekko
import pekko.Done
import pekko.stream.{
- AbruptStageTerminationException,
- ActorMaterializer,
- ActorMaterializerSettings,
- IOResult,
- Materializer
+ AbruptStageTerminationException, ActorMaterializer, ActorMaterializerSettings, IOResult, Materializer
}
import pekko.stream.scaladsl.{ Keep, Sink, StreamConverters }
import pekko.stream.testkit._
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/CancellationStrategySpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/CancellationStrategySpec.scala
index debbc352cb..05419c522c 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/CancellationStrategySpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/CancellationStrategySpec.scala
@@ -111,7 +111,8 @@ class CancellationStrategySpec extends StreamSpec("""pekko.loglevel = DEBUG
}
"cancellation races with BidiStacks" should {
- "accidentally convert errors to completions when CompleteStage strategy is chosen (2.5 default)" in new RaceTestSetup(
+ "accidentally convert errors to completions when CompleteStage strategy is chosen (2.5 default)" in
+ new RaceTestSetup(
CancellationStrategy.CompleteStage) {
val theError = TE("Duck meowed")
killSwitch.abort(theError)
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowFlatMapPrefixSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowFlatMapPrefixSpec.scala
index 6f6f25742e..ecf3818846 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowFlatMapPrefixSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowFlatMapPrefixSpec.scala
@@ -16,15 +16,8 @@ package org.apache.pekko.stream.scaladsl
import org.apache.pekko
import pekko.{ Done, NotUsed }
import pekko.stream.{
- AbruptStageTerminationException,
- AbruptTerminationException,
- Attributes,
- FlowShape,
- Inlet,
- Materializer,
- NeverMaterializedException,
- Outlet,
- SubscriptionWithCancelException
+ AbruptStageTerminationException, AbruptTerminationException, Attributes, FlowShape, Inlet, Materializer,
+ NeverMaterializedException, Outlet, SubscriptionWithCancelException
}
import pekko.stream.Attributes.Attribute
import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler }
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowGroupedWithinSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowGroupedWithinSpec.scala
index 579027256b..53092a3de8 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowGroupedWithinSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowGroupedWithinSpec.scala
@@ -235,7 +235,8 @@ class FlowGroupedWithinSpec extends StreamSpec with ScriptedTest {
downstream.expectComplete()
}
- "append zero weighted elements to a full group before timeout received, if downstream hasn't pulled yet" taggedAs TimingTest in {
+ "append zero weighted elements to a full group before timeout received, if downstream hasn't pulled yet" taggedAs
+ TimingTest in {
val upstream = TestPublisher.probe[String]()
val downstream = TestSubscriber.probe[immutable.Seq[String]]()
Source
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowIteratorSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowIteratorSpec.scala
index c310f80cb0..86158edbc4 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowIteratorSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowIteratorSpec.scala
@@ -33,8 +33,8 @@ class FlowIterableSpec extends AbstractFlowIteratorSpec {
"produce onError when iterator throws" in {
val iterable = new immutable.Iterable[Int] {
- override def iterator: Iterator[Int] =
- (1 to 3).iterator.map(x => if (x == 2) throw new IllegalStateException("not two") else x)
+ override def iterator: Iterator[Int] = (1 to 3).iterator.map(x =>
+ if (x == 2) throw new IllegalStateException("not two") else x)
}
val p = Source(iterable).runWith(Sink.asPublisher(false))
val c = TestSubscriber.manualProbe[Int]()
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowOrElseSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowOrElseSpec.scala
index f5a728f1f6..573b06d598 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowOrElseSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowOrElseSpec.scala
@@ -81,7 +81,8 @@ class FlowOrElseSpec extends PekkoSpec {
outProbe.expectComplete()
}
- "pass all available requested elements from input two when input 1 has completed without elements" in new OrElseProbedFlow {
+ "pass all available requested elements from input two when input 1 has completed without elements" in
+ new OrElseProbedFlow {
outProbe.request(5)
inProbe1.sendComplete()
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/RestartSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/RestartSpec.scala
index a1b2b373bc..5a55b91867 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/RestartSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/RestartSpec.scala
@@ -160,7 +160,8 @@ class RestartSpec
probe.cancel()
}
- "reset exponential backoff back to minimum when source runs for at least minimum backoff without completing" taggedAs TimingTest in {
+ "reset exponential backoff back to minimum when source runs for at least minimum backoff without completing" taggedAs
+ TimingTest in {
val created = new AtomicInteger()
val probe = RestartSource
.withBackoff(restartSettings) { () =>
@@ -464,7 +465,8 @@ class RestartSpec
probe.sendComplete()
}
- "reset exponential backoff back to minimum when sink runs for at least minimum backoff without completing" taggedAs TimingTest in {
+ "reset exponential backoff back to minimum when sink runs for at least minimum backoff without completing" taggedAs
+ TimingTest in {
val created = new AtomicInteger()
val (queue, sinkProbe) = TestSource[String]().toMat(TestSink())(Keep.both).run()
val probe = TestSource[String]()
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/RetryFlowSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/RetryFlowSpec.scala
index ac2d54e592..25b25720b3 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/RetryFlowSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/RetryFlowSpec.scala
@@ -513,7 +513,8 @@ class RetryFlowSpec extends StreamSpec("""
externalOut.expectError() shouldBe an[IllegalStateException]
}
- "allow more demand in inner flow (but never pass in more than one element into the retrying cycle)" in new AllSucceedBench[
+ "allow more demand in inner flow (but never pass in more than one element into the retrying cycle)" in
+ new AllSucceedBench[
InData, Ctx2, OutData] {
externalOut.request(1)
internalIn.expectRequest() shouldBe 1L
diff --git a/stream-typed-tests/src/test/scala/org/apache/pekko/stream/MapAsyncPartitionedSpec.scala b/stream-typed-tests/src/test/scala/org/apache/pekko/stream/MapAsyncPartitionedSpec.scala
index 765b335a78..05ef299b5a 100644
--- a/stream-typed-tests/src/test/scala/org/apache/pekko/stream/MapAsyncPartitionedSpec.scala
+++ b/stream-typed-tests/src/test/scala/org/apache/pekko/stream/MapAsyncPartitionedSpec.scala
@@ -32,16 +32,7 @@ import org.apache.pekko
import pekko.actor.typed.ActorSystem
import pekko.actor.typed.scaladsl.Behaviors
import pekko.stream.scaladsl.{
- Flow,
- FlowWithContext,
- GraphDSL,
- Keep,
- RunnableGraph,
- Sink,
- Source,
- SourceWithContext,
- Unzip,
- Zip
+ Flow, FlowWithContext, GraphDSL, Keep, RunnableGraph, Sink, Source, SourceWithContext, Unzip, Zip
}
import pekko.stream.testkit.scaladsl.TestSink
diff --git a/stream/src/main/scala/org/apache/pekko/stream/Shape.scala b/stream/src/main/scala/org/apache/pekko/stream/Shape.scala
index 3febf7fd7f..201b9a75f2 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/Shape.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/Shape.scala
@@ -167,7 +167,8 @@ final class Outlet[T] private (val s: String) extends OutPort {
* checks the size of 1 in an optimized way.
*/
def hasOnePort(ports: immutable.Seq[_]): Boolean = {
- ports.nonEmpty && (ports match {
+ ports.nonEmpty &&
+ (ports match {
case l: List[_] => l.tail.isEmpty // assuming List is most common
case _ => ports.size == 1 // e.g. Vector
})
diff --git a/stream/src/main/scala/org/apache/pekko/stream/SystemMaterializer.scala b/stream/src/main/scala/org/apache/pekko/stream/SystemMaterializer.scala
index a2e6d4a88d..b388de38bb 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/SystemMaterializer.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/SystemMaterializer.scala
@@ -75,8 +75,8 @@ final class SystemMaterializer(system: ExtendedActorSystem) extends Extension {
*/
@InternalApi
private[pekko] def createAdditionalSystemMaterializer(): Materializer = {
- val started =
- (materializerGuardian ? MaterializerGuardian.StartMaterializer).mapTo[MaterializerGuardian.MaterializerStarted]
+ val started = (materializerGuardian ? MaterializerGuardian.StartMaterializer).mapTo[
+ MaterializerGuardian.MaterializerStarted]
Await.result(started, materializerTimeout.duration).materializer
}
@@ -88,9 +88,8 @@ final class SystemMaterializer(system: ExtendedActorSystem) extends Extension {
private[pekko] def createAdditionalLegacySystemMaterializer(
namePrefix: String,
settings: ActorMaterializerSettings): Materializer = {
- val started =
- (materializerGuardian ? MaterializerGuardian.LegacyStartMaterializer(namePrefix, settings))
- .mapTo[MaterializerGuardian.MaterializerStarted]
+ val started = (materializerGuardian ? MaterializerGuardian.LegacyStartMaterializer(namePrefix, settings))
+ .mapTo[MaterializerGuardian.MaterializerStarted]
Await.result(started, materializerTimeout.duration).materializer
}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/FanIn.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/FanIn.scala
index acea3bac9a..32cc959efe 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/FanIn.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/FanIn.scala
@@ -78,8 +78,7 @@ import org.reactivestreams.{ Subscriber, Subscription }
private var receivedInput = false
private var completedCounter = 0
- private[this] final def hasState(index: Int, flag: Int): Boolean =
- (states(index) & flag) != 0
+ private[this] final def hasState(index: Int, flag: Int): Boolean = (states(index) & flag) != 0
private[this] final def setState(index: Int, flag: Int, on: Boolean): Unit =
states(index) = if (on) (states(index) | flag).toByte else (states(index) & ~flag).toByte
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/FutureFlow.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/FutureFlow.scala
index 0deea88007..914e05d13c 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/FutureFlow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/FutureFlow.scala
@@ -21,12 +21,7 @@ import scala.util.control.NonFatal
import org.apache.pekko
import pekko.annotation.InternalApi
import pekko.stream.{
- AbruptStageTerminationException,
- Attributes,
- FlowShape,
- Inlet,
- NeverMaterializedException,
- Outlet
+ AbruptStageTerminationException, Attributes, FlowShape, Inlet, NeverMaterializedException, Outlet
}
import pekko.stream.Attributes.SourceLocation
import pekko.stream.impl.Stages.DefaultAttributes
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala
index 352bd64fa7..6bcbb63dea 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala
@@ -37,20 +37,12 @@ import pekko.stream.Attributes.SourceLocation
import pekko.stream.OverflowStrategies._
import pekko.stream.Supervision.Decider
import pekko.stream.impl.{
- Buffer => BufferImpl,
- ContextPropagation,
- FailedSource,
- JavaStreamSource,
- ReactiveStreamsCompliance,
- TraversalBuilder
+ Buffer => BufferImpl, ContextPropagation, FailedSource, JavaStreamSource, ReactiveStreamsCompliance, TraversalBuilder
}
import pekko.stream.impl.Stages.DefaultAttributes
import pekko.stream.impl.fusing.GraphStages.{ FutureSource, SimpleLinearGraphStage, SingleSource }
import pekko.stream.scaladsl.{
- DelayStrategy,
- Source,
- StatefulMapConcatAccumulator,
- StatefulMapConcatAccumulatorFactory
+ DelayStrategy, Source, StatefulMapConcatAccumulator, StatefulMapConcatAccumulatorFactory
}
import pekko.stream.stage._
import pekko.util.{ ConstantFun, OptionVal }
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/SourceSink.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/SourceSink.scala
index 2a447293fa..3dbdd058f0 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/SourceSink.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/SourceSink.scala
@@ -25,11 +25,7 @@ import pekko.stream.{ ActorAttributes, Attributes, Inlet, SinkShape, StreamSubsc
import pekko.stream.ActorAttributes.StreamSubscriptionTimeout
import pekko.stream.scaladsl.Source
import pekko.stream.stage.{
- GraphStageLogic,
- GraphStageWithMaterializedValue,
- InHandler,
- OutHandler,
- TimerGraphStageLogic
+ GraphStageLogic, GraphStageWithMaterializedValue, InHandler, OutHandler, TimerGraphStageLogic
}
/**
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/FileOutputStage.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/FileOutputStage.scala
index fd665b51a6..eebcb01def 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/FileOutputStage.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/FileOutputStage.scala
@@ -25,12 +25,7 @@ import scala.util.control.NonFatal
import org.apache.pekko
import pekko.annotation.InternalApi
import pekko.stream.{
- AbruptStageTerminationException,
- Attributes,
- IOOperationIncompleteException,
- IOResult,
- Inlet,
- SinkShape
+ AbruptStageTerminationException, Attributes, IOOperationIncompleteException, IOResult, Inlet, SinkShape
}
import pekko.stream.impl.Stages.DefaultAttributes
import pekko.stream.stage.{ GraphStageLogic, GraphStageWithMaterializedValue, InHandler }
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSource.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSource.scala
index fd7c51c8aa..5fe0631ba0 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSource.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSource.scala
@@ -21,12 +21,7 @@ import scala.util.control.NonFatal
import org.apache.pekko
import pekko.annotation.InternalApi
import pekko.stream.{
- AbruptStageTerminationException,
- Attributes,
- IOOperationIncompleteException,
- IOResult,
- Outlet,
- SourceShape,
+ AbruptStageTerminationException, Attributes, IOOperationIncompleteException, IOResult, Outlet, SourceShape,
SubscriptionWithCancelException
}
import pekko.stream.impl.Stages.DefaultAttributes
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala
index a8332cc299..79c95e9383 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala
@@ -359,7 +359,8 @@ private[stream] object ConnectionSourceStage {
// round trip to the connection actor, or if reaching the configured maximum number of round trips, or
// if writeBuffer capacity has been exceeded.
writeDelayCountDown -= 1
- if (writeDelayCountDown == 0 || previousWriteBufferSize == writeBuffer.length || writeBuffer.length >= writeBufferSize)
+ if (writeDelayCountDown == 0 || previousWriteBufferSize == writeBuffer.length ||
+ writeBuffer.length >= writeBufferSize)
sendWriteBuffer()
else
sendWriteDelay()
diff --git a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
index 7f8972ac2d..e9a35a78b6 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
@@ -773,8 +773,7 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
/**
* Return true if the given output port is ready to be pushed.
*/
- final def isAvailable[T](out: Outlet[T]): Boolean =
- (conn(out).portState & (OutReady | OutClosed)) == OutReady
+ final def isAvailable[T](out: Outlet[T]): Boolean = (conn(out).portState & (OutReady | OutClosed)) == OutReady
/**
* Indicates whether the port has been closed. A closed port cannot be pushed.
diff --git a/testkit/src/main/scala/org/apache/pekko/testkit/CallingThreadDispatcher.scala b/testkit/src/main/scala/org/apache/pekko/testkit/CallingThreadDispatcher.scala
index 8a34ecb0d4..e43abc9370 100644
--- a/testkit/src/main/scala/org/apache/pekko/testkit/CallingThreadDispatcher.scala
+++ b/testkit/src/main/scala/org/apache/pekko/testkit/CallingThreadDispatcher.scala
@@ -24,24 +24,11 @@ import scala.util.control.NonFatal
import org.apache.pekko
import pekko.actor.{
- ActorCell,
- ActorInitializationException,
- ActorRef,
- ExtendedActorSystem,
- Extension,
- ExtensionId,
- ExtensionIdProvider
+ ActorCell, ActorInitializationException, ActorRef, ExtendedActorSystem, Extension, ExtensionId, ExtensionIdProvider
}
import pekko.dispatch.{
- DefaultSystemMessageQueue,
- DispatcherPrerequisites,
- Envelope,
- Mailbox,
- MailboxType,
- MessageDispatcher,
- MessageDispatcherConfigurator,
- MessageQueue,
- TaskInvocation
+ DefaultSystemMessageQueue, DispatcherPrerequisites, Envelope, Mailbox, MailboxType, MessageDispatcher,
+ MessageDispatcherConfigurator, MessageQueue, TaskInvocation
}
import pekko.dispatch.sysmsg.{ Resume, Suspend, SystemMessage }
import pekko.util.Switch
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org