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

chore: Avoid forwarding method on ArrayDequeue in stream module. (#1687)

Browse source
This commit is contained in:
He-Pin(kerr) 2025-01-06 10:45:35 +08:00 • committed by GitHub
parent 7782cf55e8
commit c953f50ded
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 6 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

6
stream/src/main/scala/org/apache/pekko/stream/impl/ActorRefBackpressureSinkStage.scala
View file

@ -45,7 +45,7 @@ import pekko.stream.stage._
private val maxBuffer = inheritedAttributes.get[InputBuffer](InputBuffer(16, 16)).max
require(maxBuffer > 0, "Buffer size must be greater than 0")
private val buffer: util.Deque[In] = new util.ArrayDeque[In]()
private val buffer: util.ArrayDeque[In] = new util.ArrayDeque[In]()
private var acknowledgementReceived = false
private var completeReceived = false
private var completionSignalled = false
@ -75,7 +75,7 @@ import pekko.stream.stage._
}
private def dequeueAndSend(): Unit = {
ref ! messageAdapter(self)(buffer.poll())
ref ! messageAdapter(self)(buffer.pollFirst())
}
private def finish(): Unit = {
@ -85,7 +85,7 @@ import pekko.stream.stage._
}
def onPush(): Unit = {
buffer.offer(grab(in))
buffer.offerLast(grab(in))
if (acknowledgementReceived) {
dequeueAndSend()
acknowledgementReceived = false

2
stream/src/main/scala/org/apache/pekko/stream/impl/PhasedFusingActorMaterializer.scala
View file

@ -524,7 +524,7 @@ private final case class SavedIslandData(
if (Debug) println(s"PUSH: $matValue => $matValueStack")
case Concat(first, next) =>
if (next ne EmptyTraversal) traversalStack.add(next)
if (next ne EmptyTraversal) traversalStack.addLast(next)
nextStep = first
case Pop =>
val popped = matValueStack.removeLast()

4
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
View file

@ -793,7 +793,7 @@ import org.reactivestreams.Subscription
else if (currentLimit == 0) {
self ! Resume
} else {
shortCircuitBuffer.poll() match {
shortCircuitBuffer.pollFirst() match {
case b: BoundaryEvent => processEvent(b)
case Resume => finishShellRegistration()
case unexpected =>
@ -842,7 +842,7 @@ import org.reactivestreams.Subscription
override def postStop(): Unit = {
if (shortCircuitBuffer ne null) {
while (!shortCircuitBuffer.isEmpty) {
shortCircuitBuffer.poll() match {
shortCircuitBuffer.pollFirst() match {
case b: BoundaryEvent =>
// signal to telemetry that this event won't be processed
b.cancel()