Source Streaming

Akka HTTP supports completing a request with an Akka Source<T, ?>Source[T, _], which makes it possible to easily build and consume streaming end-to-end APIs which apply back pressure throughout the entire stack.

It is possible to complete requests with raw Source<ByteString, ?>Source[ByteString, _], however often it is more convenient to stream on an element-by-element basis, and allow Akka HTTP to handle the rendering internally - for example as a JSON array, or CSV stream (where each element is separated by a newline).

In the following sections we investigate how to make use of the JSON Streaming infrastructure, however the general hints apply to any kind of element-by-element streaming you could imagine.

JSON Streaming

JSON Streaming is a term referring to streaming a (possibly infinite) stream of element as independent JSON objects as a continuous HTTP request or response. The elements are most often separated using newlines, however do not have to be. Concatenating elements side-by-side or emitting “very long” JSON array is also another use case.

In the below examples, we’ll be referring to the Tweet and Measurement case classes as our model, which are defined as:

Scala

case class Tweet(uid: Int, txt: String)
case class Measurement(id: String, value: Int)

Java

private static final class JavaTweet {
  private int id; 
  private String message;

  public JavaTweet(int id, String message) {
    this.id = id;
    this.message = message;
  }

  public int getId() {
    return id;
  }

  public void setId(int id) {
    this.id = id;
  }

  public void setMessage(String message) {
    this.message = message;
  }

  public String getMessage() {
    return message;
  }

}

Scala

object MyJsonProtocol
  extends akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
  with spray.json.DefaultJsonProtocol {

  implicit val tweetFormat = jsonFormat2(Tweet.apply)
  implicit val measurementFormat = jsonFormat2(Measurement.apply)
}

Java

final Unmarshaller<ByteString, JavaTweet> JavaTweets = Jackson.byteStringUnmarshaller(JavaTweet.class);

Responding with JSON Streams

In this example we implement an API representing an infinite stream of tweets, very much like Twitter’s Streaming API.

The final step is simply completing a request using a Source of tweets, as simple as that:

Scala

// [1] import "my protocol", for marshalling Tweet objects:
import MyJsonProtocol._

// [2] pick a Source rendering support trait:
// Note that the default support renders the Source as JSON Array
implicit val jsonStreamingSupport: JsonEntityStreamingSupport = EntityStreamingSupport.json()

val route =
  path("tweets") {
    // [3] simply complete a request with a source of tweets:
    val tweets: Source[Tweet, NotUsed] = getTweets
    complete(tweets)
  }

// tests ------------------------------------------------------------
val AcceptJson = Accept(MediaRange(MediaTypes.`application/json`))
val AcceptXml = Accept(MediaRange(MediaTypes.`text/xml`))

Get("/tweets").withHeaders(AcceptJson) ~> route ~> check {
  responseAs[String] shouldEqual
    """[""" +
    """{"uid":1,"txt":"#Akka rocks!"},""" +
    """{"uid":2,"txt":"Streaming is so hot right now!"},""" +
    """{"uid":3,"txt":"You cannot enter the same river twice."}""" +
    """]"""
}

// endpoint can only marshal Json, so it will *reject* requests for application/xml:
Get("/tweets").withHeaders(AcceptXml) ~> route ~> check {
  handled should ===(false)
  rejection should ===(UnacceptedResponseContentTypeRejection(Set(ContentTypes.`application/json`)))
}

Java

// Step 1: Enable JSON streaming
// we're not using this in the example, but it's the simplest way to start:
// The default rendering is a JSON array: `[el, el, el , ...]`
final JsonEntityStreamingSupport jsonStreaming = EntityStreamingSupport.json();

// Step 1.1: Enable and customise how we'll render the JSON, as a compact array:
final ByteString start = ByteString.fromString("[");
final ByteString between = ByteString.fromString(",");
final ByteString end = ByteString.fromString("]");
final Flow<ByteString, ByteString, NotUsed> compactArrayRendering = 
  Flow.of(ByteString.class).intersperse(start, between, end);

