akka/akka-core

Source.fromPublisher does not pick up attributes correctly

開放

#30,076 建立於 2021年3月3日

 (12 則留言) (0 個反應) (0 位負責人)Scala (3,547 個分叉)batch import
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.

貢獻者指南