pjfanning commented on code in PR #1870:
URL: https://github.com/apache/pekko/pull/1870#discussion_r2106192399
##########
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 looks like an overhead applied to all users of StatefulMapConcat - is
there any way to make this optional - something that is only created if a user
opts in to it?
--
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
