This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 4749eda97f refactor: replace private[this] with private and = _ with
explicit defaults for Scala 3.8 (#3072)
4749eda97f is described below
commit 4749eda97ffcc20f8fb576d8e064e31f60957a6d
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Wed Jun 17 23:40:19 2026 +0800
refactor: replace private[this] with private and = _ with explicit defaults
for Scala 3.8 (#3072)
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
---
.../actor/typed/internal/ActorContextImpl.scala | 2 +-
.../pekko/actor/typed/javadsl/Behaviors.scala | 4 +-
.../pekko/actor/typed/scaladsl/AskPattern.scala | 2 +-
.../main/scala/org/apache/pekko/actor/Actor.scala | 2 +-
.../scala/org/apache/pekko/actor/ActorCell.scala | 6 +--
.../scala/org/apache/pekko/actor/ActorRef.scala | 4 +-
.../org/apache/pekko/actor/ActorRefProvider.scala | 4 +-
.../scala/org/apache/pekko/actor/ActorSystem.scala | 8 ++--
.../main/scala/org/apache/pekko/actor/FSM.scala | 6 +--
.../apache/pekko/actor/IndirectActorProducer.scala | 2 +-
.../pekko/actor/LightArrayRevolverScheduler.scala | 6 +--
.../main/scala/org/apache/pekko/actor/Props.scala | 6 +--
.../apache/pekko/actor/RepointableActorRef.scala | 10 ++---
.../apache/pekko/dispatch/AbstractDispatcher.scala | 4 +-
.../apache/pekko/dispatch/BatchingExecutor.scala | 10 ++---
.../scala/org/apache/pekko/dispatch/Mailbox.scala | 6 +--
.../pekko/dispatch/affinity/AffinityPool.scala | 28 +++++++-------
.../pekko/dispatch/sysmsg/SystemMessage.scala | 2 +-
.../scala/org/apache/pekko/event/Logging.scala | 2 +-
.../org/apache/pekko/io/DirectByteBufferPool.scala | 4 +-
.../org/apache/pekko/io/SelectionHandler.scala | 12 +++---
actor/src/main/scala/org/apache/pekko/io/Tcp.scala | 4 +-
.../scala/org/apache/pekko/io/TcpConnection.scala | 14 +++----
actor/src/main/scala/org/apache/pekko/io/Udp.scala | 2 +-
.../org/apache/pekko/pattern/AskSupport.scala | 20 +++++-----
.../org/apache/pekko/pattern/CircuitBreaker.scala | 12 +++---
.../apache/pekko/serialization/Serialization.scala | 2 +-
.../apache/pekko/serialization/Serializer.scala | 4 +-
.../scala/org/apache/pekko/util/ByteString.scala | 6 +--
.../scala/org/apache/pekko/util/Collections.scala | 4 +-
.../scala/org/apache/pekko/util/ConstantFun.scala | 2 +-
.../org/apache/pekko/util/DoubleLinkedList.scala | 6 +--
.../org/apache/pekko/util/FrequencySketch.scala | 40 ++++++++++----------
.../org/apache/pekko/util/ImmutableIntMap.scala | 6 +--
.../apache/pekko/util/StablePriorityQueue.scala | 2 +-
.../org/apache/pekko/util/SubclassifiedIndex.scala | 2 +-
.../scala/org/apache/pekko/util/TokenBucket.scala | 4 +-
.../apache/pekko/stream/FusedGraphsBenchmark.scala | 4 +-
.../apache/pekko/stream/RangeSourceBenchmark.scala | 2 +-
.../pekko/util/FastFrequencySketchBenchmark.scala | 10 ++---
.../pekko/util/FrequencySketchBenchmark.scala | 10 ++---
.../apache/pekko/util/ImmutableIntMapBench.scala | 12 +++---
.../typed/internal/ClusterShardingImpl.scala | 2 +-
.../journal/leveldb/LeveldbCompaction.scala | 2 +-
.../persistence/journal/SteppingInmemJournal.scala | 2 +-
.../aeron/AeronStreamMaxThroughputSpec.scala | 4 +-
.../pekko/remote/artery/ArteryTransport.scala | 14 +++----
.../apache/pekko/remote/artery/Association.scala | 20 +++++-----
.../pekko/remote/artery/EnvelopeBufferPool.scala | 2 +-
.../pekko/remote/artery/LruBoundedCache.scala | 12 +++---
.../artery/aeron/ArteryAeronUdpTransport.scala | 10 ++---
.../pekko/remote/artery/aeron/TaskRunner.scala | 8 ++--
.../artery/compress/DecompressionTable.scala | 4 +-
.../artery/compress/InboundCompressions.scala | 18 ++++-----
.../remote/artery/compress/TopHeavyHitters.scala | 10 ++---
.../remote/transport/netty/NettyTransport.scala | 2 +-
.../pekko/remote/artery/MetadataCarryingSpec.scala | 2 +-
.../scaladsl/ZipLatestWithApply.scala.template | 2 +-
.../scala/org/apache/pekko/stream/KillSwitch.scala | 8 ++--
.../scala/org/apache/pekko/stream/impl/FanIn.scala | 26 ++++++-------
.../pekko/stream/impl/JsonObjectParser.scala | 14 +++----
.../impl/PhasedFusingActorMaterializer.scala | 6 +--
.../impl/ResizableMultiReaderRingBuffer.scala | 8 ++--
.../pekko/stream/impl/SinkholeSubscriber.scala | 2 +-
.../scala/org/apache/pekko/stream/impl/Sinks.scala | 4 +-
.../pekko/stream/impl/SubscriberManagement.scala | 10 ++---
.../pekko/stream/impl/TraversalBuilder.scala | 4 +-
.../org/apache/pekko/stream/impl/Unfold.scala | 10 ++---
.../stream/impl/fusing/ActorGraphInterpreter.scala | 2 +-
.../stream/impl/fusing/AggregateWithBoundary.scala | 2 +-
.../stream/impl/fusing/GraphInterpreter.scala | 32 ++++++++--------
.../org/apache/pekko/stream/impl/fusing/Ops.scala | 10 ++---
.../pekko/stream/impl/fusing/RangeSource.scala | 10 ++---
.../pekko/stream/impl/fusing/StreamOfStreams.scala | 4 +-
.../pekko/stream/impl/io/ByteStringParser.scala | 2 +-
.../stream/impl/io/InputStreamSinkStage.scala | 10 ++---
.../stream/impl/io/OutputStreamSourceStage.scala | 2 +-
.../io/compression/DeflateDecompressorBase.scala | 2 +-
.../pekko/stream/impl/streamref/SinkRefImpl.scala | 10 ++---
.../stream/impl/streamref/SourceRefImpl.scala | 16 ++++----
.../stream/impl/streamref/StreamRefsMaster.scala | 4 +-
.../org/apache/pekko/stream/javadsl/BidiFlow.scala | 2 +-
.../pekko/stream/javadsl/DelayStrategy.scala | 2 +-
.../org/apache/pekko/stream/javadsl/Graph.scala | 4 +-
.../apache/pekko/stream/scaladsl/BidiFlow.scala | 2 +-
.../pekko/stream/scaladsl/DelayStrategy.scala | 2 +-
.../org/apache/pekko/stream/scaladsl/Flow.scala | 2 +-
.../org/apache/pekko/stream/scaladsl/Graph.scala | 4 +-
.../org/apache/pekko/stream/scaladsl/Hub.scala | 44 +++++++++++-----------
.../org/apache/pekko/stream/scaladsl/Sink.scala | 4 +-
.../org/apache/pekko/stream/scaladsl/Source.scala | 4 +-
.../stream/serialization/StreamRefSerializer.scala | 20 +++++-----
.../org/apache/pekko/stream/stage/GraphStage.scala | 4 +-
.../apache/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 6f2e98b962..7f3842dff3 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 78a3d5314e..bc3ee7b8a0 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 5ab409bccb..0c811f764b 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 28e0b5e65a..404c7e40e0 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 7f853edd8e..7d36e05889 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 abd9d06af6..9f34739009 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 472adfbf4a..8655e38bf8 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 cc0ae959a9..9c5accd751 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 49aef7c78b..fd93a273b7 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 fa6ec6f4a5..39dec1dae9 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 eb7ee9921d..2d4e409501 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 69f2ef1f79..2d8c4abf46 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 f9d300d9b7..0e9e562d81 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 a0ed462f10..f5e9e277dc 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 d0f0ec0fa1..93d278a00a 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 6dc048430e..af728fabf9 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 35d10b36ca..4262c2ae61 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 1e7b3e2ae1..803f81c685 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 7a1ef23063..2cfefc1f27 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 50595a4d0a..6331b9ebec 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 338b823b87..c6e2f869cf 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 38ec52095f..70d7ced59c 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 cc8f494704..839dfcfff1 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 f2cb936ee0..995f463f4f 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 6de334f1f4..8c85c4eff0 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 7b9f81f0df..c64f76909d 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 6a0301af64..9515ec1b23 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 01cc91c381..39275ef6a7 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 44db0e80c7..57c7762ef4 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 12f9114959..9311c3dadc 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 0a686df677..8cf9f5d186 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 bf2c79b22c..cb56ba06b4 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 f7c7339341..4f56d77811 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 d53b72296a..8405e0fa07 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 594877e9e5..4bb863aa8d 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 256ff53c35..477f386d1f 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 a07a2866c8..e7fb53d98d 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 fd7425525f..5fe6676060 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 59925f9612..eb53966012 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 3dd27b0a6c..38f6ed4d73 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 a92b0defdd..f89bdd9b18 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 6e385202fe..fc3779f6b3 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 98406919d6..df01f21032 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 b21f920544..46cd1789ff 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 9ff0473f8e..0a228b5807 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 9e78a90f3c..6c32947673 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 b9bcfd45ef..ba3f8e650b 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 d2e543414d..eb9bf1176a 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 5da684ebe0..267c33cd24 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 669ae73b35..e145a7e1c9 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 be3f89c477..93160b85ba 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 a757de2858..706b72b51c 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 1279eddf0b..575725ff25 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 4238af2490..dbb43678a8 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 be6e27f532..c5458c9ed6 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 6cad95e682..f64a39ac02 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 a1f50c4e92..32aa6c9bb0 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 80cc5d9baa..d513e5ea63 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 cd120bf0e1..018538fa01 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 32cc959efe..ea16196600 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 6f00d61caa..369835419c 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 0a3861984e..9445ae796b 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 fc215f2dab..1e5c684c78 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 4fd02e2162..2e43555e3c 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 de215da3ea..5d61e9f5bf 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 91c59a6bff..c44b3c43b7 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 d4a7da2074..658f668890 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 44cf73d6d7..9bcd869b5c 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 61eb9b8a29..a7156ff29e 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 79b547c4db..7d886a8ed0 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 aedcd0b301..edaa45f31f 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 ba21468469..3832ccf269 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 4e9cad700e..e8078e3854 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 1d7ffff8b4..882e717f5d 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 bef6e0e7d4..bd81867026 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 5e6be230fd..be3d18e784 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 d0dbb81c27..a4904761a5 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 8c523fd4c2..922bdf7ba9 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 67121c7421..53b8a89c8d 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 f5cf889272..4e392cc35b 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 f2be78e206..0bd307f85e 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 cc9931759e..b7e0c9b2dd 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 993b0c81a2..9f4addc0a8 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 0a919fde2a..ba9264a8b9 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 9b84b0af80..47598bf82c 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 549a158394..1323c344e7 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 59e524fc4c..4ae16e46ce 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 16b9a605c0..6cde3ffa9f 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 8849529e6d..32e647841f 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 2fa336579b..d4e78bebf9 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 39b8155526..4cd4594e1d 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 daa66f753c..5badc9b2a5 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 8a8168ff26..dcd6faab5f 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
@@ -275,7 +275,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 = {
@@ -539,7 +539,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 dcc12cbd8f..ff68a4cc87 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
