akka-http: send element to akka sink from http route

Question

How can I send elements/messages to an Akka Sink from an Akka HTTP route? My HTTP route still needs to return a normal HTTP response.

I imagine this requires a stream branch/junction. The normal HTTP routes are flows from HttpRequest -> HttpResponse. I would like to add a branch/junction so that HttpRequests can trigger events to my separate sink as well as generate the normal HttpResponse.

Below is a very simple single route akka-http app. For simplicity, I'm using a simple println sink. My production use case, will obviously involve a less trivial sink.

def main(args: Array[String]): Unit = {
  implicit val actorSystem = ActorSystem("my-akka-http-test")
  val executor = actorSystem.dispatcher
  implicit val materializer = ActorMaterializer()(actorSystem)

  // I would like to send elements to this sink in response to HTTP GET operations.
  val sink: Sink[Any, Future[Done]] = Sink.foreach(println)

  val route: akka.http.scaladsl.server.Route =
    path("hello" / Segment) { p =>
      get {
        // I'd like to send a message to an Akka Sink as well as return an HTTP response.
        complete {
          s"<h1>Say hello to akka-http. p=$p</h1>"
        }
      }
    }

  val httpExt: akka.http.scaladsl.HttpExt = Http(actorSystem)
  val bindingFuture = httpExt.bindAndHandle(RouteResult.route2HandlerFlow(route), "localhost", 8080)

  println("Server online at http://localhost:8080/")
  println("Press RETURN to stop...")
  scala.io.StdIn.readLine()

  bindingFuture
    .flatMap(_.unbind())(executor) // trigger unbinding from the port
    .onComplete(_ => Await.result(actorSystem.terminate(), Duration.Inf))(executor) // and shutdown when done
}

EDIT: Or in using the low-level akka-http API, how could I send specific messages to a sink from a specific route handler?

def main(args: Array[String]): Unit = {
  implicit val actorSystem = ActorSystem("my-akka-http-test")
  val executor = actorSystem.dispatcher
  implicit val materializer = ActorMaterializer()(actorSystem)

  // I would like to send elements to this sink in response to HTTP GET operations.
  val sink: Sink[Any, Future[Done]] = Sink.foreach(println)

  val requestHandler: HttpRequest => HttpResponse = {
    case HttpRequest(GET, Uri.Path("/"), _, _, _) =>
      HttpResponse(entity = HttpEntity(
        ContentTypes.`text/html(UTF-8)`,
        "<html><body>Hello world!</body></html>"))

    case HttpRequest(GET, Uri.Path("/ping"), _, _, _) =>
      HttpResponse(entity = "PONG!")

    case HttpRequest(GET, Uri.Path("/crash"), _, _, _) =>
      sys.error("BOOM!")

    case r: HttpRequest =>
      r.discardEntityBytes() // important to drain incoming HTTP Entity stream
      HttpResponse(404, entity = "Unknown resource!")
  }

  val serverSource = Http().bind(interface = "localhost", port = 8080)

  val bindingFuture: Future[Http.ServerBinding] =
    serverSource.to(Sink.foreach { connection =>
      println("Accepted new connection from " + connection.remoteAddress)

      connection handleWithSyncHandler requestHandler
      // this is equivalent to
      // connection handleWith { Flow[HttpRequest] map requestHandler }
    }).run()

  println("Server online at http://localhost:8080/")
  println("Press RETURN to stop...")
  scala.io.StdIn.readLine()

  bindingFuture
    .flatMap(_.unbind())(executor) // trigger unbinding from the port
    .onComplete(_ => Await.result(actorSystem.terminate(), Duration.Inf))(executor) // and shutdown when done
}

Stefano Bonetti Stefano Bonetti · Accepted Answer · 2017-02-24T17:01:27

IF you want to send the whole HttpRequest to a sink of yours, I'd say the simplest way is to use the alsoTo combinator. The result would be something along the lines of

val mySink: Sink[HttpRequest, NotUsed] = ???

val handlerFlow = Flow[HttpRequest].alsoTo(mySink).via(RouteResult.route2HandlerFlow(route))

val bindingFuture = Http().bindAndHandle(handlerFlow, "localhost", 8080)

FYI: alsoTo in fact hides a Broadcast stage.

IF instead you need to selectively send a message to a Sink from a specific subroute, you have no other choice but to materialize a new flow for each incoming request. See example below

val sink: Sink[Any, Future[Done]] = Sink.foreach(println)

val route: akka.http.scaladsl.server.Route =
  path("hello" / Segment) { p =>
    get {

      (extract(_.request) & extractMaterializer) { (req, mat) ⇒
        Source.single(req).runWith(sink)(mat)

        complete {
          s"<h1>Say hello to akka-http. p=$p</h1>"
        }
      }
    }
  }

Also, keep in mind you can always ditch the high-level DSL completely, and model you whole route using the lower-level streams DSL. This will result in more verbose code - but will give you full control of your stream materialization.

EDIT: example below

val sink: Sink[Any, Future[Done]] = Sink.foreach(println)

val handlerFlow =
  Flow.fromGraph(GraphDSL.create() { implicit b =>
    import GraphDSL.Implicits._

    val partition = b.add(Partition[HttpRequest](2, {
      case HttpRequest(GET, Uri.Path("/"), _, _, _) ⇒ 0
      case _                                        ⇒ 1
    }))
    val merge = b.add(Merge[HttpResponse](2))

    val happyPath = Flow[HttpRequest].map{ req ⇒
      HttpResponse(entity = HttpEntity(
        ContentTypes.`text/html(UTF-8)`,
        "<html><body>Hello world!</body></html>"))
    }        

    val unhappyPath = Flow[HttpRequest].map{
      case HttpRequest(GET, Uri.Path("/ping"), _, _, _) =>
      HttpResponse(entity = "PONG!")

      case HttpRequest(GET, Uri.Path("/crash"), _, _, _) =>
      sys.error("BOOM!")

      case r: HttpRequest =>
        r.discardEntityBytes() // important to drain incoming HTTP Entity stream
        HttpResponse(404, entity = "Unknown resource!")
    }

    partition.out(0).alsoTo(sink) ~> happyPath   ~> merge
    partition.out(1)              ~> unhappyPath ~> merge

    FlowShape(partition.in, merge.out)
  })

val bindingFuture = Http().bindAndHandle(handlerFlow, "localhost", 8080)

akka-http: send element to akka sink from http route

2 Answers