Source.queue and SourceRef. How to decrease queue buffer?

[pbernet (Paul)]

Hmm, did you try with bufferSize = 0?

From the Doc:
The buffer can be disabled by using bufferSize of 0 and then received message will wait for downstream demand unless there is another message waiting for downstream demand, in that case offer result will be completed according to the overflow strategy.

See also this discussion:

Source.queue bug?

My code is below val (chromeQueue, chromeSources) = Source .queue[Int](1, OverflowStrategy.dropNew) .preMaterialize() val (orderQueue, orderSources) = Source .queue[String](1, OverflowStrategy.dropNew) .preMaterialize() orderSources .zip(chromeSources) .to(Sink.ignore) .run() Source(Seq("a", "b", "c", "d")) .mapAsync(1) { i => orderQueue.offer(i).map { case result: QueueCompletionResult => println("completion") case QueueOfferResult.Enqueued => …