title: How to send a realtime count of active WebSocket connections over an Akka HTTP WebSocket? (works with just Akka Streams)
body: <p>How can a realtime count of active WebSocket connections be sent over an Akka WebSocket? It seems Akka HTTP WebSockets change the running of Akka Stream's prepend stage.</p>
  
  <p>In the first code block below, the count is only sent immediately when it is the only connection. Any clients which connect when there is already a count more than 1 do not receive an update until the <strong>next</strong> client connects.</p>
  
  <p>The second code block has the WebSocket code removed so that only Akka Streams are used, and every stream subscription <strong>does</strong> receive an immediate count as shown on stdout.</p>
  
  <h2>Using Akka HTTP WebSocket</h2>
  
  <p>Running this code and accessing <a href="http://localhost:8080" rel="nofollow noreferrer">http://localhost:8080</a> with multiple browser windows shows this in their browser consoles.</p>
  
  <pre><code>import akka.actor.ActorSystem
  import akka.http.scaladsl.Http
  import akka.http.scaladsl.model.ws.TextMessage
  import akka.http.scaladsl.model.{ContentTypes, HttpEntity}
  import akka.http.scaladsl.server.Directives._
  import akka.http.scaladsl.server.Route
  import akka.stream.scaladsl.{BroadcastHub, Flow, Keep, Sink, Source}
  import akka.stream.{ActorMaterializer, OverflowStrategy}
  
  import scala.concurrent.ExecutionContextExecutor
  
  object CounterFlow {
  
      private implicit val actorSystem: ActorSystem = ActorSystem("CounterFlowTest")
      private implicit val materializer: ActorMaterializer = ActorMaterializer()
      private implicit val executionContext: ExecutionContextExecutor = actorSystem.dispatcher
  
      val (counterSourceQueue, counterSource) =
          Source.queue[Int](0, OverflowStrategy.backpressure)
              .conflate(_ + _)
              .scan(0)(_ + _)
              .toMat(BroadcastHub.sink(bufferSize = 1))(Keep.both)
              .run()
  
      val clientFlow: Source[TextMessage.Strict, Unit] =
          counterSource
              .map(_.toString)
              .map(TextMessage.Strict)
              .prepend(Source.fromIterator(() =&gt; {
                  counterSourceQueue.offer(1)
                  Iterator.empty
              }))
              .watchTermination()((_, done) =&gt; done.foreach(_ =&gt; counterSourceQueue.offer(-1)))
  
      def main(args: Array[String]): Unit = {
  
          val route: Route =
              pathEndOrSingleSlash {
                  get {
                      complete(HttpEntity(ContentTypes.`text/html(UTF-8)`,
                          "&lt;h1&gt;Check browser console for connection count&lt;/h1&gt;" +
                              "&lt;script&gt;const ws = new WebSocket(\"ws://localhost:8080/ws\");ws.onmessage = e =&gt; console.log(e.data);&lt;/script&gt;"))
                  }
              } ~
                  path("ws") {
                      handleWebSocketMessages(
                          Flow.fromSinkAndSourceCoupled(
                              Sink.ignore,
                              clientFlow))
                  }
  
          Http().bindAndHandle(route, "0.0.0.0", 8080)
          () // discard non-Unit value
      }
  }
  </code></pre>
  
  <h2>Akka Streams Only</h2>
  
  <p>As seen on stdout, each client receives an immediate count as desired.</p>
  
  <pre><code>import akka.actor.ActorSystem
  import akka.stream.scaladsl.{BroadcastHub, Keep, Sink, Source}
  import akka.stream.{ActorMaterializer, OverflowStrategy}
  
  import scala.concurrent.ExecutionContext.Implicits.global
  
  object CounterFlow {
  
      private implicit val actorSystem: ActorSystem = ActorSystem("CounterFlowTEst")
      private implicit val materializer: ActorMaterializer = ActorMaterializer()
  
      val (counterSourceQueue, counterSource) = Source.queue[Int](0, OverflowStrategy.backpressure)
          .conflate(_ + _)
          .scan(0)(_ + _)
          .toMat(BroadcastHub.sink)(Keep.both)
          .run()
  
      def subscribeClient(clientName: String): Unit =
          counterSource
              .prepend(Source.fromIterator(() =&gt; {
                  counterSourceQueue.offer(1)
                  Iterator.empty
              }))
              .watchTermination()((_, done) =&gt; done.foreach(_ =&gt; counterSourceQueue.offer(-1)))
              .runWith(Sink.foreach(msg =&gt; println(s"$clientName $msg")))
              .foreach(_ =&gt; println(s"$clientName Done"))
  
      def main(args: Array[String]): Unit = {
          subscribeClient("A")
          subscribeClient("B")
          Thread.sleep(500L)
          subscribeClient("C")
      }
  }
  </code></pre>