title: Akka streams Source.actorRef vs Source.queue vs buffer, which one to use?
body: <p>I am using akka-streams-kafka to created a stream consumer from a kafka topic. 
  Using broadcast to serve events from kafka topic to web socket clients.</p>
  
  <p>I have found following three approaches to create a stream Source. 
  Question:
   My goal is to serve hundreds/thousands of websocket clients (some of which might be slow consumers). Which approach scales better?</p>
  
  <p>Appreciate any thoughts?
  Broadcast lowers the rate down to slowest consumer.<br>
  BUFFER_SIZE = 100000</p>
  
  <ol>
  <li><p>Source.ActorRef (source actor does not support backpressure option)</p>
  
  <pre><code>val kafkaSourceActorWithBroadcast = {
  
    val (sourceActorRef, kafkaSource) = Source.actorRef[String](BUFFER_SIZE, OverflowStrategy.fail)
                                        .toMat(BroadcastHub.sink(bufferSize = 256))(Keep.both).run
  
    Consumer.plainSource(consumerSettings, 
               Subscriptions.topics(KAFKA_TOPIC))
       .runForeach(record =&gt; sourceActorRef ! Util.toJson(record.value()))
  
  kafkaSource
  }
  </code></pre></li>
  <li><p>Source.queue</p>
  
  <pre><code>val kafkaSourceQueueWithBroadcast = {
  
    val (futureQueue, kafkaQueueSource) = Source.queue[String](BUFFER_SIZE, OverflowStrategy.backpressure)
                                          .toMat(BroadcastHub.sink(bufferSize = 256))(Keep.both).run
  
  Consumer.plainSource(consumerSettings, Subscriptions.topics(KAFKA_TOPIC))
          .runForeach(record =&gt; futureQueue.offer(Util.toJson(record.value())))
  
  kafkaQueueSource
  }
  </code></pre></li>
  <li><p>buffer</p>
  
  <pre><code>val kafkaSourceWithBuffer = Consumer.plainSource(consumerSettings, Subscriptions.topics(KAFKA_TOPIC))
                                .map(record =&gt; Util.toJson(record.value()))
                                .buffer(BUFFER_SIZE, OverflowStrategy.backpressure)
                                .toMat(BroadcastHub.sink(bufferSize = 256))(Keep.right).run
  </code></pre></li>
  </ol>
  
  <p>Websocket route code for completeness: </p>
  
  <pre><code>val streamRoute =
      path("stream") {
        handleWebSocketMessages(websocketFlow)
      }
  
  
  def websocketFlow(where: String): Flow[Message, Message, NotUsed] = {
  
    Flow[Message]
     .collect {
      case TextMessage.Strict(msg) =&gt; Future.successful(msg)
      case TextMessage.Streamed(stream) =&gt;
        stream.runFold("")(_ + _).flatMap(msg =&gt; Future.successful(msg))
    }
    .mapAsync(parallelism = PARALLELISM)(identity)
    .via(logicStreamFlow)
    .map { msg: String =&gt; TextMessage.Strict(msg) }
  }
  
  
  private def logicStreamFlow: Flow[String, String, NotUsed] = 
           Flow.fromSinkAndSource(Sink.ignore, kafkaSourceActorWithBroadcast)
  </code></pre>