From abaf7597d5eaba266e0e64977d3eee38ac588d35 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= Date: Tue, 16 Jun 2026 00:27:38 +0800 Subject: [PATCH] refactor: replace private[this] with private and = _ with explicit defaults for Scala 3.8 Motivation: Scala 3.8 deprecates private[this] in favor of private, and deprecates uninitialized vars (= _) in favor of explicit default values. These deprecations cause compilation warnings under Scala 3.8. Modification: - Replace private[this] and protected[this] with private and protected across ~94 files in actor, actor-typed, bench-jmh, cluster-sharding-typed, persistence, remote, and stream modules - Replace = _ initializers with explicit defaults (= null, = None, = 0, = 0L, = false, etc.) in affected files - No behavioral changes; these are purely mechanical deprecation fixes Result: Code compiles cleanly under Scala 3.8 without private[this] or uninitialized var deprecation warnings. No runtime behavior changes. Tests: Not run - mechanical refactoring, compilation test sufficient References: None - proactive Scala 3.8 forward compatibility --- .../typed/internal/ActorContextImpl.scala | 2 +- .../pekko/actor/typed/javadsl/Behaviors.scala | 4 +- .../actor/typed/scaladsl/AskPattern.scala | 2 +- .../scala/org/apache/pekko/actor/Actor.scala | 2 +- .../org/apache/pekko/actor/ActorCell.scala | 6 +-- .../org/apache/pekko/actor/ActorRef.scala | 4 +- .../apache/pekko/actor/ActorRefProvider.scala | 4 +- .../org/apache/pekko/actor/ActorSystem.scala | 8 ++-- .../scala/org/apache/pekko/actor/FSM.scala | 6 +-- .../pekko/actor/IndirectActorProducer.scala | 2 +- .../actor/LightArrayRevolverScheduler.scala | 6 +-- .../scala/org/apache/pekko/actor/Props.scala | 6 +-- .../pekko/actor/RepointableActorRef.scala | 10 ++--- .../pekko/dispatch/AbstractDispatcher.scala | 4 +- .../pekko/dispatch/BatchingExecutor.scala | 10 ++--- .../org/apache/pekko/dispatch/Mailbox.scala | 6 +-- .../dispatch/affinity/AffinityPool.scala | 28 ++++++------ .../pekko/dispatch/sysmsg/SystemMessage.scala | 2 +- .../org/apache/pekko/event/Logging.scala | 2 +- .../pekko/io/DirectByteBufferPool.scala | 4 +- .../apache/pekko/io/SelectionHandler.scala | 12 ++--- .../main/scala/org/apache/pekko/io/Tcp.scala | 4 +- .../org/apache/pekko/io/TcpConnection.scala | 14 +++--- .../main/scala/org/apache/pekko/io/Udp.scala | 2 +- .../org/apache/pekko/pattern/AskSupport.scala | 20 ++++----- .../apache/pekko/pattern/CircuitBreaker.scala | 12 ++--- .../pekko/serialization/Serialization.scala | 2 +- .../pekko/serialization/Serializer.scala | 4 +- .../org/apache/pekko/util/ByteString.scala | 6 +-- .../org/apache/pekko/util/Collections.scala | 4 +- .../org/apache/pekko/util/ConstantFun.scala | 2 +- .../apache/pekko/util/DoubleLinkedList.scala | 6 +-- .../apache/pekko/util/FrequencySketch.scala | 40 ++++++++--------- .../apache/pekko/util/ImmutableIntMap.scala | 6 +-- .../pekko/util/StablePriorityQueue.scala | 2 +- .../pekko/util/SubclassifiedIndex.scala | 2 +- .../org/apache/pekko/util/TokenBucket.scala | 4 +- .../pekko/stream/FusedGraphsBenchmark.scala | 4 +- .../pekko/stream/RangeSourceBenchmark.scala | 2 +- .../util/FastFrequencySketchBenchmark.scala | 10 ++--- .../pekko/util/FrequencySketchBenchmark.scala | 10 ++--- .../pekko/util/ImmutableIntMapBench.scala | 12 ++--- .../typed/internal/ClusterShardingImpl.scala | 2 +- .../journal/leveldb/LeveldbCompaction.scala | 2 +- .../journal/SteppingInmemJournal.scala | 2 +- .../aeron/AeronStreamMaxThroughputSpec.scala | 4 +- .../pekko/remote/artery/ArteryTransport.scala | 14 +++--- .../pekko/remote/artery/Association.scala | 20 ++++----- .../remote/artery/EnvelopeBufferPool.scala | 2 +- .../pekko/remote/artery/LruBoundedCache.scala | 12 ++--- .../aeron/ArteryAeronUdpTransport.scala | 10 ++--- .../remote/artery/aeron/TaskRunner.scala | 8 ++-- .../artery/compress/DecompressionTable.scala | 4 +- .../artery/compress/InboundCompressions.scala | 18 ++++---- .../artery/compress/TopHeavyHitters.scala | 10 ++--- .../transport/netty/NettyTransport.scala | 2 +- .../remote/artery/MetadataCarryingSpec.scala | 2 +- .../ZipLatestWithApply.scala.template | 2 +- .../org/apache/pekko/stream/KillSwitch.scala | 8 ++-- .../org/apache/pekko/stream/impl/FanIn.scala | 26 +++++------ .../pekko/stream/impl/JsonObjectParser.scala | 14 +++--- .../impl/PhasedFusingActorMaterializer.scala | 6 +-- .../impl/ResizableMultiReaderRingBuffer.scala | 8 ++-- .../stream/impl/SinkholeSubscriber.scala | 2 +- .../org/apache/pekko/stream/impl/Sinks.scala | 4 +- .../stream/impl/SubscriberManagement.scala | 10 ++--- .../pekko/stream/impl/TraversalBuilder.scala | 4 +- .../org/apache/pekko/stream/impl/Unfold.scala | 10 ++--- .../impl/fusing/ActorGraphInterpreter.scala | 2 +- .../impl/fusing/AggregateWithBoundary.scala | 2 +- .../stream/impl/fusing/GraphInterpreter.scala | 32 +++++++------- .../apache/pekko/stream/impl/fusing/Ops.scala | 10 ++--- .../stream/impl/fusing/RangeSource.scala | 10 ++--- .../stream/impl/fusing/StreamOfStreams.scala | 4 +- .../stream/impl/io/ByteStringParser.scala | 2 +- .../stream/impl/io/InputStreamSinkStage.scala | 10 ++--- .../impl/io/OutputStreamSourceStage.scala | 2 +- .../compression/DeflateDecompressorBase.scala | 2 +- .../stream/impl/streamref/SinkRefImpl.scala | 10 ++--- .../stream/impl/streamref/SourceRefImpl.scala | 16 +++---- .../impl/streamref/StreamRefsMaster.scala | 4 +- .../pekko/stream/javadsl/BidiFlow.scala | 2 +- .../pekko/stream/javadsl/DelayStrategy.scala | 2 +- .../apache/pekko/stream/javadsl/Graph.scala | 4 +- .../pekko/stream/scaladsl/BidiFlow.scala | 2 +- .../pekko/stream/scaladsl/DelayStrategy.scala | 2 +- .../apache/pekko/stream/scaladsl/Flow.scala | 2 +- .../apache/pekko/stream/scaladsl/Graph.scala | 4 +- .../apache/pekko/stream/scaladsl/Hub.scala | 44 +++++++++---------- .../apache/pekko/stream/scaladsl/Sink.scala | 4 +- .../apache/pekko/stream/scaladsl/Source.scala | 4 +- .../serialization/StreamRefSerializer.scala | 20 ++++----- .../pekko/stream/stage/GraphStage.scala | 4 +- .../pekko/stream/stage/StageLogging.scala | 2 +- 94 files changed, 353 insertions(+), 353 deletions(-) diff --git a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/ActorContextImpl.scala b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/ActorContextImpl.scala index 6f2e98b9627..7f3842dff36 100644 --- a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/ActorContextImpl.scala +++ b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/internal/ActorContextImpl.scala @@ -123,7 +123,7 @@ import org.slf4j.LoggerFactory timer } - protected[this] def mkTimer(): TimerSchedulerCrossDslSupport[T] = new TimerSchedulerImpl[T](this) + protected def mkTimer(): TimerSchedulerCrossDslSupport[T] = new TimerSchedulerImpl[T](this) override private[pekko] def hasTimer: Boolean = _timer.isDefined diff --git a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/javadsl/Behaviors.scala b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/javadsl/Behaviors.scala index 78a3d5314ee..bc3ee7b8a0c 100644 --- a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/javadsl/Behaviors.scala +++ b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/javadsl/Behaviors.scala @@ -31,10 +31,10 @@ import pekko.japi.pf.PFBuilder */ object Behaviors { - private[this] val _two2same = new JapiFunction2[ActorContext[Any], Any, Behavior[Any]] { + private val _two2same = new JapiFunction2[ActorContext[Any], Any, Behavior[Any]] { override def apply(context: ActorContext[Any], msg: Any): Behavior[Any] = same } - private[this] def two2same[T] = _two2same.asInstanceOf[JapiFunction2[ActorContext[T], T, Behavior[T]]] + private def two2same[T] = _two2same.asInstanceOf[JapiFunction2[ActorContext[T], T, Behavior[T]]] /** * `setup` is a factory for a behavior. Creation of the behavior instance is deferred until diff --git a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/scaladsl/AskPattern.scala b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/scaladsl/AskPattern.scala index 5ab409bccbe..0c811f764b5 100644 --- a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/scaladsl/AskPattern.scala +++ b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/scaladsl/AskPattern.scala @@ -144,7 +144,7 @@ object AskPattern { private final class PromiseRef[U](target: InternalRecipientRef[?], timeout: Timeout) { // Note: _promiseRef mustn't have a type pattern, since it can be null - private[this] val (_ref: ActorRef[U], _future: Future[U], _promiseRef) = + private val (_ref: ActorRef[U], _future: Future[U], _promiseRef) = if (target.isTerminated) ( adapt.ActorRefAdapter[U](target.provider.deadLetters), diff --git a/actor/src/main/scala/org/apache/pekko/actor/Actor.scala b/actor/src/main/scala/org/apache/pekko/actor/Actor.scala index 28e0b5e65a1..404c7e40e07 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/Actor.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/Actor.scala @@ -339,7 +339,7 @@ object Status { * }}} */ trait ActorLogging { this: Actor => - private var _log: LoggingAdapter = _ + private var _log: LoggingAdapter = null def log: LoggingAdapter = { // only used in Actor, i.e. thread safe diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala b/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala index 7f853edd8ef..7d36e058895 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/ActorCell.scala @@ -430,7 +430,7 @@ private[pekko] class ActorCell( with dungeon.DeathWatch with dungeon.FaultHandling { - private[this] var _props = _initialProps + private var _props = _initialProps def props: Props = _props import ActorCell._ @@ -445,11 +445,11 @@ private[pekko] class ActorCell( override final def classicActorContext: ActorContext = this protected def uid: Int = self.path.uid - private[this] var _actor: Actor = _ + private var _actor: Actor = _ def actor: Actor = _actor var currentMessage: Envelope = _ private var behaviorStack: List[Actor.Receive] = emptyBehaviorStack - private[this] var sysmsgStash: LatestFirstSystemMessageList = SystemMessageList.LNil + private var sysmsgStash: LatestFirstSystemMessageList = SystemMessageList.LNil // Java API final def getParent() = parent diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorRef.scala b/actor/src/main/scala/org/apache/pekko/actor/ActorRef.scala index abd9d06af65..9f347390097 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/ActorRef.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/ActorRef.scala @@ -886,8 +886,8 @@ private[pekko] class VirtualPathContainer( // AddressTerminatedTopic must be updated together with the variables here. // Important: don't include calls to sendSystemMessage inside the synchronized since that can // result in deadlock, see issue #26326 - private[this] var watching = ActorCell.emptyActorRefSet - private[this] var _watchedBy: OptionVal[Set[ActorRef]] = OptionVal.Some(ActorCell.emptyActorRefSet) + private var watching = ActorCell.emptyActorRefSet + private var _watchedBy: OptionVal[Set[ActorRef]] = OptionVal.Some(ActorCell.emptyActorRefSet) /** * INTERNAL API diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorRefProvider.scala b/actor/src/main/scala/org/apache/pekko/actor/ActorRefProvider.scala index 472adfbf4ad..8655e38bf88 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/ActorRefProvider.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/ActorRefProvider.scala @@ -418,7 +418,7 @@ private[pekko] class LocalActorRefProvider private[pekko] ( override val ignoreRef: ActorRef = new IgnoreActorRef(this) - private[this] final val terminationPromise: Promise[Terminated] = Promise[Terminated]() + private final val terminationPromise: Promise[Terminated] = Promise[Terminated]() def terminationFuture: Future[Terminated] = terminationPromise.future @@ -494,7 +494,7 @@ private[pekko] class LocalActorRefProvider private[pekko] ( * but it also requires these references to be @volatile and lazy. */ @volatile - private var system: ActorSystemImpl = _ + private var system: ActorSystemImpl = null @volatile private var extraNames: Map[String, InternalActorRef] = Map() 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 cc0ae959a9d..9c5accd751f 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala @@ -979,7 +979,7 @@ private[pekko] class ActorSystemImpl( dynamicAccess.createInstanceFor[LoggingFilter](LoggingFilter, arguments).get } - private[this] val markerLogging = + private val markerLogging = new MarkerLoggingAdapter(eventStream, getClass.getName + "(" + name + ")", this.getClass, logFilter) val log: LoggingAdapter = markerLogging @@ -1018,7 +1018,7 @@ private[pekko] class ActorSystemImpl( val dispatcher: ExecutionContextExecutor = dispatchers.defaultGlobalDispatcher - private[this] final val terminationCallbacks = new TerminationCallbacks(provider.terminationFuture)(dispatcher) + private final val terminationCallbacks = new TerminationCallbacks(provider.terminationFuture)(dispatcher) override def whenTerminated: Future[Terminated] = terminationCallbacks.terminationFuture override def getWhenTerminated: CompletionStage[Terminated] = whenTerminated.asJava @@ -1323,8 +1323,8 @@ private[pekko] class ActorSystemImpl( } final class TerminationCallbacks[T](upStreamTerminated: Future[T])(implicit ec: ExecutionContext) { - private[this] final val done = Promise[T]() - private[this] final val ref = new AtomicReference(done) + private final val done = Promise[T]() + private final val ref = new AtomicReference(done) // onComplete never fires twice so safe to avoid null check upStreamTerminated.onComplete { t => diff --git a/actor/src/main/scala/org/apache/pekko/actor/FSM.scala b/actor/src/main/scala/org/apache/pekko/actor/FSM.scala index 49aef7c78bd..fd93a273b7c 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/FSM.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/FSM.scala @@ -131,7 +131,7 @@ object FSM { private[pekko] final case class Timer(name: String, msg: Any, mode: TimerMode, generation: Int, owner: AnyRef)( context: ActorContext) extends NoSerializationVerificationNeeded { - private var ref: Option[Cancellable] = _ + private var ref: Option[Cancellable] = None private val scheduler = context.system.scheduler private implicit val executionContext: ExecutionContextExecutor = context.dispatcher @@ -733,9 +733,9 @@ trait FSM[S, D] extends Actor with Listeners with ActorLogging { /* * FSM State data and current timeout handling */ - private var currentState: State = _ + private var currentState: State = null private var timeoutFuture: Option[Cancellable] = None - private var nextState: State = _ + private var nextState: State = null private var generation: Long = 0L /* diff --git a/actor/src/main/scala/org/apache/pekko/actor/IndirectActorProducer.scala b/actor/src/main/scala/org/apache/pekko/actor/IndirectActorProducer.scala index fa6ec6f4a5f..39dec1dae9d 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/IndirectActorProducer.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/IndirectActorProducer.scala @@ -105,7 +105,7 @@ private[pekko] class TypedCreatorFunctionConsumer(clz: Class[? <: Actor], creato */ private[pekko] class ArgsReflectConstructor(clz: Class[? <: Actor], args: immutable.Seq[Any]) extends IndirectActorProducer { - private[this] val constructor = Reflect.findConstructor(clz, args) + private val constructor = Reflect.findConstructor(clz, args) override def actorClass = clz override def produce() = Reflect.instantiate(constructor, args) } 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 eb7ee9921d9..2d4e409501f 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala @@ -357,7 +357,7 @@ class LightArrayRevolverScheduler(config: Config, log: LoggingAdapter, threadFac } object LightArrayRevolverScheduler { - private[this] val taskHandle: VarHandle = { + private val taskHandle: VarHandle = { val lookup = MethodHandles.privateLookupIn(classOf[TaskHolder], MethodHandles.lookup()) lookup.findVarHandle(classOf[TaskHolder], "task", classOf[Runnable]) } @@ -417,8 +417,8 @@ object LightArrayRevolverScheduler { override def isCancelled: Boolean = task eq CancelledTask } - private[this] val CancelledTask = new Runnable { def run = () } - private[this] val ExecutedTask = new Runnable { def run = () } + private val CancelledTask = new Runnable { def run = () } + private val ExecutedTask = new Runnable { def run = () } private val NotCancellable: TimerTask = new TimerTask { def cancel(): Boolean = false diff --git a/actor/src/main/scala/org/apache/pekko/actor/Props.scala b/actor/src/main/scala/org/apache/pekko/actor/Props.scala index 69f2ef1f79c..2d8c4abf467 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/Props.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/Props.scala @@ -127,11 +127,11 @@ final case class Props(deploy: Deploy, clazz: Class[?], args: immutable.Seq[Any] // derived property, does not need to be serialized @transient - private[this] var _producer: IndirectActorProducer = _ + private var _producer: IndirectActorProducer = null // derived property, does not need to be serialized @transient - private[this] var _cachedActorClass: Class[? <: Actor] = _ + private var _cachedActorClass: Class[? <: Actor] = null /** * INTERNAL API @@ -143,7 +143,7 @@ final case class Props(deploy: Deploy, clazz: Class[?], args: immutable.Seq[Any] _producer } - private[this] def cachedActorClass: Class[? <: Actor] = { + private def cachedActorClass: Class[? <: Actor] = { if (_cachedActorClass eq null) _cachedActorClass = producer.actorClass diff --git a/actor/src/main/scala/org/apache/pekko/actor/RepointableActorRef.scala b/actor/src/main/scala/org/apache/pekko/actor/RepointableActorRef.scala index f9d300d9b73..0e9e562d812 100644 --- a/actor/src/main/scala/org/apache/pekko/actor/RepointableActorRef.scala +++ b/actor/src/main/scala/org/apache/pekko/actor/RepointableActorRef.scala @@ -212,13 +212,13 @@ private[pekko] class UnstartedCell( * This lock protects all accesses to this cell’s queues. It also ensures * safe switching to the started ActorCell. */ - private[this] final val lock = new ReentrantLock + private final val lock = new ReentrantLock // use Envelope to keep on-send checks in the same place ACCESS MUST BE PROTECTED BY THE LOCK - private[this] final val queue = new JLinkedList[Envelope]() + private final val queue = new JLinkedList[Envelope]() // ACCESS MUST BE PROTECTED BY THE LOCK - private[this] var sysmsgQueue: LatestFirstSystemMessageList = SystemMessageList.LNil + private var sysmsgQueue: LatestFirstSystemMessageList = SystemMessageList.LNil import systemImpl.settings.UnstartedPushTimeout.{ duration => timeout } @@ -305,7 +305,7 @@ private[pekko] class UnstartedCell( def isLocal = true - private[this] final def cellIsReady(cell: Cell): Boolean = (cell ne this) && (cell ne null) + private final def cellIsReady(cell: Cell): Boolean = (cell ne this) && (cell ne null) def hasMessages: Boolean = locked { val cell = self.underlying @@ -317,7 +317,7 @@ private[pekko] class UnstartedCell( if (cellIsReady(cell)) cell.numberOfMessages else queue.size } - private[this] final def locked[T](body: => T): T = { + private final def locked[T](body: => T): T = { lock.lock() try body finally lock.unlock() diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala b/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala index a0ed462f10b..f5e9e277dcc 100644 --- a/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala +++ b/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala @@ -116,8 +116,8 @@ abstract class MessageDispatcher(val configurator: MessageDispatcherConfigurator val mailboxes = prerequisites.mailboxes val eventStream = prerequisites.eventStream - @nowarn @volatile private[this] var _inhabitantsDoNotCallMeDirectly: Long = _ // DO NOT TOUCH! - @nowarn @volatile private[this] var _shutdownScheduleDoNotCallMeDirectly: Int = _ // DO NOT TOUCH! + @nowarn @volatile private var _inhabitantsDoNotCallMeDirectly: Long = _ // DO NOT TOUCH! + @nowarn @volatile private var _shutdownScheduleDoNotCallMeDirectly: Int = _ // DO NOT TOUCH! @nowarn private def _preventPrivateUnusedErasure = { _inhabitantsDoNotCallMeDirectly _shutdownScheduleDoNotCallMeDirectly diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/BatchingExecutor.scala b/actor/src/main/scala/org/apache/pekko/dispatch/BatchingExecutor.scala index d0f0ec0fa1f..93d278a00af 100644 --- a/actor/src/main/scala/org/apache/pekko/dispatch/BatchingExecutor.scala +++ b/actor/src/main/scala/org/apache/pekko/dispatch/BatchingExecutor.scala @@ -65,9 +65,9 @@ private[pekko] trait Batchable extends Runnable { private[pekko] trait BatchingExecutor extends Executor { // invariant: if "_tasksLocal.get ne null" then we are inside Batch.run; if it is null, we are outside - private[this] val _tasksLocal = new ThreadLocal[AbstractBatch]() + private val _tasksLocal = new ThreadLocal[AbstractBatch]() - private[this] abstract class AbstractBatch extends java.util.ArrayDeque[Runnable](4) with Runnable { + private abstract class AbstractBatch extends java.util.ArrayDeque[Runnable](4) with Runnable { @tailrec final def processBatch(batch: AbstractBatch): Unit = if ((batch eq this) && !batch.isEmpty) { batch.pollFirst().run() @@ -84,7 +84,7 @@ private[pekko] trait BatchingExecutor extends Executor { } } - private[this] final class Batch extends AbstractBatch { + private final class Batch extends AbstractBatch { override final def run(): Unit = { require(_tasksLocal.get eq null) _tasksLocal.set(this) // Install ourselves as the current batch @@ -97,9 +97,9 @@ private[pekko] trait BatchingExecutor extends Executor { } } - private[this] val _blockContext = new ThreadLocal[BlockContext]() + private val _blockContext = new ThreadLocal[BlockContext]() - private[this] final class BlockableBatch extends AbstractBatch with BlockContext { + private final class BlockableBatch extends AbstractBatch with BlockContext { // this method runs in the delegate ExecutionContext's thread override final def run(): Unit = { require(_tasksLocal.get eq null) diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala b/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala index 6dc048430e6..af728fabf93 100644 --- a/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala +++ b/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala @@ -88,7 +88,7 @@ private[pekko] abstract class Mailbox(val messageQueue: MessageQueue) * stay as it is. */ @volatile - var actor: ActorCell = _ + var actor: ActorCell = null def setActor(cell: ActorCell): Unit = actor = cell def dispatcher: MessageDispatcher = actor.dispatcher @@ -116,10 +116,10 @@ private[pekko] abstract class Mailbox(val messageQueue: MessageQueue) def numberOfMessages: Int = messageQueue.numberOfMessages @volatile - protected var _statusDoNotCallMeDirectly: Status = _ // 0 by default + protected var _statusDoNotCallMeDirectly: Status = 0 // 0 by default @volatile - protected var _systemQueueDoNotCallMeDirectly: SystemMessage = _ // null by default + protected var _systemQueueDoNotCallMeDirectly: SystemMessage = null // null by default // volatile read: the status is published across threads via compareAndSet/setVolatile // below; a plain VarHandle.get could observe a stale status (e.g. miss a Scheduled or diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/affinity/AffinityPool.scala b/actor/src/main/scala/org/apache/pekko/dispatch/affinity/AffinityPool.scala index 35d10b36ca0..4262c2ae614 100644 --- a/actor/src/main/scala/org/apache/pekko/dispatch/affinity/AffinityPool.scala +++ b/actor/src/main/scala/org/apache/pekko/dispatch/affinity/AffinityPool.scala @@ -62,17 +62,17 @@ private[affinity] object AffinityPool { // Following are auxiliary class and trait definitions private final class IdleStrategy(idleCpuLevel: Int) { - private[this] val maxSpins = 1100 * idleCpuLevel - 1000 - private[this] val maxYields = 5 * idleCpuLevel - private[this] val minParkPeriodNs = 1 - private[this] val maxParkPeriodNs = MICROSECONDS.toNanos(250 - ((80 * (idleCpuLevel - 1)) / 3)) + private val maxSpins = 1100 * idleCpuLevel - 1000 + private val maxYields = 5 * idleCpuLevel + private val minParkPeriodNs = 1 + private val maxParkPeriodNs = MICROSECONDS.toNanos(250 - ((80 * (idleCpuLevel - 1)) / 3)) - private[this] var state: IdleState = Initial - private[this] var turns = 0L - private[this] var parkPeriodNs = 0L - @volatile private[this] var idling = false + private var state: IdleState = Initial + private var turns = 0L + private var parkPeriodNs = 0L + @volatile private var idling = false - private[this] def transitionTo(newState: IdleState): Unit = { + private def transitionTo(newState: IdleState): Unit = { state = newState turns = 0 } @@ -148,8 +148,8 @@ private[pekko] class AffinityPool( // indicates the current state of the pool @volatile final private var poolState: PoolState = Uninitialized - private[this] final val workQueues = Array.fill(parallelism)(new BoundedAffinityTaskQueue(affinityGroupSize)) - private[this] final val workers = mutable.Set[AffinityPoolWorker]() + private final val workQueues = Array.fill(parallelism)(new BoundedAffinityTaskQueue(affinityGroupSize)) + private final val workers = mutable.Set[AffinityPoolWorker]() def start(): this.type = { bookKeepingLock.lock() @@ -263,7 +263,7 @@ private[pekko] class AffinityPool( override def toString: String = s"${Logging.simpleName(this)}(id = $id, parallelism = $parallelism, affinityGroupSize = $affinityGroupSize, threadFactory = $threadFactory, idleCpuLevel = $idleCpuLevel, queueSelector = $queueSelector, rejectionHandler = $rejectionHandler)" - private[this] final class AffinityPoolWorker(val q: BoundedAffinityTaskQueue, val idleStrategy: IdleStrategy) + private final class AffinityPoolWorker(val q: BoundedAffinityTaskQueue, val idleStrategy: IdleStrategy) extends Runnable { val thread: Thread = threadFactory.newThread(this) @@ -427,7 +427,7 @@ private[pekko] final class ThrowOnOverflowRejectionHandler extends RejectionHand private[pekko] final class FairDistributionHashCache(val config: Config) extends QueueSelectorFactory { private final val MaxFairDistributionThreshold = 2048 - private[this] final val fairDistributionThreshold = config + private final val fairDistributionThreshold = config .getInt("fair-work-distribution.threshold") .requiring( thr => 0 <= thr && thr <= MaxFairDistributionThreshold, @@ -437,7 +437,7 @@ private[pekko] final class FairDistributionHashCache(val config: Config) extends new AtomicReference[ImmutableIntMap](ImmutableIntMap.empty) with QueueSelector { override def toString: String = s"FairDistributionHashCache(fairDistributionThreshold = $fairDistributionThreshold)" - private[this] final def improve(h: Int): Int = + private final def improve(h: Int): Int = 0x7FFFFFFF & (reverseBytes(h * 0x9E3775CD) * 0x9E3775CD) // `sbhash`: In memory of Phil Bagwell. override final def getQueue(command: Runnable, queues: Int): Int = { val runnableHash = command.hashCode() diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/sysmsg/SystemMessage.scala b/actor/src/main/scala/org/apache/pekko/dispatch/sysmsg/SystemMessage.scala index 1e7b3e2ae13..803f81c6858 100644 --- a/actor/src/main/scala/org/apache/pekko/dispatch/sysmsg/SystemMessage.scala +++ b/actor/src/main/scala/org/apache/pekko/dispatch/sysmsg/SystemMessage.scala @@ -202,7 +202,7 @@ private[pekko] class EarliestFirstSystemMessageList(val head: SystemMessage) ext private[pekko] sealed trait SystemMessage extends PossiblyHarmful with Serializable { // Next fields are only modifiable via the SystemMessageList value class @transient - private[sysmsg] var next: SystemMessage = _ + private[sysmsg] var next: SystemMessage = null def unlink(): Unit = next = null diff --git a/actor/src/main/scala/org/apache/pekko/event/Logging.scala b/actor/src/main/scala/org/apache/pekko/event/Logging.scala index 7a1ef230638..2cfefc1f27e 100644 --- a/actor/src/main/scala/org/apache/pekko/event/Logging.scala +++ b/actor/src/main/scala/org/apache/pekko/event/Logging.scala @@ -51,7 +51,7 @@ trait LoggingBus extends ActorEventBus { private val guard = new ReentrantLock() private var loggers = Seq.empty[ActorRef] - @volatile private var _logLevel: LogLevel = _ + @volatile private var _logLevel: LogLevel = OffLevel /** * Query currently set log level. See object Logging for more information. diff --git a/actor/src/main/scala/org/apache/pekko/io/DirectByteBufferPool.scala b/actor/src/main/scala/org/apache/pekko/io/DirectByteBufferPool.scala index 50595a4d0af..6331b9ebec8 100644 --- a/actor/src/main/scala/org/apache/pekko/io/DirectByteBufferPool.scala +++ b/actor/src/main/scala/org/apache/pekko/io/DirectByteBufferPool.scala @@ -36,8 +36,8 @@ trait BufferPool { * benefit to wrapping in-heap JVM data when writing with NIO. */ private[pekko] class DirectByteBufferPool(defaultBufferSize: Int, maxPoolEntries: Int) extends BufferPool { - private[this] val pool: Array[ByteBuffer] = new Array[ByteBuffer](maxPoolEntries) - private[this] var buffersInPool: Int = 0 + private val pool: Array[ByteBuffer] = new Array[ByteBuffer](maxPoolEntries) + private var buffersInPool: Int = 0 def acquire(): ByteBuffer = takeBufferFromPool() diff --git a/actor/src/main/scala/org/apache/pekko/io/SelectionHandler.scala b/actor/src/main/scala/org/apache/pekko/io/SelectionHandler.scala index 338b823b87c..c6e2f869cf5 100644 --- a/actor/src/main/scala/org/apache/pekko/io/SelectionHandler.scala +++ b/actor/src/main/scala/org/apache/pekko/io/SelectionHandler.scala @@ -153,12 +153,12 @@ private[io] object SelectionHandler { settings: SelectionHandlerSettings, log: LoggingAdapter) extends ChannelRegistry { - private[this] val selector = SelectorProvider.provider.openSelector - private[this] val wakeUp = new AtomicBoolean(false) + private val selector = SelectorProvider.provider.openSelector + private val wakeUp = new AtomicBoolean(false) final val OP_READ_AND_WRITE = OP_READ | OP_WRITE // compile-time constant - private[this] val select = new Task { + private val select = new Task { def tryRun(): Unit = { if (selector.select(MaxSelectMillis) > 0) { // This assumes select return value == selectedKeys.size val keys = selector.selectedKeys @@ -317,9 +317,9 @@ private[io] class SelectionHandler(settings: SelectionHandlerSettings) import SelectionHandler._ import settings._ - private[this] var sequenceNumber = 0L // should be Long to prevent overflow - private[this] var childCount = 0 - private[this] val registry = { + private var sequenceNumber = 0L // should be Long to prevent overflow + private var childCount = 0 + private val registry = { val dispatcher = context.system.dispatchers.lookup(SelectorDispatcher) new ChannelRegistryImpl(SerializedSuspendableExecutionContext(dispatcher.throughput)(dispatcher), settings, log) } diff --git a/actor/src/main/scala/org/apache/pekko/io/Tcp.scala b/actor/src/main/scala/org/apache/pekko/io/Tcp.scala index 38ec52095fc..70d7ced59ce 100644 --- a/actor/src/main/scala/org/apache/pekko/io/Tcp.scala +++ b/actor/src/main/scala/org/apache/pekko/io/Tcp.scala @@ -395,7 +395,7 @@ object Tcp extends ExtensionId[TcpExt] with ExtensionIdProvider { def iterator: Iterator[SimpleWriteCommand] = new Iterator[SimpleWriteCommand] { - private[this] var current: WriteCommand = CompoundWrite.this + private var current: WriteCommand = CompoundWrite.this def hasNext: Boolean = current ne null def next(): SimpleWriteCommand = current match { @@ -632,7 +632,7 @@ class TcpExt(system: ExtendedActorSystem) extends IO.Extension { case _ => getBoolean("windows-connection-abort-workaround-enabled") } - private[this] def getIntBytes(path: String): Int = { + private def getIntBytes(path: String): Int = { val size = getBytes(path) require(size < Int.MaxValue, s"$path must be < 2 GiB") require(size >= 0, s"$path must be non-negative") diff --git a/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala b/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala index cc8f4947046..839dfcfff1f 100644 --- a/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala +++ b/actor/src/main/scala/org/apache/pekko/io/TcpConnection.scala @@ -48,12 +48,12 @@ private[io] abstract class TcpConnection(val tcp: TcpExt, val channel: SocketCha import tcp.Settings._ import tcp.bufferPool - private[this] var pendingWrite: PendingWrite = EmptyPendingWrite - private[this] var peerClosed = false - private[this] var writingSuspended = false - private[this] var readingSuspended = pullMode - private[this] var interestedInResume: Option[ActorRef] = None - private[this] var closedMessage: Option[CloseInformation] = None // for ConnectionClosed message in postStop + private var pendingWrite: PendingWrite = EmptyPendingWrite + private var peerClosed = false + private var writingSuspended = false + private var readingSuspended = pullMode + private var interestedInResume: Option[ActorRef] = None + private var closedMessage: Option[CloseInformation] = None // for ConnectionClosed message in postStop private var watchedActor: ActorRef = context.system.deadLetters private var registration: Option[ChannelRegistration] = None @@ -364,7 +364,7 @@ private[io] abstract class TcpConnection(val tcp: TcpExt, val channel: SocketCha case _: SocketException => false } - @tailrec private[this] def extractMsg(t: Throwable): String = + @tailrec private def extractMsg(t: Throwable): String = if (t eq null) "unknown" else { t.getMessage match { diff --git a/actor/src/main/scala/org/apache/pekko/io/Udp.scala b/actor/src/main/scala/org/apache/pekko/io/Udp.scala index f2cb936ee07..995f463f4f5 100644 --- a/actor/src/main/scala/org/apache/pekko/io/Udp.scala +++ b/actor/src/main/scala/org/apache/pekko/io/Udp.scala @@ -217,7 +217,7 @@ object Udp extends ExtensionId[UdpExt] with ExtensionIdProvider { override val MaxChannelsPerSelector: Int = if (MaxChannels == -1) -1 else math.max(MaxChannels / NrOfSelectors, 1) - private[this] def getIntBytes(path: String): Int = { + private def getIntBytes(path: String): Int = { val size = getBytes(path) require(size < Int.MaxValue, s"$path must be < 2 GiB") size.toInt diff --git a/actor/src/main/scala/org/apache/pekko/pattern/AskSupport.scala b/actor/src/main/scala/org/apache/pekko/pattern/AskSupport.scala index 6de334f1f41..8c85c4eff0f 100644 --- a/actor/src/main/scala/org/apache/pekko/pattern/AskSupport.scala +++ b/actor/src/main/scala/org/apache/pekko/pattern/AskSupport.scala @@ -536,11 +536,11 @@ private[pekko] final class PromiseActorRef( */ @volatile @nowarn("msg=is never updated") - private[this] var _stateDoNotCallMeDirectly: AnyRef = _ + private var _stateDoNotCallMeDirectly: AnyRef = null @volatile @nowarn("msg=is never updated") - private[this] var _watchedByDoNotCallMeDirectly: immutable.Set[ActorRef] = ActorCell.emptyActorRefSet + private var _watchedByDoNotCallMeDirectly: immutable.Set[ActorRef] = ActorCell.emptyActorRefSet @nowarn private def _preventPrivateUnusedErasure = { _stateDoNotCallMeDirectly @@ -548,38 +548,38 @@ private[pekko] final class PromiseActorRef( } // volatile read: published across threads via compareAndSet in updateWatchedBy - private[this] def watchedBy: Set[ActorRef] = watchedByHandle.getVolatile(this) + private def watchedBy: Set[ActorRef] = watchedByHandle.getVolatile(this) - private[this] def updateWatchedBy(oldWatchedBy: Set[ActorRef], newWatchedBy: Set[ActorRef]): Boolean = + private def updateWatchedBy(oldWatchedBy: Set[ActorRef], newWatchedBy: Set[ActorRef]): Boolean = watchedByHandle.compareAndSet(this, oldWatchedBy, newWatchedBy) @tailrec // Returns false if the Promise is already completed - private[this] final def addWatcher(watcher: ActorRef): Boolean = watchedBy match { + private final def addWatcher(watcher: ActorRef): Boolean = watchedBy match { case null => false case other => updateWatchedBy(other, other + watcher) || addWatcher(watcher) } @tailrec - private[this] final def remWatcher(watcher: ActorRef): Unit = watchedBy match { + private final def remWatcher(watcher: ActorRef): Unit = watchedBy match { case null => () case other => if (!updateWatchedBy(other, other - watcher)) remWatcher(watcher) } @tailrec - private[this] final def clearWatchers(): Set[ActorRef] = watchedBy match { + private final def clearWatchers(): Set[ActorRef] = watchedBy match { case null => ActorCell.emptyActorRefSet case other => if (!updateWatchedBy(other, null)) clearWatchers() else other } // volatile read: published across threads via compareAndSet/setVolatile below - private[this] def state: AnyRef = stateHandle.getVolatile(this) + private def state: AnyRef = stateHandle.getVolatile(this) - private[this] def updateState(oldState: AnyRef, newState: AnyRef): Boolean = + private def updateState(oldState: AnyRef, newState: AnyRef): Boolean = stateHandle.compareAndSet(this, oldState, newState) // volatile write: ordered against the concurrent reads in state; restores the // putObjectVolatile semantics this had before the VarHandle migration - private[this] def setState(newState: AnyRef): Unit = stateHandle.setVolatile(this, newState) + private def setState(newState: AnyRef): Unit = stateHandle.setVolatile(this, newState) override def getParent: InternalActorRef = provider.tempContainer 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 7b9f81f0df9..c64f76909d2 100644 --- a/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala +++ b/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala @@ -261,14 +261,14 @@ class CircuitBreaker( */ @nowarn("msg=is never updated") @volatile - private[this] var _currentStateDoNotCallMeDirectly: State = Closed + private var _currentStateDoNotCallMeDirectly: State = Closed /** * Holds reference to current resetTimeout of CircuitBreaker - *access only via helper methods* */ @nowarn("msg=is never updated") @volatile - private[this] var _currentResetTimeoutDoNotCallMeDirectly: FiniteDuration = resetTimeout + private var _currentResetTimeoutDoNotCallMeDirectly: FiniteDuration = resetTimeout @nowarn private def _preventPrivateUnusedErasure = { _currentStateDoNotCallMeDirectly @@ -282,7 +282,7 @@ class CircuitBreaker( * @param newState Next state on transition * @return Whether the previous state matched correctly */ - private[this] def swapState(oldState: State, newState: State): Boolean = + private def swapState(oldState: State, newState: State): Boolean = AbstractCircuitBreaker.stateHandle.compareAndSet(this, oldState, newState) /** @@ -290,7 +290,7 @@ class CircuitBreaker( * * @return Reference to current state */ - private[this] def currentState: State = + private def currentState: State = // volatile read: state is published across threads via compareAndSet in swapState; // restores the getObjectVolatile semantics this had before the VarHandle migration AbstractCircuitBreaker.stateHandle.getVolatile(this) @@ -298,13 +298,13 @@ class CircuitBreaker( /** * Helper method for updating the underlying resetTimeout via VarHandle */ - private[this] def swapResetTimeout(oldResetTimeout: FiniteDuration, newResetTimeout: FiniteDuration): Boolean = + private def swapResetTimeout(oldResetTimeout: FiniteDuration, newResetTimeout: FiniteDuration): Boolean = AbstractCircuitBreaker.resetTimeoutHandle.compareAndSet(this, oldResetTimeout, newResetTimeout) /** * Helper method for accessing to the underlying resetTimeout via VarHandle */ - private[this] def currentResetTimeout: FiniteDuration = + private def currentResetTimeout: FiniteDuration = // volatile read: see currentState; published via compareAndSet in swapResetTimeout AbstractCircuitBreaker.resetTimeoutHandle.getVolatile(this) diff --git a/actor/src/main/scala/org/apache/pekko/serialization/Serialization.scala b/actor/src/main/scala/org/apache/pekko/serialization/Serialization.scala index 6a0301af64c..9515ec1b238 100644 --- a/actor/src/main/scala/org/apache/pekko/serialization/Serialization.scala +++ b/actor/src/main/scala/org/apache/pekko/serialization/Serialization.scala @@ -150,7 +150,7 @@ class Serialization(val system: ExtendedActorSystem) extends Extension { val settings = new Settings(system.settings.config) val AllowJavaSerialization: Boolean = system.settings.AllowJavaSerialization - private[this] val _log = Logging.withMarker(system, getClass.getName) + private val _log = Logging.withMarker(system, getClass.getName) val log: LoggingAdapter = _log private val manifestCache = new AtomicReference[Map[String, Option[Class[?]]]](Map.empty[String, Option[Class[?]]]) diff --git a/actor/src/main/scala/org/apache/pekko/serialization/Serializer.scala b/actor/src/main/scala/org/apache/pekko/serialization/Serializer.scala index 01cc91c3814..39275ef6a7b 100644 --- a/actor/src/main/scala/org/apache/pekko/serialization/Serializer.scala +++ b/actor/src/main/scala/org/apache/pekko/serialization/Serializer.scala @@ -382,9 +382,9 @@ final case class DisabledJavaSerializer(system: ExtendedActorSystem) extends Ser // use same identifier as JavaSerializer, since it's a replacement override val identifier: Int = BaseSerializer.identifierFromConfig(classOf[JavaSerializer], system) - private[this] val empty = Array.empty[Byte] + private val empty = Array.empty[Byte] - private[this] val log = Logging.withMarker(system, classOf[DisabledJavaSerializer]) + private val log = Logging.withMarker(system, classOf[DisabledJavaSerializer]) def includeManifest: Boolean = false diff --git a/actor/src/main/scala/org/apache/pekko/util/ByteString.scala b/actor/src/main/scala/org/apache/pekko/util/ByteString.scala index 44db0e80c75..57c7762ef47 100644 --- a/actor/src/main/scala/org/apache/pekko/util/ByteString.scala +++ b/actor/src/main/scala/org/apache/pekko/util/ByteString.scala @@ -1083,7 +1083,7 @@ object ByteString { first.length + second.length == length, s"ByteString2 length ${first.length + second.length} did not match $length") - private[this] val firstLength: Int = first.length + private val firstLength: Int = first.length def apply(idx: Int): Byte = if (0 <= idx && idx < length) { @@ -1338,7 +1338,7 @@ object ByteString { ByteString1C(result) } - private[this] def byteAt(offset: Int, firstLength: Int): Byte = + private def byteAt(offset: Int, firstLength: Int): Byte = if (offset < firstLength) first.byteAtUnchecked(offset) else second.byteAtUnchecked(offset - firstLength) private[pekko] override def byteAtUnchecked(offset: Int): Byte = { @@ -2759,7 +2759,7 @@ final class ByteStringBuilder extends Builder[Byte, ByteString] { import ByteString.{ ByteString1, ByteString1C, ByteString2, ByteStrings } private var _length: Int = 0 private val _builder: VectorBuilder[ByteString1] = new VectorBuilder[ByteString1]() - private var _temp: Array[Byte] = _ + private var _temp: Array[Byte] = null private var _tempLength: Int = 0 private var _tempCapacity: Int = 0 diff --git a/actor/src/main/scala/org/apache/pekko/util/Collections.scala b/actor/src/main/scala/org/apache/pekko/util/Collections.scala index 12f9114959e..9311c3dadcd 100644 --- a/actor/src/main/scala/org/apache/pekko/util/Collections.scala +++ b/actor/src/main/scala/org/apache/pekko/util/Collections.scala @@ -58,8 +58,8 @@ private[pekko] object Collections { final def iterator: Iterator[To] = { val superIterator = valuesIterator new Iterator[To] { - private[this] var _next: To = _ - private[this] var _hasNext = false + private var _next: To = null.asInstanceOf[To] + private var _hasNext = false override final def hasNext: Boolean = { @tailrec def tailrecHasNext(): Boolean = { diff --git a/actor/src/main/scala/org/apache/pekko/util/ConstantFun.scala b/actor/src/main/scala/org/apache/pekko/util/ConstantFun.scala index 0a686df677f..8cf9f5d1860 100644 --- a/actor/src/main/scala/org/apache/pekko/util/ConstantFun.scala +++ b/actor/src/main/scala/org/apache/pekko/util/ConstantFun.scala @@ -25,7 +25,7 @@ import pekko.japi.function.{ Function => JFun, Function2 => JFun2 } */ @InternalApi private[pekko] object ConstantFun { - private[this] val JavaIdentityFunction = new JFun[Any, Any] { + private val JavaIdentityFunction = new JFun[Any, Any] { @throws(classOf[Exception]) override def apply(param: Any): Any = param } diff --git a/actor/src/main/scala/org/apache/pekko/util/DoubleLinkedList.scala b/actor/src/main/scala/org/apache/pekko/util/DoubleLinkedList.scala index bf2c79b22c1..cb56ba06b4a 100644 --- a/actor/src/main/scala/org/apache/pekko/util/DoubleLinkedList.scala +++ b/actor/src/main/scala/org/apache/pekko/util/DoubleLinkedList.scala @@ -29,8 +29,8 @@ private[pekko] final class DoubleLinkedList[Node]( setPrevious: (Node, OptionVal[Node]) => Unit, setNext: (Node, OptionVal[Node]) => Unit) { - private[this] var first: OptionVal[Node] = OptionVal.none - private[this] var last: OptionVal[Node] = OptionVal.none + private var first: OptionVal[Node] = OptionVal.none + private var last: OptionVal[Node] = OptionVal.none def isEmpty: Boolean = first.isEmpty @@ -138,7 +138,7 @@ private[pekko] final class DoubleLinkedList[Node]( private def iteratorFrom(start: OptionVal[Node], shift: Node => OptionVal[Node]): Iterator[Node] = new AbstractIterator[Node] { - private[this] var cursor: OptionVal[Node] = start + private var cursor: OptionVal[Node] = start override def hasNext: Boolean = cursor.isDefined override def next(): Node = { val node = cursor 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 f7c73393415..4f56d77811a 100644 --- a/actor/src/main/scala/org/apache/pekko/util/FrequencySketch.scala +++ b/actor/src/main/scala/org/apache/pekko/util/FrequencySketch.scala @@ -138,25 +138,25 @@ private[pekko] final class FrequencySketch[A]( private final val SlotBits = 64 - private[this] val counterWidth = counterBits - private[this] val slots = SlotBits / counterWidth - private[this] val rowWidth = math.max(1, width / slots) - private[this] val columnMask = width - 1 - private[this] val slotShift = FrequencySketch.Bits.powerOfTwoExponent(slots) - private[this] val slotMask = slots - 1 - private[this] val counterShift = FrequencySketch.Bits.powerOfTwoExponent(counterWidth) - private[this] val counterMask = if (counterBits == 64) Long.MaxValue else (1L << counterWidth) - 1 - - private[this] val oddMask = (1 to slots).foldLeft(1L)((mask, count) => mask | (1L << (count * counterWidth))) - - private[this] val resetMask = { + private val counterWidth = counterBits + private val slots = SlotBits / counterWidth + private val rowWidth = math.max(1, width / slots) + private val columnMask = width - 1 + private val slotShift = FrequencySketch.Bits.powerOfTwoExponent(slots) + private val slotMask = slots - 1 + private val counterShift = FrequencySketch.Bits.powerOfTwoExponent(counterWidth) + private val counterMask = if (counterBits == 64) Long.MaxValue else (1L << counterWidth) - 1 + + private val oddMask = (1 to slots).foldLeft(1L)((mask, count) => mask | (1L << (count * counterWidth))) + + private val resetMask = { val counterResetMask = counterMask >> 1 (1 to slots).foldLeft(counterResetMask)((mask, count) => mask | (counterResetMask << (count * counterWidth))) } - private[this] val matrix = Array.fill[Array[Long]](depth)(Array.ofDim[Long](rowWidth)) - private[this] val rowSizes = Array.ofDim[Int](depth) - private[this] var updatedSize = 0 + private val matrix = Array.fill[Array[Long]](depth)(Array.ofDim[Long](rowWidth)) + private val rowSizes = Array.ofDim[Int](depth) + private var updatedSize = 0 /** * Get the current size of the sketch (the number of incremented counters). @@ -300,12 +300,12 @@ private[pekko] final class FastFrequencySketch[A](width: Int, resetSize: Int) { private final val Seed2 = 0x9AE16A3B2F90404FL private final val Seed3 = 0xCBF29CE484222325L - private[this] val rowWidth = math.max(1, width >>> SlotShift) - private[this] val indexMask = width - 1 + private val rowWidth = math.max(1, width >>> SlotShift) + private val indexMask = width - 1 - private[this] val matrix = Array.fill[Array[Long]](Depth)(Array.ofDim[Long](rowWidth)) - private[this] val rowSizes = Array.ofDim[Int](Depth) - private[this] var updatedSize = 0 + private val matrix = Array.fill[Array[Long]](Depth)(Array.ofDim[Long](rowWidth)) + private val rowSizes = Array.ofDim[Int](Depth) + private var updatedSize = 0 def size: Int = updatedSize diff --git a/actor/src/main/scala/org/apache/pekko/util/ImmutableIntMap.scala b/actor/src/main/scala/org/apache/pekko/util/ImmutableIntMap.scala index d53b72296a5..8405e0fa075 100644 --- a/actor/src/main/scala/org/apache/pekko/util/ImmutableIntMap.scala +++ b/actor/src/main/scala/org/apache/pekko/util/ImmutableIntMap.scala @@ -41,7 +41,7 @@ import org.apache.pekko.annotation.InternalApi kvs(1) = value } - private[this] final def indexForKey(key: Int): Int = { + private final def indexForKey(key: Int): Int = { // Custom implementation of binary search since we encode key + value in consecutive indices. // We do the binary search on half the size of the array then project to the full size. // >>> 1 for division by 2: https://research.googleblog.com/2006/06/extra-extra-read-all-about-it-nearly.html @@ -107,13 +107,13 @@ import org.apache.pekko.annotation.InternalApi } else insert(key, value, i) } else new ImmutableIntMap(key, value) - private[this] final def update(value: Int, valueIndex: Int): ImmutableIntMap = { + private final def update(value: Int, valueIndex: Int): ImmutableIntMap = { val newKvs = kvs.clone() // clone() can in theory be faster since it could do a malloc + memcpy iso. calloc etc newKvs(valueIndex) = value new ImmutableIntMap(newKvs, size) } - private[this] final def insert(key: Int, value: Int, index: Int): ImmutableIntMap = { + private final def insert(key: Int, value: Int, index: Int): ImmutableIntMap = { val at = ~index // ~n == -(n + 1): insert the entry at the right position—keep the array sorted val newKvs = new Array[Int](kvs.length + 2) System.arraycopy(kvs, 0, newKvs, 0, at) diff --git a/actor/src/main/scala/org/apache/pekko/util/StablePriorityQueue.scala b/actor/src/main/scala/org/apache/pekko/util/StablePriorityQueue.scala index 594877e9e51..4bb863aa8de 100644 --- a/actor/src/main/scala/org/apache/pekko/util/StablePriorityQueue.scala +++ b/actor/src/main/scala/org/apache/pekko/util/StablePriorityQueue.scala @@ -38,7 +38,7 @@ trait PriorityQueueStabilizer[E <: AnyRef] extends AbstractQueue[E] { } override def iterator(): Iterator[E] = new Iterator[E] { - private[this] val backingIterator = backingQueue.iterator() + private val backingIterator = backingQueue.iterator() def hasNext: Boolean = backingIterator.hasNext def next(): E = backingIterator.next().element override def remove() = backingIterator.remove() diff --git a/actor/src/main/scala/org/apache/pekko/util/SubclassifiedIndex.scala b/actor/src/main/scala/org/apache/pekko/util/SubclassifiedIndex.scala index 256ff53c35c..477f386d1f0 100644 --- a/actor/src/main/scala/org/apache/pekko/util/SubclassifiedIndex.scala +++ b/actor/src/main/scala/org/apache/pekko/util/SubclassifiedIndex.scala @@ -71,7 +71,7 @@ private[pekko] object SubclassifiedIndex { } private[SubclassifiedIndex] def emptyMergeMap[K, V] = internalEmptyMergeMap.asInstanceOf[Map[K, Set[V]]] - private[this] val internalEmptyMergeMap = Map[AnyRef, Set[AnyRef]]().withDefaultValue(Set[AnyRef]()) + private val internalEmptyMergeMap = Map[AnyRef, Set[AnyRef]]().withDefaultValue(Set[AnyRef]()) } /** diff --git a/actor/src/main/scala/org/apache/pekko/util/TokenBucket.scala b/actor/src/main/scala/org/apache/pekko/util/TokenBucket.scala index a07a2866c85..e7fb53d98d3 100644 --- a/actor/src/main/scala/org/apache/pekko/util/TokenBucket.scala +++ b/actor/src/main/scala/org/apache/pekko/util/TokenBucket.scala @@ -20,8 +20,8 @@ private[pekko] abstract class TokenBucket(capacity: Long, nanosBetweenTokens: Lo require(capacity >= 0, "Capacity must be non-negative.") require(nanosBetweenTokens > 0, "Time between tokens must be larger than zero nanoseconds.") - private[this] var availableTokens: Long = _ - private[this] var lastUpdate: Long = _ + private var availableTokens: Long = 0L + private var lastUpdate: Long = 0L /** * This method must be called before the token bucket can be used. diff --git a/bench-jmh/src/main/scala/org/apache/pekko/stream/FusedGraphsBenchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/stream/FusedGraphsBenchmark.scala index fd7425525f6..5fe66760606 100644 --- a/bench-jmh/src/main/scala/org/apache/pekko/stream/FusedGraphsBenchmark.scala +++ b/bench-jmh/src/main/scala/org/apache/pekko/stream/FusedGraphsBenchmark.scala @@ -46,7 +46,7 @@ class TestSource(elems: Array[MutableElement]) extends GraphStage[SourceShape[Mu override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with OutHandler { - private[this] var left = FusedGraphsBenchmark.ElementCount - 1 + private var left = FusedGraphsBenchmark.ElementCount - 1 override def onPull(): Unit = { if (left >= 0) { @@ -66,7 +66,7 @@ class JitSafeCompletionLatch extends GraphStageWithMaterializedValue[SinkShape[M override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, CountDownLatch) = { val latch = new CountDownLatch(1) val logic = new GraphStageLogic(shape) with InHandler { - private[this] var sum = 0 + private var sum = 0 override def preStart(): Unit = pull(in) override def onPush(): Unit = { diff --git a/bench-jmh/src/main/scala/org/apache/pekko/stream/RangeSourceBenchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/stream/RangeSourceBenchmark.scala index 59925f9612d..eb53966012b 100644 --- a/bench-jmh/src/main/scala/org/apache/pekko/stream/RangeSourceBenchmark.scala +++ b/bench-jmh/src/main/scala/org/apache/pekko/stream/RangeSourceBenchmark.scala @@ -42,7 +42,7 @@ final class IntCompletionLatch extends GraphStageWithMaterializedValue[SinkShape override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, CountDownLatch) = { val latch = new CountDownLatch(1) val logic = new GraphStageLogic(shape) with InHandler { - private[this] var sum = 0 + private var sum = 0 override def preStart(): Unit = pull(in) override def onPush(): Unit = { diff --git a/bench-jmh/src/main/scala/org/apache/pekko/util/FastFrequencySketchBenchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/util/FastFrequencySketchBenchmark.scala index 3dd27b0a6ca..38f6ed4d730 100644 --- a/bench-jmh/src/main/scala/org/apache/pekko/util/FastFrequencySketchBenchmark.scala +++ b/bench-jmh/src/main/scala/org/apache/pekko/util/FastFrequencySketchBenchmark.scala @@ -28,13 +28,13 @@ import org.openjdk.jmh.annotations.Warmup @Warmup(iterations = 3, time = 20, timeUnit = TimeUnit.SECONDS) @Measurement(iterations = 3, time = 10, timeUnit = TimeUnit.SECONDS) class FastFrequencySketchBenchmark { - private[this] val Capacity = 10000 - private[this] val GeneratedSize = 1 << 16 + private val Capacity = 10000 + private val GeneratedSize = 1 << 16 private final val IndexMask = 0xFFFF - private[this] var sketch: FastFrequencySketch[String] = _ - private[this] var generated: Array[String] = _ - private[this] var index: Int = 0 + private var sketch: FastFrequencySketch[String] = _ + private var generated: Array[String] = _ + private var index: Int = 0 @Setup def setup(): Unit = { diff --git a/bench-jmh/src/main/scala/org/apache/pekko/util/FrequencySketchBenchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/util/FrequencySketchBenchmark.scala index a92b0defdd7..f89bdd9b181 100644 --- a/bench-jmh/src/main/scala/org/apache/pekko/util/FrequencySketchBenchmark.scala +++ b/bench-jmh/src/main/scala/org/apache/pekko/util/FrequencySketchBenchmark.scala @@ -28,13 +28,13 @@ import org.openjdk.jmh.annotations.Warmup @Warmup(iterations = 3, time = 20, timeUnit = TimeUnit.SECONDS) @Measurement(iterations = 3, time = 10, timeUnit = TimeUnit.SECONDS) class FrequencySketchBenchmark { - private[this] val Capacity = 10000 - private[this] val GeneratedSize = 1 << 16 + private val Capacity = 10000 + private val GeneratedSize = 1 << 16 private final val IndexMask = 0xFFFF - private[this] var sketch: FrequencySketch[String] = _ - private[this] var generated: Array[String] = _ - private[this] var index: Int = 0 + private var sketch: FrequencySketch[String] = _ + private var generated: Array[String] = _ + private var index: Int = 0 @Setup def setup(): Unit = { diff --git a/bench-jmh/src/main/scala/org/apache/pekko/util/ImmutableIntMapBench.scala b/bench-jmh/src/main/scala/org/apache/pekko/util/ImmutableIntMapBench.scala index 6e385202fe3..fc3779f6b36 100644 --- a/bench-jmh/src/main/scala/org/apache/pekko/util/ImmutableIntMapBench.scala +++ b/bench-jmh/src/main/scala/org/apache/pekko/util/ImmutableIntMapBench.scala @@ -27,33 +27,33 @@ import org.openjdk.jmh.annotations._ @Measurement(iterations = 10, time = 15, timeUnit = TimeUnit.MICROSECONDS, batchSize = 1) class ImmutableIntMapBench { - @tailrec private[this] final def add(n: Int, c: ImmutableIntMap = ImmutableIntMap.empty): ImmutableIntMap = + @tailrec private final def add(n: Int, c: ImmutableIntMap = ImmutableIntMap.empty): ImmutableIntMap = if (n >= 0) add(n - 1, c.updated(n, n)) else c - @tailrec private[this] final def contains(n: Int, by: Int, to: Int, in: ImmutableIntMap, b: Boolean): Boolean = + @tailrec private final def contains(n: Int, by: Int, to: Int, in: ImmutableIntMap, b: Boolean): Boolean = if (n <= to) { val result = in.contains(n) contains(n + by, by, to, in, result) } else b - @tailrec private[this] final def get(n: Int, by: Int, to: Int, in: ImmutableIntMap, b: Int): Int = + @tailrec private final def get(n: Int, by: Int, to: Int, in: ImmutableIntMap, b: Int): Int = if (n <= to) { val result = in.get(n) get(n + by, by, to, in, result) } else b - @tailrec private[this] final def hashCode(n: Int, in: ImmutableIntMap, b: Int): Int = + @tailrec private final def hashCode(n: Int, in: ImmutableIntMap, b: Int): Int = if (n >= 0) { val result = in.hashCode hashCode(n - 1, in, result) } else b - @tailrec private[this] final def updateIfAbsent(n: Int, by: Int, to: Int, in: ImmutableIntMap): ImmutableIntMap = + @tailrec private final def updateIfAbsent(n: Int, by: Int, to: Int, in: ImmutableIntMap): ImmutableIntMap = if (n <= to) updateIfAbsent(n + by, by, to, in.updateIfAbsent(n, n)) else in - @tailrec private[this] final def getKey(iterations: Int, key: Int, from: ImmutableIntMap): ImmutableIntMap = { + @tailrec private final def getKey(iterations: Int, key: Int, from: ImmutableIntMap): ImmutableIntMap = { if (iterations > 0 && key != Int.MinValue) { val k = from.get(key) getKey(iterations - 1, k, from) diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala index 98406919d64..df01f21032c 100644 --- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala +++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala @@ -361,7 +361,7 @@ import pekko.util.{ ByteString, Timeout } import pekko.actor.typed.internal.{ adapter => adapt } // Note: _promiseRef mustn't have a type pattern, since it can be null - private[this] val (_ref: ActorRef[U], _future: Future[U], _promiseRef) = + private val (_ref: ActorRef[U], _future: Future[U], _promiseRef) = if (classic.isTerminated) ( adapt.ActorRefAdapter[U](classic.provider.deadLetters), diff --git a/persistence/src/main/scala/org/apache/pekko/persistence/journal/leveldb/LeveldbCompaction.scala b/persistence/src/main/scala/org/apache/pekko/persistence/journal/leveldb/LeveldbCompaction.scala index b21f9205440..46cd1789ff5 100644 --- a/persistence/src/main/scala/org/apache/pekko/persistence/journal/leveldb/LeveldbCompaction.scala +++ b/persistence/src/main/scala/org/apache/pekko/persistence/journal/leveldb/LeveldbCompaction.scala @@ -71,7 +71,7 @@ private[persistence] trait CompactionSegmentManagement { import CompactionSegmentManagement._ - private[this] var latestCompactionSegments = Map.empty[String, Long] + private var latestCompactionSegments = Map.empty[String, Long] def compactionIntervals: Map[String, Long] diff --git a/persistence/src/test/scala/org/apache/pekko/persistence/journal/SteppingInmemJournal.scala b/persistence/src/test/scala/org/apache/pekko/persistence/journal/SteppingInmemJournal.scala index 9ff0473f8e5..0a228b5807c 100644 --- a/persistence/src/test/scala/org/apache/pekko/persistence/journal/SteppingInmemJournal.scala +++ b/persistence/src/test/scala/org/apache/pekko/persistence/journal/SteppingInmemJournal.scala @@ -51,7 +51,7 @@ object SteppingInmemJournal { // keep it in a thread safe:d global so that tests can get their // hand on the actor ref and send Steps to it - private[this] var _current: Map[String, ActorRef] = Map() + private var _current: Map[String, ActorRef] = Map() // shhh don't tell anyone I sinn-croniz-ed /** get the actor ref to the journal for a given instance id, throws exception if not found */ diff --git a/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/artery/aeron/AeronStreamMaxThroughputSpec.scala b/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/artery/aeron/AeronStreamMaxThroughputSpec.scala index 9e78a90f3cb..6c32947673d 100644 --- a/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/artery/aeron/AeronStreamMaxThroughputSpec.scala +++ b/remote-tests/src/multi-jvm/scala/org/apache/pekko/remote/artery/aeron/AeronStreamMaxThroughputSpec.scala @@ -59,8 +59,8 @@ object AeronStreamMaxThroughputSpec extends MultiNodeConfig { final case class TestSettings(testName: String, totalMessages: Long, payloadSize: Int) def iterate(start: Long, end: Long): Iterator[Long] = new AbstractIterator[Long] { - private[this] var first = true - private[this] var acc = start + private var first = true + private var acc = start def hasNext: Boolean = acc < end def next(): Long = { if (!hasNext) throw new NoSuchElementException("next on empty iterator") 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 b9bcfd45ef7..ba3f8e650b3 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 @@ -285,13 +285,13 @@ private[remote] abstract class ArteryTransport(_system: ExtendedActorSystem, _pr type LifeCycle // these vars are initialized once in the start method - @volatile private[this] var _localAddress: UniqueAddress = _ - @volatile private[this] var _bindAddress: UniqueAddress = _ - @volatile private[this] var _addresses: Set[Address] = _ + @volatile private var _localAddress: UniqueAddress = _ + @volatile private var _bindAddress: UniqueAddress = _ + @volatile private var _addresses: Set[Address] = _ @volatile protected var materializer: Materializer = _ @volatile protected var controlMaterializer: Materializer = _ - @volatile private[this] var controlSubject: ControlMessageSubject = _ - @volatile private[this] var messageDispatcher: MessageDispatcher = _ + @volatile private var controlSubject: ControlMessageSubject = _ + @volatile private var messageDispatcher: MessageDispatcher = _ override val log: MarkerLoggingAdapter = Logging.withMarker(system, classOf[ArteryTransport]) @@ -310,7 +310,7 @@ private[remote] abstract class ArteryTransport(_system: ExtendedActorSystem, _pr } else NoInboundCompressions } - @volatile private[this] var _inboundCompressionAccess: OptionVal[InboundCompressionAccess] = OptionVal.None + @volatile private var _inboundCompressionAccess: OptionVal[InboundCompressionAccess] = OptionVal.None /** Only access compression tables via the CompressionAccess */ def inboundCompressionAccess: OptionVal[InboundCompressionAccess] = _inboundCompressionAccess @@ -327,7 +327,7 @@ private[remote] abstract class ArteryTransport(_system: ExtendedActorSystem, _pr // keyed by the streamId protected val streamMatValues = new AtomicReference(Map.empty[Int, InboundStreamMatValues[LifeCycle]]) - private[this] val hasBeenShutdown = new AtomicBoolean(false) + private val hasBeenShutdown = new AtomicBoolean(false) private val testState = new SharedTestState 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 d2e543414d2..eb9bf1176a5 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 @@ -181,7 +181,7 @@ private[remote] class Association( private val queueSize = advancedSettings.OutboundMessageQueueSize private val largeQueueSize = advancedSettings.OutboundLargeMessageQueueSize - private[this] val queues: Array[SendQueue.ProducerApi[OutboundEnvelope]] = new Array(2 + outboundLanes) + private val queues: Array[SendQueue.ProducerApi[OutboundEnvelope]] = new Array(2 + outboundLanes) queues(ControlQueueIndex) = QueueWrapperImpl(createQueue(controlQueueSize, ControlQueueIndex)) // control stream queues(LargeQueueIndex) = if (transport.largeMessageChannelEnabled) // large messages stream @@ -192,19 +192,19 @@ private[remote] class Association( (0 until outboundLanes).foreach { i => queues(OrdinaryQueueIndex + i) = QueueWrapperImpl(createQueue(queueSize, OrdinaryQueueIndex + i)) // ordinary messages stream } - @volatile private[this] var queuesVisibility = false + @volatile private var queuesVisibility = false private def controlQueue: SendQueue.ProducerApi[OutboundEnvelope] = queues(ControlQueueIndex) - @volatile private[this] var _outboundControlIngress: OptionVal[OutboundControlIngress] = OptionVal.None - @volatile private[this] var materializing = new CountDownLatch(1) - @volatile private[this] var outboundCompressionAccess: Vector[OutboundCompressionAccess] = Vector.empty + @volatile private var _outboundControlIngress: OptionVal[OutboundControlIngress] = OptionVal.None + @volatile private var materializing = new CountDownLatch(1) + @volatile private var outboundCompressionAccess: Vector[OutboundCompressionAccess] = Vector.empty // keyed by stream queue index - private[this] val streamMatValues = new AtomicReference(Map.empty[Int, OutboundStreamMatValues]) + private val streamMatValues = new AtomicReference(Map.empty[Int, OutboundStreamMatValues]) - private[this] val idleTimer = new AtomicReference[Option[Cancellable]](None) - private[this] val stopQuarantinedTimer = new AtomicReference[Option[Cancellable]](None) + private val idleTimer = new AtomicReference[Option[Cancellable]](None) + private val stopQuarantinedTimer = new AtomicReference[Option[Cancellable]](None) private[remote] def changeActorRefCompression(table: CompressionTable[ActorRef]): Future[Done] = updateOutboundCompression(c => c.changeActorRefCompression(table)) @@ -1132,8 +1132,8 @@ private[remote] class Association( * INTERNAL API */ private[remote] class AssociationRegistry(createAssociation: Address => Association) { - private[this] val associationsByAddress = new AtomicReference[Map[Address, Association]](Map.empty) - private[this] val associationsByUid = new AtomicReference[ImmutableLongMap[Association]](ImmutableLongMap.empty) + private val associationsByAddress = new AtomicReference[Map[Address, Association]](Map.empty) + private val associationsByUid = new AtomicReference[ImmutableLongMap[Association]](ImmutableLongMap.empty) /** * @throws ShuttingDown if called while the transport is shutting down 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 5da684ebe03..267c33cd24c 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 @@ -240,7 +240,7 @@ private[remote] final class HeaderBuilderImpl( extends HeaderBuilder { import HeaderBuilder.DeadLettersCode - private[this] val toSerializationFormat: SerializationFormatCache = new SerializationFormatCache + private val toSerializationFormat: SerializationFormatCache = new SerializationFormatCache // Fields only available for EnvelopeBuffer var _version: Byte = 0 diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/LruBoundedCache.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/LruBoundedCache.scala index 669ae73b35a..e145a7e1c93 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/artery/LruBoundedCache.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/artery/LruBoundedCache.scala @@ -39,15 +39,15 @@ private[pekko] abstract class LruBoundedCache[K <: AnyRef: ClassTag, V <: AnyRef require((capacity & (capacity - 1)) == 0, "Capacity must be power of two") require(evictAgeThreshold <= capacity, "Age threshold must be less than capacity.") - private[this] val Mask = capacity - 1 + private val Mask = capacity - 1 // Practically guarantee an overflow - private[this] var epoch = Int.MaxValue - 1 + private var epoch = Int.MaxValue - 1 - private[this] val keys = Array.ofDim[K](capacity) - private[this] val values = Array.ofDim[V](capacity) - private[this] val hashes = new Array[Int](capacity) - private[this] val epochs = Array.fill[Int](capacity)(epoch - evictAgeThreshold) // Guarantee existing "values" are stale + private val keys = Array.ofDim[K](capacity) + private val values = Array.ofDim[V](capacity) + private val hashes = new Array[Int](capacity) + private val epochs = Array.fill[Int](capacity)(epoch - evictAgeThreshold) // Guarantee existing "values" are stale final def get(k: K): Option[V] = { val h = hash(k) 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 be3f89c477c..93160b85ba2 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 @@ -69,11 +69,11 @@ private[remote] class ArteryAeronUdpTransport(_system: ExtendedActorSystem, _pro override type LifeCycle = AeronLifecycle - private[this] val mediaDriver = new AtomicReference[Option[MediaDriver]](None) - @volatile private[this] var aeron: Aeron = _ - @volatile private[this] var aeronCounterTask: Cancellable = _ - @volatile private[this] var aeronErrorLogTask: Cancellable = _ - @volatile private[this] var aeronErrorLog: AeronErrorLog = _ + private val mediaDriver = new AtomicReference[Option[MediaDriver]](None) + @volatile private var aeron: Aeron = _ + @volatile private var aeronCounterTask: Cancellable = _ + @volatile private var aeronErrorLogTask: Cancellable = _ + @volatile private var aeronErrorLog: AeronErrorLog = _ private val taskRunner = new TaskRunner(system, settings.Advanced.Aeron.IdleCpuLevel) diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/TaskRunner.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/TaskRunner.scala index a757de28586..706b72b51c8 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/TaskRunner.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/TaskRunner.scala @@ -127,10 +127,10 @@ private[pekko] class TaskRunner(system: ExtendedActorSystem, val idleCpuLevel: I import TaskRunner._ private val log = Logging(system, classOf[TaskRunner]) - private[this] var running = false - private[this] val cmdQueue = new CommandQueue - private[this] val tasks = new ArrayBag[Task] - private[this] val shutdown = Promise[Done]() + private var running = false + private val cmdQueue = new CommandQueue + private val tasks = new ArrayBag[Task] + private val shutdown = Promise[Done]() private val idleStrategy = createIdleStrategy(idleCpuLevel) private var reset = false diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/DecompressionTable.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/DecompressionTable.scala index 1279eddf0b9..575725ff258 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/DecompressionTable.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/DecompressionTable.scala @@ -20,7 +20,7 @@ package org.apache.pekko.remote.artery.compress */ private[remote] final case class DecompressionTable[T](originUid: Long, version: Byte, table: Array[T]) { - private[this] val length = table.length + private val length = table.length def get(idx: Int): T = { if (idx >= length) @@ -44,7 +44,7 @@ private[remote] object DecompressionTable { val DisabledVersion: Byte = -1 - private[this] val _empty = DecompressionTable(0, 0, Array.empty[Any]) + private val _empty = DecompressionTable(0, 0, Array.empty[Any]) def empty[T] = _empty.asInstanceOf[DecompressionTable[T]] def disabled[T] = empty[T].copy(version = DisabledVersion) } diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/InboundCompressions.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/InboundCompressions.scala index 4238af24906..dbb43678a84 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/InboundCompressions.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/InboundCompressions.scala @@ -73,8 +73,8 @@ private[remote] final class InboundCompressionsImpl( flightRecorder: RemotingFlightRecorder = NoOpRemotingFlightRecorder) extends InboundCompressions { - private[this] val _actorRefsIns = new Long2ObjectHashMap[InboundActorRefCompression]() - private[this] val _inboundActorRefsLog = Logging(system, classOf[InboundActorRefCompression]) + private val _actorRefsIns = new Long2ObjectHashMap[InboundActorRefCompression]() + private val _inboundActorRefsLog = Logging(system, classOf[InboundActorRefCompression]) private val createInboundActorRefsForOrigin = new LongFunction[InboundActorRefCompression] { override def apply(originUid: Long): InboundActorRefCompression = { val actorRefHitters = new TopHeavyHitters[ActorRef](settings.ActorRefs.Max) @@ -84,9 +84,9 @@ private[remote] final class InboundCompressionsImpl( private def actorRefsIn(originUid: Long): InboundActorRefCompression = _actorRefsIns.computeIfAbsent(originUid, createInboundActorRefsForOrigin) - private[this] val _classManifestsIns = new Long2ObjectHashMap[InboundManifestCompression]() + private val _classManifestsIns = new Long2ObjectHashMap[InboundManifestCompression]() - private[this] val _inboundManifestLog = Logging(system, classOf[InboundManifestCompression]) + private val _inboundManifestLog = Logging(system, classOf[InboundManifestCompression]) private val createInboundManifestsForOrigin = new LongFunction[InboundManifestCompression] { override def apply(originUid: Long): InboundManifestCompression = { val manifestHitters = new TopHeavyHitters[String](settings.Manifests.Max) @@ -345,14 +345,14 @@ private[remote] abstract class InboundCompression[T >: Null]( inboundContext: InboundContext, val heavyHitters: TopHeavyHitters[T]) { - private[this] var tables: InboundCompression.Tables[T] = InboundCompression.Tables.empty + private var tables: InboundCompression.Tables[T] = InboundCompression.Tables.empty // We should not continue sending advertisements to an association that might be dead (not quarantined yet) - @volatile private[this] var alive = true - private[this] var resendCount = 0 - private[this] val maxResendCount = 3 + @volatile private var alive = true + private var resendCount = 0 + private val maxResendCount = 3 - private[this] val cms = new CountMinSketch(16, 1024, System.currentTimeMillis().toInt) + private val cms = new CountMinSketch(16, 1024, System.currentTimeMillis().toInt) log.debug("Initializing {} for originUid [{}]", Logging.simpleName(getClass), originUid) diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/TopHeavyHitters.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/TopHeavyHitters.scala index be6e27f5327..c5458c9ed63 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/TopHeavyHitters.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/TopHeavyHitters.scala @@ -46,17 +46,17 @@ private[remote] final class TopHeavyHitters[T >: Null](val max: Int)(implicit cl // Contains the hash value for each entry in the hashmap. Used for quicker lookups (equality check can be avoided // if hashes don't match) - private[this] val hashes: Array[Int] = new Array(capacity) + private val hashes: Array[Int] = new Array(capacity) // Actual stored elements in the hashmap - private[this] val items: Array[T] = Array.ofDim[T](capacity) + private val items: Array[T] = Array.ofDim[T](capacity) // Index of stored element in the associated heap - private[this] val heapIndex: Array[Int] = Array.fill(capacity)(-1) + private val heapIndex: Array[Int] = Array.fill(capacity)(-1) // Weights associated with an entry in the hashmap. Used to maintain the heap property and give easy access to low // weight entries - private[this] val weights: Array[Long] = new Array(capacity) + private val weights: Array[Long] = new Array(capacity) // Heap structure containing indices to slots in the hashmap - private[this] val heap: Array[Int] = Array.fill(adjustedMax)(-1) + private val heap: Array[Int] = Array.fill(adjustedMax)(-1) /* * Invariants (apart from heap and hashmap invariants): 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 6cad95e682c..f64a39ac02c 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 @@ -140,7 +140,7 @@ class NettyTransportSettings(config: Config) { case dispatcher => Some(dispatcher) } - private[this] def optionSize(s: String): Option[Int] = getBytes(s).toInt match { + private def optionSize(s: String): Option[Int] = getBytes(s).toInt match { case 0 => None case x if x < 0 => throw new ConfigurationException(s"Setting '$s' must be 0 or positive (and fit in an Int)") case other => Some(other) diff --git a/remote/src/test/scala/org/apache/pekko/remote/artery/MetadataCarryingSpec.scala b/remote/src/test/scala/org/apache/pekko/remote/artery/MetadataCarryingSpec.scala index a1f50c4e929..32aa6c9bb06 100644 --- a/remote/src/test/scala/org/apache/pekko/remote/artery/MetadataCarryingSpec.scala +++ b/remote/src/test/scala/org/apache/pekko/remote/artery/MetadataCarryingSpec.scala @@ -40,7 +40,7 @@ object MetadataCarryingSpy extends ExtensionId[MetadataCarryingSpy] with Extensi class MetadataCarryingSpy extends Extension { def ref: Option[ActorRef] = Option(_ref.get()) def setProbe(bs: ActorRef): Unit = _ref.set(bs) - private[this] val _ref = new AtomicReference[ActorRef]() + private val _ref = new AtomicReference[ActorRef]() } class TestInstrument(system: ExtendedActorSystem) extends RemoteInstrument { diff --git a/stream/src/main/boilerplate/org/apache/pekko/stream/scaladsl/ZipLatestWithApply.scala.template b/stream/src/main/boilerplate/org/apache/pekko/stream/scaladsl/ZipLatestWithApply.scala.template index 80cc5d9baad..d513e5ea638 100644 --- a/stream/src/main/boilerplate/org/apache/pekko/stream/scaladsl/ZipLatestWithApply.scala.template +++ b/stream/src/main/boilerplate/org/apache/pekko/stream/scaladsl/ZipLatestWithApply.scala.template @@ -107,7 +107,7 @@ class ZipLatestWith1[[#A1#], O] (val zipper: ([#A1#]) => O, val eagerComplete: B } private class ZipLatestInlet[T](in: Inlet[T]) extends InHandler { - var value: T = _ + var value: T = null.asInstanceOf[T] var hasValue = false override def onPush() = { diff --git a/stream/src/main/scala/org/apache/pekko/stream/KillSwitch.scala b/stream/src/main/scala/org/apache/pekko/stream/KillSwitch.scala index cd120bf0e12..018538fa01f 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/KillSwitch.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/KillSwitch.scala @@ -171,8 +171,8 @@ private[stream] final class TerminationSignal { def unregister(): Unit = removeListener(this) } - private[this] val _listeners = TrieMap.empty[Listener, NotUsed] - private[this] val _completedWith: AtomicReference[Option[Try[Done]]] = new AtomicReference(None) + private val _listeners = TrieMap.empty[Listener, NotUsed] + private val _completedWith: AtomicReference[Option[Try[Done]]] = new AtomicReference(None) def tryComplete(result: Try[Done]): Unit = { if (_completedWith.compareAndSet(None, Some(result))) { @@ -255,8 +255,8 @@ final class UniqueKillSwitch private[stream] (private val promise: Promise[Done] * This class is thread-safe, the instance can be passed safely among threads and its methods may be invoked concurrently. */ final class SharedKillSwitch private[stream] (val name: String) extends KillSwitch { - private[this] val terminationSignal = new TerminationSignal - private[this] val _flow: Graph[FlowShape[Any, Any], SharedKillSwitch] = new SharedKillSwitchFlow + private val terminationSignal = new TerminationSignal + private val _flow: Graph[FlowShape[Any, Any], SharedKillSwitch] = new SharedKillSwitchFlow /** * After calling [[SharedKillSwitch#shutdown]] all materialized, running instances of all [[Graph]]s provided by the 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 32cc959efe5..ea16196600c 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 @@ -70,7 +70,7 @@ import org.reactivestreams.{ Subscriber, Subscription } } } - private[this] final val states = new Array[State](inputCount) + private final val states = new Array[State](inputCount) private var markCount = 0 private var markedPending = 0 private var markedDepleted = 0 @@ -78,27 +78,27 @@ 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 setState(index: Int, flag: Int, on: Boolean): Unit = + private final def hasState(index: Int, flag: Int): Boolean = (states(index) & flag) != 0 + private final def setState(index: Int, flag: Int, on: Boolean): Unit = states(index) = if (on) (states(index) | flag).toByte else (states(index) & ~flag).toByte - private[this] final def cancelled(index: Int): Boolean = hasState(index, Cancelled) - private[this] final def cancelled(index: Int, on: Boolean): Unit = setState(index, Cancelled, on) + private final def cancelled(index: Int): Boolean = hasState(index, Cancelled) + private final def cancelled(index: Int, on: Boolean): Unit = setState(index, Cancelled, on) - private[this] final def completed(index: Int): Boolean = hasState(index, Completed) - private[this] final def registerCompleted(index: Int): Unit = { + private final def completed(index: Int): Boolean = hasState(index, Completed) + private final def registerCompleted(index: Int): Unit = { completedCounter += 1 setState(index, Completed, true) } - private[this] final def depleted(index: Int): Boolean = hasState(index, Depleted) - private[this] final def depleted(index: Int, on: Boolean): Unit = setState(index, Depleted, on) + private final def depleted(index: Int): Boolean = hasState(index, Depleted) + private final def depleted(index: Int, on: Boolean): Unit = setState(index, Depleted, on) - private[this] final def pending(index: Int): Boolean = hasState(index, Pending) - private[this] final def pending(index: Int, on: Boolean): Unit = setState(index, Pending, on) + private final def pending(index: Int): Boolean = hasState(index, Pending) + private final def pending(index: Int, on: Boolean): Unit = setState(index, Pending, on) - private[this] final def marked(index: Int): Boolean = hasState(index, Marked) - private[this] final def marked(index: Int, on: Boolean): Unit = setState(index, Marked, on) + private final def marked(index: Int): Boolean = hasState(index, Marked) + private final def marked(index: Int, on: Boolean): Unit = setState(index, Marked, on) override def toString: String = s"""|InputBunch diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/JsonObjectParser.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/JsonObjectParser.scala index 6f00d61caa9..369835419c1 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/JsonObjectParser.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/JsonObjectParser.scala @@ -60,15 +60,15 @@ import pekko.util.ByteString @InternalApi private[pekko] class JsonObjectParser(maximumObjectLength: Int = Int.MaxValue) { import JsonObjectParser._ - private[this] var buffer: Array[Byte] = Array.empty + private var buffer: Array[Byte] = Array.empty - private[this] var pos = 0 // latest position of pointer while scanning for json object end - private[this] var start = 0 // number of chars to drop from the front of the bytestring before emitting (skip whitespace etc) - private[this] var depth = 0 // counter of object-nesting depth, once hits 0 an object should be emitted + private var pos = 0 // latest position of pointer while scanning for json object end + private var start = 0 // number of chars to drop from the front of the bytestring before emitting (skip whitespace etc) + private var depth = 0 // counter of object-nesting depth, once hits 0 an object should be emitted - private[this] var completedObject = false - private[this] var inStringExpression = false - private[this] var inBackslashEscape = false + private var completedObject = false + private var inStringExpression = false + private var inBackslashEscape = false /** * Appends input ByteString to internal buffer. diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/PhasedFusingActorMaterializer.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/PhasedFusingActorMaterializer.scala index 0a3861984eb..9445ae796bb 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/PhasedFusingActorMaterializer.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/PhasedFusingActorMaterializer.scala @@ -451,7 +451,7 @@ private final case class SavedIslandData( override def withNamePrefix(name: String): PhasedFusingActorMaterializer = this.copy(flowNames = flowNames.copy(name)) - private[this] def createFlowName(): String = flowNames.next() + private def createFlowName(): String = flowNames.next() // note that this will never be overridden on a per-graph-stage basis regardless of more specific attributes override lazy val executionContext: ExecutionContextExecutor = @@ -725,7 +725,7 @@ private[pekko] object GraphStageIsland { islandName: String, subflowFuser: OptionVal[GraphInterpreterShell => ActorRef]) extends PhaseIsland[GraphStageLogic] { - private[this] val logics = new util.ArrayList[GraphStageLogic](16) + private val logics = new util.ArrayList[GraphStageLogic](16) private var connections = new Array[Connection](16) private var maxConnections = 0 @@ -972,7 +972,7 @@ private[pekko] object GraphStageIsland { */ @InternalApi private[pekko] final class ProcessorModulePhase() extends PhaseIsland[Processor[Any, Any]] { override def name: String = "ProcessorModulePhase" - private[this] var processor: Processor[Any, Any] = _ + private var processor: Processor[Any, Any] = _ override def materializeAtomic(mod: AtomicModule[Shape, Any], attributes: Attributes): (Processor[Any, Any], Any) = { val procAndMat = mod.asInstanceOf[ProcessorModule[Any, Any, Any]].createProcessor() diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/ResizableMultiReaderRingBuffer.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/ResizableMultiReaderRingBuffer.scala index fc215f2dab0..1e5c684c78b 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/ResizableMultiReaderRingBuffer.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/ResizableMultiReaderRingBuffer.scala @@ -37,15 +37,15 @@ import org.apache.pekko.annotation.InternalApi Integer.lowestOneBit(initialSize) == initialSize && 0 < initialSize && initialSize <= maxSize, "initialSize must be a power of 2 that is > 0 and <= maxSize") - private[this] val maxSizeBit = Integer.numberOfTrailingZeros(maxSize) - private[this] var array = new Array[Any](initialSize) + private val maxSizeBit = Integer.numberOfTrailingZeros(maxSize) + private var array = new Array[Any](initialSize) /* * two counters counting the number of elements ever written and read; wrap-around is * handled by always looking at differences or masked values */ - private[this] var writeIx = 0 - private[this] var readIx = 0 // the "oldest" of all read cursor indices, i.e. the one that is most behind + private var writeIx = 0 + private var readIx = 0 // the "oldest" of all read cursor indices, i.e. the one that is most behind // current array.length log2, we don't keep it as an extra field because `Integer.numberOfTrailingZeros` // is a JVM intrinsic compiling down to a `BSF` instruction on x86, which is very fast on modern CPUs diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/SinkholeSubscriber.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/SinkholeSubscriber.scala index 4fd02e2162a..2e43555e3c6 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/SinkholeSubscriber.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/SinkholeSubscriber.scala @@ -25,7 +25,7 @@ import org.reactivestreams.{ Subscriber, Subscription } * INTERNAL API */ @InternalApi private[pekko] final class SinkholeSubscriber[T](whenComplete: Promise[Done]) extends Subscriber[T] { - private[this] var running: Boolean = false + private var running: Boolean = false override def onSubscribe(sub: Subscription): Unit = { ReactiveStreamsCompliance.requireNonNullSubscription(sub) diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/Sinks.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/Sinks.scala index de215da3eae..5d61e9f5bfd 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/Sinks.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/Sinks.scala @@ -170,8 +170,8 @@ import org.reactivestreams.Subscriber override def createLogicAndMaterializedValue(inheritedAttributes: Attributes) = { val p: Promise[immutable.Seq[T]] = Promise() (new GraphStageLogic(shape) with InHandler { - private[this] val buffer = mutable.Queue.empty[T] - private[this] var count = 0 + private val buffer = mutable.Queue.empty[T] + private var count = 0 override def preStart(): Unit = pull(in) diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/SubscriberManagement.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/SubscriberManagement.scala index 91c59a6bff6..c44b3c43b79 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/SubscriberManagement.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/SubscriberManagement.scala @@ -93,18 +93,18 @@ private[pekko] trait SubscriberManagement[T] extends ResizableMultiReaderRingBuf */ protected def createSubscription(subscriber: Subscriber[? >: T]): S - private[this] val buffer = new ResizableMultiReaderRingBuffer[T](initialBufferSize, maxBufferSize, this) + private val buffer = new ResizableMultiReaderRingBuffer[T](initialBufferSize, maxBufferSize, this) protected def bufferDebug: String = buffer.toString // optimize for small numbers of subscribers by keeping subscribers in a plain list - private[this] var subscriptions: Subscriptions = Nil + private var subscriptions: Subscriptions = Nil // number of elements already requested but not yet received from upstream - private[this] var pendingFromUpstream: Long = 0 + private var pendingFromUpstream: Long = 0 // if non-null, holds the end-of-stream state - private[this] var endOfStream: EndOfStream = NotReached + private var endOfStream: EndOfStream = NotReached def cursors = subscriptions @@ -158,7 +158,7 @@ private[pekko] trait SubscriberManagement[T] extends ResizableMultiReaderRingBuf } } - private[this] final def requestFromUpstreamIfRequired(): Unit = { + private final def requestFromUpstreamIfRequired(): Unit = { @tailrec def maxRequested(remaining: Subscriptions, result: Long = 0): Long = remaining match { case head :: tail => maxRequested(tail, math.max(head.totalDemand, result)) diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/TraversalBuilder.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/TraversalBuilder.scala index d4a7da20749..658f668890a 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/TraversalBuilder.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/TraversalBuilder.scala @@ -693,8 +693,8 @@ import pekko.util.OptionVal private val cachedEmptyLinear = LinearTraversalBuilder(OptionVal.None, OptionVal.None, 0, 0, PushNotUsed, OptionVal.None, Attributes.none) - private[this] final val wireBackward: Array[Int] = Array(-1) - private[this] final val noWire: Array[Int] = Array() + private final val wireBackward: Array[Int] = Array(-1) + private final val noWire: Array[Int] = Array() def empty(attributes: Attributes = Attributes.none): LinearTraversalBuilder = if (attributes eq Attributes.none) cachedEmptyLinear diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/Unfold.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/Unfold.scala index 44cf73d6d78..9bcd869b5cf 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/Unfold.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/Unfold.scala @@ -36,7 +36,7 @@ import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler } override def initialAttributes: Attributes = DefaultAttributes.unfold and SourceLocation.forLambda(f) override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with OutHandler { - private[this] var state = s + private var state = s def onPull(): Unit = f(state) match { case Some((newState, v)) => { @@ -91,8 +91,8 @@ private[pekko] final class UnfoldJava[S, E](s: S, f: function.Function[S, Option override def initialAttributes: Attributes = DefaultAttributes.unfoldAsync override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with OutHandler { - private[this] var state = s - private[this] var asyncHandler: Try[Option[(S, E)]] => Unit = _ + private var state = s + private var asyncHandler: Try[Option[(S, E)]] => Unit = _ override def preStart(): Unit = { asyncHandler = getAsyncCallback[Try[Option[(S, E)]]](handle).invoke @@ -135,8 +135,8 @@ private[pekko] final class UnfoldJava[S, E](s: S, f: function.Function[S, Option override def initialAttributes: Attributes = DefaultAttributes.unfoldAsync override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with OutHandler { - private[this] var state = s - private[this] var asyncHandler: Try[Optional[Pair[S, E]]] => Unit = _ + private var state = s + private var asyncHandler: Try[Optional[Pair[S, E]]] => Unit = _ override def preStart(): Unit = { asyncHandler = getAsyncCallback[Try[Optional[Pair[S, E]]]](handle).invoke diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala index 61eb9b8a293..a7156ff29ec 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala @@ -403,7 +403,7 @@ import org.reactivestreams.Subscription private var downstreamDemand: Long = 0L // This flag is only used if complete/fail is called externally since this op turns into a Finished one inside the // interpreter (i.e. inside this op this flag has no effects since if it is completed the op will not be invoked) - private[this] var downstreamCompletionCause: Option[Throwable] = None + private var downstreamCompletionCause: Option[Throwable] = None def downstreamCompleted: Boolean = downstreamCompletionCause.isDefined // when upstream failed before we got the exposed publisher diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/AggregateWithBoundary.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/AggregateWithBoundary.scala index 79b547c4db4..7d886a8ed09 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/AggregateWithBoundary.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/AggregateWithBoundary.scala @@ -42,7 +42,7 @@ private[pekko] final case class AggregateWithBoundary[In, Agg, Out]( override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new TimerGraphStageLogic(shape) with InHandler with OutHandler { - private[this] var aggregated: Agg = null.asInstanceOf[Agg] + private var aggregated: Agg = null.asInstanceOf[Agg] override def preStart(): Unit = { emitOnTimer.foreach { diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala index aedcd0b301a..edaa45f31f8 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala @@ -225,7 +225,7 @@ import pekko.stream.stage._ val context: ActorRef) { import GraphInterpreter._ - private[this] val ChaseLimit = if (fuzzingMode) 0 else 16 + private val ChaseLimit = if (fuzzingMode) 0 else 16 /** * INTERNAL API @@ -234,35 +234,35 @@ import pekko.stream.stage._ // The number of currently running stages. Once this counter reaches zero, the interpreter is considered to be // completed - private[this] var runningStages = logics.length + private var runningStages = logics.length // Counts how many active connections a stage has. Once it reaches zero, the stage is automatically stopped. - private[this] val shutdownCounter = Array.tabulate(logics.length) { i => + private val shutdownCounter = Array.tabulate(logics.length) { i => logics(i).handlers.length } // Marks whether a stage has been finalized (finalizeStage been called) or not - private[this] val finalizedMark = Array.fill(logics.length)(false) + private val finalizedMark = Array.fill(logics.length)(false) - private[this] var _subFusingMaterializer: Materializer = _ - private[this] lazy val defaultErrorReportingLogLevel = LogLevels.defaultErrorLevel(materializer.system) + private var _subFusingMaterializer: Materializer = _ + private lazy val defaultErrorReportingLogLevel = LogLevels.defaultErrorLevel(materializer.system) def subFusingMaterializer: Materializer = _subFusingMaterializer // An event queue implemented as a circular buffer // FIXME: This calculates the maximum size ever needed, but most assemblies can run on a smaller queue - private[this] val eventQueue = new Array[Connection](1 << (32 - Integer.numberOfLeadingZeros(connections.length - 1))) - private[this] val mask = eventQueue.length - 1 - private[this] var queueHead: Int = 0 - private[this] var queueTail: Int = 0 - - private[this] var chaseCounter = 0 // the first events in preStart blocks should be not chased - private[this] var chasedPush: Connection = NoEvent - private[this] var chasedPull: Connection = NoEvent + private val eventQueue = new Array[Connection](1 << (32 - Integer.numberOfLeadingZeros(connections.length - 1))) + private val mask = eventQueue.length - 1 + private var queueHead: Int = 0 + private var queueTail: Int = 0 + + private var chaseCounter = 0 // the first events in preStart blocks should be not chased + private var chasedPush: Connection = NoEvent + private var chasedPull: Connection = NoEvent // Set whenever a stage's shutdownCounter transitions to 0 (i.e. the stage just became completed and // needs finalization). Lets the chase / dispatch loops skip the per-iteration shutdownCounter array // load in afterStageHasRun when no stage has completed since the last finalization pass. - private[this] var pendingFinalization: Boolean = false + private var pendingFinalization: Boolean = false private def queueStatus: String = { val contents = (queueHead until queueTail).map(idx => { @@ -271,7 +271,7 @@ import pekko.stream.stage._ }) s"(${eventQueue.length}, $queueHead, $queueTail)(${contents.mkString(", ")})" } - private[this] var _Name: String = _ + private var _Name: String = _ def Name: String = if (_Name eq null) { _Name = f"${System.identityHashCode(this)}%08X" 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 ba21468469a..3832ccf2695 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 @@ -1422,7 +1422,7 @@ private[stream] object Collect { private var buffer: BufferImpl[Out] = _ private val invokeFutureCB: Try[Out] => Unit = getAsyncCallback(futureCompleted).invoke - private[this] def todo: Int = inFlight + buffer.used + private def todo: Int = inFlight + buffer.used override def preStart(): Unit = buffer = BufferImpl(parallelism, inheritedAttributes) @@ -1920,14 +1920,14 @@ private[stream] object Collect { new TimerGraphStageLogic(shape) with InHandler with OutHandler { import Delay._ - private[this] val size = inheritedAttributes.mandatoryAttribute[InputBuffer].max + private val size = inheritedAttributes.mandatoryAttribute[InputBuffer].max - private[this] val delayStrategy = delayStrategySupplier() + private val delayStrategy = delayStrategySupplier() // buffer has pairs of timestamp of expected push and element - private[this] val buffer = BufferImpl[(Long, T)](size, inheritedAttributes) + private val buffer = BufferImpl[(Long, T)](size, inheritedAttributes) - private[this] val onPushWhenBufferFull: () => Unit = overflowStrategy match { + private val onPushWhenBufferFull: () => Unit = overflowStrategy match { case EmitEarly => () => { if (isAvailable(out)) { diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/RangeSource.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/RangeSource.scala index 4e9cad700e9..e8078e38545 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/RangeSource.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/RangeSource.scala @@ -36,14 +36,14 @@ private[pekko] final class RangeSource[T](val range: immutable.Range, defaultAtt private val out = Outlet[T]("RangeSource.out") override val shape: SourceShape[T] = SourceShape(out) - private[this] val isEmptyRange = range.isEmpty - private[this] val rangeStart = range.start - private[this] val rangeLast = if (isEmptyRange) 0 else range.last - private[this] val rangeStep = range.step + private val isEmptyRange = range.isEmpty + private val rangeStart = range.start + private val rangeLast = if (isEmptyRange) 0 else range.last + private val rangeStep = range.step override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with OutHandler { - private[this] var nextElement = rangeStart + private var nextElement = rangeStart override def preStart(): Unit = if (isEmptyRange) completeStage() diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/StreamOfStreams.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/StreamOfStreams.scala index 1d7ffff8b45..882e717f5d9 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/StreamOfStreams.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/StreamOfStreams.scala @@ -851,7 +851,7 @@ import pekko.util.OptionVal override def createLogic(attr: Attributes) = new GraphStageLogic(shape) with InHandler { // check for previous materialization eagerly so we fail with a more useful stacktrace - private[this] val materializationException: OptionVal[IllegalStateException] = + private val materializationException: OptionVal[IllegalStateException] = if (status.get.isInstanceOf[AsyncCallback[?]]) OptionVal.Some(createMaterializedTwiceException()) else @@ -946,7 +946,7 @@ import pekko.util.OptionVal override def createLogic(inheritedAttributes: Attributes) = new GraphStageLogic(shape) with OutHandler { // check for previous materialization eagerly so we fail with a more useful stacktrace - private[this] val materializationException: OptionVal[IllegalStateException] = + private val materializationException: OptionVal[IllegalStateException] = if (status.get.isInstanceOf[AsyncCallback[?]]) OptionVal.Some(createMaterializedTwiceException()) else diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/ByteStringParser.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/ByteStringParser.scala index bef6e0e7d41..bd818670260 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/ByteStringParser.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/ByteStringParser.scala @@ -191,7 +191,7 @@ import pekko.util.ByteString class ByteReader(input: ByteString) { - private[this] var off = 0 + private var off = 0 def hasRemaining: Boolean = off < input.length def remainingSize: Int = input.length - off diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSinkStage.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSinkStage.scala index 5e6be230fdd..be3d18e7849 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSinkStage.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSinkStage.scala @@ -135,7 +135,7 @@ private[stream] object InputStreamSinkStage { var detachedChunk: Option[ByteString] = None @scala.throws(classOf[IOException]) - private[this] def executeIfNotClosed[T](f: () => T): T = + private def executeIfNotClosed[T](f: () => T): T = if (isActive.get()) { waitIfNotInitialized() f() @@ -190,7 +190,7 @@ private[stream] object InputStreamSinkStage { } else -1) } - private[this] def readBytes(a: Array[Byte], begin: Int, length: Int): Int = { + private def readBytes(a: Array[Byte], begin: Int, length: Int): Int = { require(detachedChunk.nonEmpty, "Chunk must be pulled from shared buffer") val availableInChunk = detachedChunk.get.size val readBytes = getData(a, begin, length, 0) @@ -207,7 +207,7 @@ private[stream] object InputStreamSinkStage { } @tailrec - private[this] def getData(arr: Array[Byte], begin: Int, length: Int, gotBytes: Int): Int = { + private def getData(arr: Array[Byte], begin: Int, length: Int, gotBytes: Int): Int = { grabDataChunk() match { case Some(data) => val size = data.size @@ -227,7 +227,7 @@ private[stream] object InputStreamSinkStage { } } - private[this] def waitIfNotInitialized(): Unit = { + private def waitIfNotInitialized(): Unit = { if (!isInitialized) { sharedBuffer.poll(readTimeout.toMillis, TimeUnit.MILLISECONDS) match { case Initialized => isInitialized = true @@ -237,7 +237,7 @@ private[stream] object InputStreamSinkStage { } } - private[this] def grabDataChunk(): Option[ByteString] = { + private def grabDataChunk(): Option[ByteString] = { detachedChunk match { case None => sharedBuffer.poll() match { diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/OutputStreamSourceStage.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/OutputStreamSourceStage.scala index d0dbb81c27e..a4904761a5f 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/OutputStreamSourceStage.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/OutputStreamSourceStage.scala @@ -81,7 +81,7 @@ private[pekko] class OutputStreamAdapter( extends OutputStream { @scala.throws(classOf[IOException]) - private[this] def sendData(data: ByteString): Unit = { + private def sendData(data: ByteString): Unit = { if (!unfulfilledDemand.tryAcquire(writeTimeout.toMillis, TimeUnit.MILLISECONDS)) { throw new IOException("Timed out trying to write data to stream") } diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/DeflateDecompressorBase.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/DeflateDecompressorBase.scala index 8c523fd4c23..922bdf7ba91 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/DeflateDecompressorBase.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/DeflateDecompressorBase.scala @@ -37,7 +37,7 @@ import pekko.util.ByteString * buffer around avoids reallocating a buffer that may be too big in many * cases for every call of `parse`. */ - private[this] val buffer = new Array[Byte](maxBytesPerChunk) + private val buffer = new Array[Byte](maxBytesPerChunk) abstract class Inflate(noPostProcessing: Boolean) extends ParseStep[ByteString] { override def canWorkWithPartialData = true diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SinkRefImpl.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SinkRefImpl.scala index 67121c74214..53b8a89c8d1 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SinkRefImpl.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SinkRefImpl.scala @@ -71,11 +71,11 @@ private[stream] final class SinkRefStageImpl[In] private[pekko] (val initialPart val logic = new TimerGraphStageLogic(shape) with StageLogging with ActorRefStage with InHandler { override protected def logSource: Class[?] = classOf[SinkRefStageImpl[?]] - private[this] val streamRefsMaster = StreamRefsMaster(eagerMaterializer.system) + private val streamRefsMaster = StreamRefsMaster(eagerMaterializer.system) // settings --- @nowarn("msg=deprecated") // can't remove this settings access without breaking compat - private[this] val subscriptionTimeout = { + private val subscriptionTimeout = { import StreamRefAttributes._ val settings = eagerMaterializer.settings.streamRefSettings inheritedAttributes.get[StreamRefAttributes.SubscriptionTimeout]( @@ -83,7 +83,7 @@ private[stream] final class SinkRefStageImpl[In] private[pekko] (val initialPart } @nowarn("msg=deprecated") // can't remove this settings access without breaking compat - private[this] val finalTerminationSignalDeadline = { + private val finalTerminationSignalDeadline = { import StreamRefAttributes._ val settings = eagerMaterializer.settings.streamRefSettings inheritedAttributes @@ -94,7 +94,7 @@ private[stream] final class SinkRefStageImpl[In] private[pekko] (val initialPart // end of settings --- override protected val stageActorName: String = streamRefsMaster.nextSinkRefStageName() - private[this] val self: GraphStageLogic.StageActor = + private val self: GraphStageLogic.StageActor = getEagerStageActor(eagerMaterializer)(initialReceive) override val ref: ActorRef = self.ref implicit def selfSender: ActorRef = ref @@ -119,7 +119,7 @@ private[stream] final class SinkRefStageImpl[In] private[pekko] (val initialPart // When this side of the stream has completed/failed, and we await the Terminated() signal back from the partner // so we can safely shut down completely; This is to avoid *our* Terminated() signal to reach the partner before the // Complete/Fail message does, which can happen on transports such as Artery which use a dedicated lane for system messages (Terminated) - private[this] var finishedWithAwaitingPartnerTermination: OptionVal[Try[Done]] = OptionVal.None + private var finishedWithAwaitingPartnerTermination: OptionVal[Try[Done]] = OptionVal.None override def preStart(): Unit = { initialPartnerRef match { diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SourceRefImpl.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SourceRefImpl.scala index f5cf889272c..4e392cc35b5 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SourceRefImpl.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SourceRefImpl.scala @@ -130,29 +130,29 @@ private[stream] final class SourceRefStageImpl[Out](val initialPartnerRef: Optio val logic = new TimerGraphStageLogic(shape) with StageLogging with ActorRefStage with OutHandler { override protected def logSource: Class[?] = classOf[SourceRefStageImpl[?]] - private[this] val streamRefsMaster = StreamRefsMaster(eagerMaterializer.system) + private val streamRefsMaster = StreamRefsMaster(eagerMaterializer.system) // settings --- import StreamRefAttributes._ @nowarn("msg=deprecated") // can't remove this settings access without breaking compat - private[this] val settings = eagerMaterializer.settings.streamRefSettings + private val settings = eagerMaterializer.settings.streamRefSettings @nowarn("msg=deprecated") // can't remove this settings access without breaking compat - private[this] val subscriptionTimeout = inheritedAttributes.get[StreamRefAttributes.SubscriptionTimeout]( + private val subscriptionTimeout = inheritedAttributes.get[StreamRefAttributes.SubscriptionTimeout]( SubscriptionTimeout(settings.subscriptionTimeout)) @nowarn("msg=deprecated") // can't remove this settings access without breaking compat - private[this] val bufferCapacity = inheritedAttributes + private val bufferCapacity = inheritedAttributes .get[StreamRefAttributes.BufferCapacity](StreamRefAttributes.BufferCapacity(settings.bufferCapacity)) .capacity @nowarn("msg=deprecated") // can't remove this settings access without breaking compat - private[this] val demandRedeliveryInterval = inheritedAttributes + private val demandRedeliveryInterval = inheritedAttributes .get[StreamRefAttributes.DemandRedeliveryInterval](DemandRedeliveryInterval(settings.demandRedeliveryInterval)) .timeout @nowarn("msg=deprecated") // can't remove this settings access without breaking compat - private[this] val finalTerminationSignalDeadline = + private val finalTerminationSignalDeadline = inheritedAttributes .get[StreamRefAttributes.FinalTerminationSignalDeadline]( FinalTerminationSignalDeadline(settings.finalTerminationSignalDeadline)) @@ -160,10 +160,10 @@ private[stream] final class SourceRefStageImpl[Out](val initialPartnerRef: Optio // end of settings --- override protected val stageActorName: String = streamRefsMaster.nextSourceRefStageName() - private[this] val self: GraphStageLogic.StageActor = + private val self: GraphStageLogic.StageActor = getEagerStageActor(eagerMaterializer)(receiveRemoteMessage) override val ref: ActorRef = self.ref - private[this] implicit def selfSender: ActorRef = ref + private implicit def selfSender: ActorRef = ref // demand management --- private var state: State = initialPartnerRef match { diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/StreamRefsMaster.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/StreamRefsMaster.scala index f2be78e2069..0bd307f85e2 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/StreamRefsMaster.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/StreamRefsMaster.scala @@ -36,8 +36,8 @@ private[stream] object StreamRefsMaster extends ExtensionId[StreamRefsMaster] wi @InternalApi private[stream] final class StreamRefsMaster extends Extension { - private[this] val sourceRefStageNames = SeqActorName("SourceRef") // "local target" - private[this] val sinkRefStageNames = SeqActorName("SinkRef") // "remote sender" + private val sourceRefStageNames = SeqActorName("SourceRef") // "local target" + private val sinkRefStageNames = SeqActorName("SinkRef") // "remote sender" // TODO introduce a master with which all stages running the streams register themselves? diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/BidiFlow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/BidiFlow.scala index cc9931759ee..b7e0c9b2dd5 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/BidiFlow.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/BidiFlow.scala @@ -20,7 +20,7 @@ import pekko.stream._ object BidiFlow { - private[this] val _identity: BidiFlow[Object, Object, Object, Object, NotUsed] = + private val _identity: BidiFlow[Object, Object, Object, Object, NotUsed] = BidiFlow.fromFlows(Flow.of(classOf[Object]), Flow.of(classOf[Object])) def identity[A, B]: BidiFlow[A, A, B, B, NotUsed] = _identity.asInstanceOf[BidiFlow[A, A, B, B, NotUsed]] diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/DelayStrategy.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/DelayStrategy.scala index 993b0c81a2c..9f4addc0a84 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/DelayStrategy.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/DelayStrategy.scala @@ -93,7 +93,7 @@ object DelayStrategy { new DelayStrategy[T] { - private[this] var delay = initialDelay + private var delay = initialDelay override def nextDelay(elem: T): java.time.Duration = { if (needsIncrease(elem)) { diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Graph.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Graph.scala index 0a919fde2a1..ba9264a8b91 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Graph.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Graph.scala @@ -479,7 +479,7 @@ object Zip { def create[A, B]: Graph[FanInShape2[A, B, A Pair B], NotUsed] = ZipWith.create(_toPair.asInstanceOf[Function2[A, B, A Pair B]]) - private[this] final val _toPair: Function2[Any, Any, Any Pair Any] = (a: Any, b: Any) => new Pair(a, b) + private final val _toPair: Function2[Any, Any, Any Pair Any] = (a: Any, b: Any) => new Pair(a, b) } /** @@ -507,7 +507,7 @@ object ZipLatest { def create[A, B]: Graph[FanInShape2[A, B, A Pair B], NotUsed] = ZipLatestWith.create(_toPair.asInstanceOf[Function2[A, B, A Pair B]]) - private[this] final val _toPair: Function2[Any, Any, Any Pair Any] = (a: Any, b: Any) => new Pair(a, b) + private final val _toPair: Function2[Any, Any, Any Pair Any] = (a: Any, b: Any) => new Pair(a, b) } /** diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/BidiFlow.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/BidiFlow.scala index 9b84b0af80d..47598bf82c3 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/BidiFlow.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/BidiFlow.scala @@ -234,7 +234,7 @@ final class BidiFlow[-I1, +O1, -I2, +O2, +Mat]( } object BidiFlow { - private[this] val _identity: BidiFlow[Any, Any, Any, Any, NotUsed] = + private val _identity: BidiFlow[Any, Any, Any, Any, NotUsed] = BidiFlow.fromFlows(Flow[Any], Flow[Any]) def identity[A, B]: BidiFlow[A, A, B, B, NotUsed] = _identity.asInstanceOf[BidiFlow[A, A, B, B, NotUsed]] diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/DelayStrategy.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/DelayStrategy.scala index 549a1583948..1323c344e77 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/DelayStrategy.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/DelayStrategy.scala @@ -58,7 +58,7 @@ object DelayStrategy { new DelayStrategy[T] { - private[this] var delay: FiniteDuration = initialDelay + private var delay: FiniteDuration = initialDelay override def nextDelay(elem: T): FiniteDuration = { if (needsIncrease(elem)) { diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala index 59e524fc4cd..4ae16e46ce4 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala @@ -428,7 +428,7 @@ object Flow { private[stream] val identityTraversalBuilder = LinearTraversalBuilder.fromBuilder(GraphStages.identity.traversalBuilder, GraphStages.identity.shape, Keep.right) - private[this] val identity: Flow[Any, Any, NotUsed] = + private val identity: Flow[Any, Any, NotUsed] = new Flow[Any, Any, NotUsed](identityTraversalBuilder, GraphStages.identity.shape) /** diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala index 16b9a605c0c..6cde3ffa9f2 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala @@ -1415,8 +1415,8 @@ private[stream] final class OrElse[T] override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with OutHandler with InHandler { - private[this] var currentIn = primary - private[this] var primaryPushed = false + private var currentIn = primary + private var primaryPushed = false override def onPull(): Unit = { pull(currentIn) diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala index 8849529e6d2..32e647841f7 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala @@ -146,7 +146,7 @@ private[pekko] class MergeHub[T](perProducerBufferSize: Int, drainingEnabled: Bo override val shape: SourceShape[T] = SourceShape(out) // Half of buffer size, rounded up - private[this] val DemandThreshold = (perProducerBufferSize / 2) + (perProducerBufferSize % 2) + private val DemandThreshold = (perProducerBufferSize / 2) + (perProducerBufferSize % 2) private sealed trait Event { def id: Long @@ -180,12 +180,12 @@ private[pekko] class MergeHub[T](perProducerBufferSize: Int, drainingEnabled: Bo * processing of control messages. This causes no issues though, see the explanation in 'tryProcessNext'. */ private val queue = new AbstractNodeQueue[Event] {} - @volatile private[this] var needWakeup = false - @volatile private[this] var shuttingDown = false - @volatile private[this] var draining = false + @volatile private var needWakeup = false + @volatile private var shuttingDown = false + @volatile private var draining = false - private[this] val demands = scala.collection.mutable.LongMap.empty[InputState] - private[this] val wakeupCallback = getAsyncCallback[NotUsed](_ => + private val demands = scala.collection.mutable.LongMap.empty[InputState] + private val wakeupCallback = getAsyncCallback[NotUsed](_ => // We are only allowed to dequeue if we are not backpressured. See comment in tryProcessNext() for details. if (isAvailable(out)) tryProcessNext(firstAttempt = true)) @@ -319,8 +319,8 @@ private[pekko] class MergeHub[T](perProducerBufferSize: Int, drainingEnabled: Bo new GraphStageLogic(shape) with InHandler { // Start from non-zero demand to avoid initial delays. // The HUB will expect this behavior. - private[this] var demand: Long = perProducerBufferSize - private[this] val id = idCounter.getAndIncrement() + private var demand: Long = perProducerBufferSize + private val id = idCounter.getAndIncrement() override def preStart(): Unit = { if (!logic.isDraining && !logic.isShuttingDown) { @@ -495,7 +495,7 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I override val shape: SinkShape[T] = SinkShape(in) // Half of buffer size, rounded up - private[this] val DemandThreshold = (bufferSize / 2) + (bufferSize % 2) + private val DemandThreshold = (bufferSize / 2) + (bufferSize % 2) private sealed trait HubEvent @@ -515,22 +515,22 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I private class BroadcastSinkLogic(_shape: Shape) extends GraphStageLogic(_shape) with InHandler { - private[this] val callbackPromise: Promise[AsyncCallback[HubEvent]] = Promise() - private[this] val noRegistrationsState = Open(callbackPromise.future, Nil) + private val callbackPromise: Promise[AsyncCallback[HubEvent]] = Promise() + private val noRegistrationsState = Open(callbackPromise.future, Nil) val state = new AtomicReference[HubState](noRegistrationsState) private var initialized = false // Start from values that will almost immediately overflow. This has no effect on performance, any starting // number will do, however, this protects from regressions as these values *almost surely* overflow and fail // tests if someone makes a mistake. - @volatile private[this] var tail = Int.MaxValue - private[this] var head = Int.MaxValue + @volatile private var tail = Int.MaxValue + private var head = Int.MaxValue /* * An Array with a published tail ("latest message") and a privately maintained head ("earliest buffered message"). * Elements are published by simply putting them into the array and bumping the tail. If necessary, certain * consumers are sent a wakeup message through an AsyncCallback. */ - private[this] val queue = new Array[AnyRef](bufferSize) + private val queue = new Array[AnyRef](bufferSize) /* This is basically a classic Bucket Queue: https://en.wikipedia.org/wiki/Bucket_queue * (in fact, this is the variant described in the Optimizations section, where the given set * of priorities always fall to a range @@ -545,9 +545,9 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I * Empty slots are null (no backing map allocated), reducing baseline memory and GC pressure. * When a slot drains to zero consumers, its map is released (set to null). */ - private[this] val consumerWheel = + private val consumerWheel = new Array[LongMap[Consumer]](bufferSize * 2) - private[this] var activeConsumers = 0 + private var activeConsumers = 0 override def preStart(): Unit = { setKeepGoing(true) @@ -812,10 +812,10 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with OutHandler { - private[this] var untilNextAdvanceSignal = DemandThreshold - private[this] val id = idCounter.getAndIncrement() - private[this] var offsetInitialized = false - private[this] var hubCallback: AsyncCallback[HubEvent] = _ + private var untilNextAdvanceSignal = DemandThreshold + private val id = idCounter.getAndIncrement() + private var offsetInitialized = false + private var hubCallback: AsyncCallback[HubEvent] = _ /* * We need to track our last offset that we published to the Hub. The reason is, that for efficiency reasons, @@ -823,8 +823,8 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I * is needed, but it also means that we need to keep track of both our current offset, and the last one that * we published. */ - private[this] var previousPublishedOffset = 0 - private[this] var offset = 0 + private var previousPublishedOffset = 0 + private var offset = 0 override def preStart(): Unit = { val callback = getAsyncCallback(onCommand) diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala index 2fa336579b6..d4e78bebf94 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala @@ -319,7 +319,7 @@ object Sink { */ def source[T]: Sink[T, Source[T, NotUsed]] = _sourceSink.asInstanceOf[Sink[T, Source[T, NotUsed]]] - private[this] val _sourceSink = fromGraph(SourceSink) + private val _sourceSink = fromGraph(SourceSink) /** * A `Sink` that will consume the stream and discard the elements. @@ -330,7 +330,7 @@ object Sink { * A [[Sink]] that will always backpressure never cancel and never consume any elements from the stream. */ def never: Sink[Any, Future[Done]] = _never - private[this] val _never: Sink[Any, Future[Done]] = fromGraph(GraphStages.NeverSink) + private val _never: Sink[Any, Future[Done]] = fromGraph(GraphStages.NeverSink) /** * A `Sink` that will invoke the given procedure for each received element. The sink is materialized diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala index 39b81555263..4cd4594e1d0 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala @@ -575,7 +575,7 @@ object Source { * A `Source` with no elements, i.e. an empty stream that is completed immediately for every connected `Sink`. */ def empty[T]: Source[T, NotUsed] = _empty - private[this] val _empty: Source[Nothing, NotUsed] = + private val _empty: Source[Nothing, NotUsed] = fromGraphStage(EmptySource) /** @@ -616,7 +616,7 @@ object Source { * This stream could be useful in tests. */ def never[T]: Source[T, NotUsed] = _never - private[this] val _never: Source[Nothing, NotUsed] = fromGraphStage(GraphStages.NeverSource) + private val _never: Source[Nothing, NotUsed] = fromGraphStage(GraphStages.NeverSource) /** * Emits a single value when the given `CompletionStage` is successfully completed and then completes the stream. diff --git a/stream/src/main/scala/org/apache/pekko/stream/serialization/StreamRefSerializer.scala b/stream/src/main/scala/org/apache/pekko/stream/serialization/StreamRefSerializer.scala index daa66f753c2..5badc9b2a5e 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/serialization/StreamRefSerializer.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/serialization/StreamRefSerializer.scala @@ -30,16 +30,16 @@ private[pekko] final class StreamRefSerializer(val system: ExtendedActorSystem) extends SerializerWithStringManifest with BaseSerializer { - private[this] lazy val serialization = SerializationExtension(system) - - private[this] val SequencedOnNextManifest = "A" - private[this] val CumulativeDemandManifest = "B" - private[this] val RemoteSinkFailureManifest = "C" - private[this] val RemoteSinkCompletedManifest = "D" - private[this] val SourceRefManifest = "E" - private[this] val SinkRefManifest = "F" - private[this] val OnSubscribeHandshakeManifest = "G" - private[this] val AckManifest = "H" + private lazy val serialization = SerializationExtension(system) + + private val SequencedOnNextManifest = "A" + private val CumulativeDemandManifest = "B" + private val RemoteSinkFailureManifest = "C" + private val RemoteSinkCompletedManifest = "D" + private val SourceRefManifest = "E" + private val SinkRefManifest = "F" + private val OnSubscribeHandshakeManifest = "G" + private val AckManifest = "H" override def manifest(o: AnyRef): String = o match { // protocol 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 8e97e885408..d74ff8e1e10 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 @@ -241,7 +241,7 @@ object GraphStageLogic { def ref: ActorRef = functionRef @volatile - private[this] var behavior = initialReceive + private var behavior = initialReceive /** INTERNAL API */ private[pekko] def internalReceive(pack: (ActorRef, Any)): Unit = { @@ -376,7 +376,7 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount: /** * INTERNAL API */ - private[this] var _interpreter: GraphInterpreter = _ + private var _interpreter: GraphInterpreter = _ /** * INTERNAL API diff --git a/stream/src/main/scala/org/apache/pekko/stream/stage/StageLogging.scala b/stream/src/main/scala/org/apache/pekko/stream/stage/StageLogging.scala index dcc12cbd8fb..ff68a4cc871 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/stage/StageLogging.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/stage/StageLogging.scala @@ -30,7 +30,7 @@ import pekko.stream.MaterializerLoggingProvider * these decisions have to be handled by the operator itself. */ trait StageLogging { self: GraphStageLogic => - private[this] var _log: LoggingAdapter = _ + private var _log: LoggingAdapter = _ /** Override to customise reported log source */ protected def logSource: Class[?] = this.getClass