akka/akka-core
Vedi su GitHubSource.fromPublisher does not pick up attributes correctly
Open
#30.076 aperta il 3 mar 2021
1 - triagedhelp wantedt:stream
Metriche repository
- Star
- (13.277 star)
- Metriche merge PR
- (Merge medio 17h 35m) (11 PR mergiate in 30 g)
Descrizione
Looks like Source.fromPublisher buffers some data from the source. I couldn't find any way to limit or disable the buffer size.
val source3 = Source.fromPublisher {
Source.fromIterator(() => Iterator.from(1))
.takeWhile(_ < 50)
.map { i =>
println(s"internal akka: $i")
i
}
.runWith(Sink.asPublisher(false))
}.map { i =>
Thread.sleep(2000)
println(s"akka: $i")
i
}
log:
internal akka: 1
internal akka: 2
internal akka: 3
internal akka: 4
internal akka: 5
internal akka: 6
internal akka: 7
internal akka: 8
internal akka: 9
akka: 1
akka: 2
akka: 3
.
.
.
.
Originally I faced with this issue converting from Monix reactive streams to Akka-streams.