queimadus commented on code in PR #1623:
URL: https://github.com/apache/pekko/pull/1623#discussion_r1894909443
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/StreamOfStreams.scala:
##########
@@ -237,7 +239,9 @@ import pekko.util.ccompat.JavaConverters._
override def onUpstreamFinish(): Unit = {
if (!prefixComplete) {
// This handles the unpulled out case as well
- emit(out, (builder.result(), Source.empty), () => completeStage())
+ val prefix = builder.result();
+ builder = null // free for GC
+ emit(out, (prefix, Source.empty), () => completeStage())
Review Comment:
I actually can't seem to reproduce the issue with the [example I
mentioned](https://github.com/apache/pekko/discussions/1566) with
`flatMapPrefix` with `prefixAndTail`.
E.g. running the following runs fine with and without these changes
```scala
val s = Source
.repeat(())
.map(_ => ByteString('a' * 4000000))
.take(1000000)
.prefixAndTail(50000)
.flatMapConcat { case (prefix, tail) => tail }
Source.empty
.concatAllLazy(List.tabulate(30000)(_ => s): _*)
.runWith(Sink.ignore).onComplete(println(_))
```
What doesn't run fine is the the [following
snippet](https://github.com/apache/pekko/pull/1622#discussion_r1894903082), but
it may be a different problem.
```scala
def myLogic(prefix: Seq[ByteString]): Flow[ByteString, ByteString, NotUsed]
=
Flow[ByteString]
val s = Source
.repeat(())
.map(_ => ByteString('a' * 400000))
.take(1000000)
.prefixAndTail(50000)
.flatMapConcat { case (prefix, tail) =>
Source(prefix).concatLazy(tail).via(myLogic(prefix))
}
Source.empty
.concatAllLazy(List.tabulate(30000)(_ => s): _*)
.runWith(Sink.ignore).onComplete(println(_))
```
That said, these changes still seem to make sense.
--
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