final JsonEntityStreamingSupport compactJsonSupport = EntityStreamingSupport.json()
  .withFramingRendererFlow(compactArrayRendering);


// Step 2: implement the route
final Route responseStreaming = path("tweets", () ->
  get(() ->
    parameter(StringUnmarshallers.INTEGER, "n", n -> {
      final Source<JavaTweet, NotUsed> tws =
        Source.repeat(new JavaTweet(12, "Hello World!")).take(n);
      
      // Step 3: call complete* with your source, marshaller, and stream rendering mode
      return completeOKWithSource(tws, Jackson.marshaller(), compactJsonSupport);
    })
  )
);
// tests:
final TestRoute routes = testRoute(tweets());

// test happy path
final Accept acceptApplication = Accept.create(MediaRanges.create(MediaTypes.APPLICATION_JSON));
routes.run(HttpRequest.GET("/tweets?n=2").addHeader(acceptApplication))
  .assertStatusCode(200)
  .assertEntity("[{\"id\":12,\"message\":\"Hello World!\"},{\"id\":12,\"message\":\"Hello World!\"}]");

// test responses to potential errors
final Accept acceptText = Accept.create(MediaRanges.ALL_TEXT);
routes.run(HttpRequest.GET("/tweets?n=3").addHeader(acceptText))
  .assertStatusCode(StatusCodes.NOT_ACCEPTABLE) // 406
  .assertEntity("Resource representation is only available with these types:\napplication/json");
// tests --------------------------------------------
final TestRoute routes = testRoute(csvTweets());

// test happy path
final Accept acceptCsv = Accept.create(MediaRanges.create(MediaTypes.TEXT_CSV));
routes.run(HttpRequest.GET("/tweets?n=2").addHeader(acceptCsv))
  .assertStatusCode(200)
  .assertEntity("12,Hello World!\n" +
    "12,Hello World!");

// test responses to potential errors
final Accept acceptText = Accept.create(MediaRanges.ALL_APPLICATION);
routes.run(HttpRequest.GET("/tweets?n=3").addHeader(acceptText))
  .assertStatusCode(StatusCodes.NOT_ACCEPTABLE) // 406
  .assertEntity("Resource representation is only available with these types:\ntext/csv; charset=UTF-8");

The reason the EntityStreamingSupportEntityStreamingSupport has to be enabled explicitly is that one might want to configure how the stream should be rendered. We’ll discuss this in depth in the next section though.

Consuming JSON Streaming uploads

Sometimes a client sends a streaming request. For example, an embedded device initiated a connection with the server and is feeding it with one line of measurement data.

In this example, we want to consume this data in a streaming fashion from the request entity and also apply back pressure to the underlying TCP connection should the server be unable to cope with the rate of incoming data. Back pressure is automatically applied thanks to Akka Streams.

Scala

// [1] import "my protocol", for unmarshalling Measurement objects:
import MyJsonProtocol._

// [2] enable Json Streaming
implicit val jsonStreamingSupport = EntityStreamingSupport.json()

// prepare your persisting logic here
val persistMetrics = Flow[Measurement]

val route =
  path("metrics") {
    // [3] extract Source[Measurement, _]
    entity(asSourceOf[Measurement]) { measurements =>
      // alternative syntax:
      // entity(as[Source[Measurement, NotUsed]]) { measurements =>
      val measurementsSubmitted: Future[Int] =
        measurements
          .via(persistMetrics)
          .runFold(0) { (cnt, _) => cnt + 1 }

      complete {
        measurementsSubmitted.map(n => Map("msg" -> s"""Total metrics received: $n"""))
      }
    }
  }

// tests ------------------------------------------------------------
// uploading an array or newline separated values works out of the box
val data = HttpEntity(
  ContentTypes.`application/json`,
  """
    |{"id":"temp","value":32}
    |{"id":"temp","value":31}
    |
  """.stripMargin)

