He-Pin commented on code in PR #1870:
URL: https://github.com/apache/pekko/pull/1870#discussion_r2108078031
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala:
##########
@@ -2337,13 +2338,25 @@ private[pekko] final class StatefulMapConcat[In,
Out](val f: () => In => Iterabl
override def initialAttributes: Attributes =
DefaultAttributes.statefulMapConcat and SourceLocation.forLambda(f)
+ private def adaptToAccumulator: StatefulMapConcatAccumulator[In, Out] = new
StatefulMapConcatAccumulator[In, Out] {
+ private val mapper = f()
+ final override def apply(in: In): IterableOnce[Out] = mapper(in)
+ final override def onComplete(): IterableOnce[Out] = mapper match {
+ case acc: StatefulMapConcatAccumulator[In, Out] @unchecked =>
acc.onComplete()
+ case acc: pekko.stream.javadsl.StatefulMapConcatAccumulator[In, Out]
@unchecked =>
+ import pekko.util.ccompat.JavaConverters._
+ acc.onComplete().asScala
+ case _ => Nil
+ }
+ }
+
def createLogic(inheritedAttributes: Attributes) = new
GraphStageLogic(shape) with InHandler with OutHandler {
- lazy val decider =
inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
- var currentIterator: Iterator[Out] = _
- var plainFun = f()
- val contextPropagation = ContextPropagation()
+ private lazy val decider =
inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
+ private var currentIterator: Iterator[Out] = _
+ private var accumulator: StatefulMapConcatAccumulator[In, Out] =
adaptToAccumulator
Review Comment:
This method is only been called once during materialization, introducing a
`StatefulMapConcatAccumulatorFacotry` will change more code, and this adoption
will be called anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@pekko.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscr...@pekko.apache.org
For additional commands, e-mail: notifications-h...@pekko.apache.org
