[incubator-pekko] branch main updated: =str Remove `futureCB` in MapAsyncUnordered.

[he...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

hepin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko.git


The following commit(s) were added to refs/heads/main by this push:
     new 9f2a9e657a =str Remove `futureCB` in MapAsyncUnordered.
9f2a9e657a is described below

commit 9f2a9e657a8e3c7281b80f29e0ba4163084e1859
Author: He-Pin <he...@alibaba-inc.com>
AuthorDate: Sat Sep 2 19:51:27 2023 +0800

    =str Remove `futureCB` in MapAsyncUnordered.
---
 .../src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala  | 8 +++-----
 1 file changed, 3 insertions(+), 5 deletions(-)

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 6995ae114c..e3c20d7e95 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
@@ -1402,13 +1402,14 @@ private[stream] object Collect {
     new GraphStageLogic(shape) with InHandler with OutHandler {
       override def toString = s"MapAsyncUnordered.Logic(inFlight=$inFlight, buffer=$buffer)"
 
-      val decider =
+      private val decider =
         inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
 
       private var inFlight = 0
       private var buffer: BufferImpl[Out] = _
+      private val invokeFutureCB: Try[Out] => Unit = getAsyncCallback(futureCompleted).invoke
 
-      private[this] def todo = inFlight + buffer.used
+      private[this] def todo: Int = inFlight + buffer.used
 
       override def preStart(): Unit = buffer = BufferImpl(parallelism, inheritedAttributes)
 
@@ -1432,9 +1433,6 @@ private[stream] object Collect {
         }
       }
 
-      private val futureCB = getAsyncCallback(futureCompleted)
-      private val invokeFutureCB: Try[Out] => Unit = futureCB.invoke
-
       override def onPush(): Unit = {
         try {
           val future = f(grab(in))


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org