Post("/metrics", entity = data) ~> route ~> check {
  status should ===(StatusCodes.OK)
  responseAs[String] should ===("""{"msg":"Total metrics received: 2"}""")
}

// the FramingWithContentType will reject any content type that it does not understand:
val xmlData = HttpEntity(
  ContentTypes.`text/xml(UTF-8)`,
  """|<data id="temp" value="32"/>
     |<data id="temp" value="31"/>""".stripMargin)

Post("/metrics", entity = xmlData) ~> route ~> check {
  handled should ===(false)
  rejection should ===(UnsupportedRequestContentTypeRejection(Set(ContentTypes.`application/json`)))
}

Java

final Route incomingStreaming = path("tweets", () ->
  post(() ->
    extractMaterializer(mat -> {
      final JsonEntityStreamingSupport jsonSupport = EntityStreamingSupport.json();

      return entityAsSourceOf(JavaTweets, jsonSupport, sourceOfTweets -> {
          final CompletionStage<Integer> tweetsCount = sourceOfTweets.runFold(0, (acc, tweet) -> acc + 1, mat);
          return onComplete(tweetsCount, c -> complete("Total number of tweets: " + c));
        });
      }
    )
  )  
);

Simple CSV streaming example

Akka HTTP provides another EntityStreamingSupportEntityStreamingSupport out of the box, namely csv (comma-separated values). For completeness, we demonstrate its usage in the snippet below. As you’ll notice, switching between streaming modes is fairly simple: You only have to make sure that an implicit MarshallerMarshaller of the requested type is available and that the streaming support operates on the same Content-Type as the rendered values. Otherwise, you’ll see an error during runtime that the marshaller did not expose the expected content type and thus we can’t render the streaming response).

Scala

// [1] provide a marshaller to ByteString
implicit val tweetAsCsv = Marshaller.strict[Tweet, ByteString] { t =>
  Marshalling.WithFixedContentType(ContentTypes.`text/csv(UTF-8)`, () => {
    val txt = t.txt.replaceAll(",", ".")
    val uid = t.uid
    ByteString(List(uid, txt).mkString(","))
  })
}

// [2] enable csv streaming:
implicit val csvStreaming = EntityStreamingSupport.csv()

val route =
  path("tweets") {
    val tweets: Source[Tweet, NotUsed] = getTweets
    complete(tweets)
  }

// tests ------------------------------------------------------------
val AcceptCsv = Accept(MediaRange(MediaTypes.`text/csv`))

Get("/tweets").withHeaders(AcceptCsv) ~> route ~> check {
  responseAs[String] shouldEqual
    "1,#Akka rocks!" + "\n" +
    "2,Streaming is so hot right now!" + "\n" +
    "3,You cannot enter the same river twice."
}

Java

final Marshaller<JavaTweet, ByteString> renderAsCsv = 
  Marshaller.withFixedContentType(ContentTypes.TEXT_CSV_UTF8, t -> 
    ByteString.fromString(t.getId() + "," + t.getMessage())
  );

final CsvEntityStreamingSupport compactJsonSupport = EntityStreamingSupport.csv();

final Route responseStreaming = path("tweets", () ->
  get(() ->
    parameter(StringUnmarshallers.INTEGER, "n", n -> {
      final Source<JavaTweet, NotUsed> tws =
        Source.repeat(new JavaTweet(12, "Hello World!")).take(n);
      return completeWithSource(tws, renderAsCsv, compactJsonSupport);
    })
  )
);

Implementing custom EntityStreamingSupport traits

The EntityStreamingSupportEntityStreamingSupport infrastructure is open for extension and not bound to any single format, content type, or marshalling library. The provided JSON support does not rely on spray-json directly, but uses Marshaller<T, ByteString>Marshaller[T, ByteString] instances, which can be provided using any JSON marshalling library (such as Circe, Jawn or Play JSON).

When implementing a custom support trait, one should simply extend the EntityStreamingSupportEntityStreamingSupport abstract class and implement all of its methods. It’s best to use the existing implementations as a guideline.