He-Pin commented on code in PR #1870:
URL: https://github.com/apache/pekko/pull/1870#discussion_r2108085600
##########
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:
1. we need to support the old `f: () => Out => IterableOnce[T]`, and the
`f`'s result needs to be adapted to `StatefulMapConcatAccumulator` and add the
`onComplete` method, jvm's jit will do the inline
--
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
