1 - triagedhelp wantedt:stream
仓库指标
- 星标
- (13,277 个星标)
- PR 合并指标
- (平均合并 8天 19小时) (30 天内合并 10 个 PR)
描述
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.