How to complete a stream of numbers as CSV values in Akka HTTP?

user_1879109 0
Viewed 143

I have a Akka source as a stream of numbers implemented as:

Source(Stream(1, 2, 3, 4, 5))

I am trying make use of Akka streaming support in Akka HTTP to return a streaming response as comma-separated values.

I have followed akka doc on simple csv source streaming to come up with the following implementation:

implicit val csvFormat = Marshaller.strict[Int, ByteString] { res =>
    Marshalling.WithFixedContentType(ContentTypes.`text/csv(UTF-8)`, () => {
    ByteString(List(res).mkString(","))
    })
}

implicit val streamingSupport: CsvEntityStreamingSupport = EntityStreamingSupport.csv()

complete(Source(Stream(1, 2, 3, 4, 5)))

But apparently this is not the right use case of CSV entity streaming support for my purpose. This results in every number being streamed in a new line.

But that is not what I want. I want to reply as comma-separated list instead, like 1,2,3,4,5.

How is it possible to achieve this using streaming support in Akka HTTP?

akka-stream akka-http akka scala
2 Answers

That newline is added by the flow-renderer defined in EntityStreamingSupport.csv().

We need to define our own custom EntityStreamingSupport for this to work.

val route =
  path("test") {
    val responseSource: Source[Int, NotUsed] =
      Source.fromIterator(() => Stream(1, 2, 3, 4, 5).iterator)

    val byteStringSource: Source[ByteString, NotUsed] =
      responseSource.map(i => ByteString(i.toString))

    val streamingSource =
      byteStringSource.map(bs => HttpEntity(ContentTypes.`text/plain(UTF-8)`, bs))

    implicit val streamingSupport =
      EntityStreamingSupport.csv(maxLineLength = 16 * 1024)
        .withSupported(ContentTypeRange(ContentTypes.`text/plain(UTF-8)`))
        .withContentType(ContentTypes.`text/plain(UTF-8)`)
        .withFramingRenderer(Flow[ByteString].map(bs => bs ++ ByteString(",")))

    complete((streamingSource))
  }

curl localhost:8080/test -v 
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 8080 (#0)
> GET /test HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.64.1
> Accept: */*
> 
< HTTP/1.1 200 OK
< Server: akka-http/10.2.4
< Date: Tue, 20 Jul 2021 07:50:46 GMT
< Transfer-Encoding: chunked
< Content-Type: text/plain; charset=UTF-8
< 
* Connection #0 to host localhost left intact
1,2,3,4,5,* Closing connection 0

Edit : to elimiate that traling comma, we can use a window hack.

val route =
  path("test") {
    val responseSource: Source[Int, NotUsed] =
      Source.fromIterator(() => Stream(1, 2, 3, 4, 5).iterator)

    val startByteString = ByteString("$start$")

    val byteStringSource: Source[ByteString, NotUsed] =
        responseSource.map(i => ByteString(i.toString)).prepend(Source.single(startByteString))

    val streamingSource =
      byteStringSource.map(bs => HttpEntity(ContentTypes.`text/plain(UTF-8)`, bs))

    implicit val streamingSupport =
      EntityStreamingSupport.csv(maxLineLength = 16 * 1024)
        .withSupported(ContentTypeRange(ContentTypes.`text/plain(UTF-8)`))
        .withContentType(ContentTypes.`text/plain(UTF-8)`)
        .withFramingRenderer(
          Flow[ByteString].sliding(2, 1)
            .map { bsSeq =>
              if (startByteString.equals(bsSeq(0))) {
                // first int; no need for comma
                bsSeq(1)
              } else {
                // not first int; add comma
                ByteString(",") ++ bsSeq(1)
              }

            }
        )

    complete((streamingSource))
  }

curl localhost:8080/test -v 
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 8080 (#0)
> GET /test HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.64.1
> Accept: */*
> 
< HTTP/1.1 200 OK
< Server: akka-http/10.2.4
< Date: Tue, 20 Jul 2021 08:28:05 GMT
< Transfer-Encoding: chunked
< Content-Type: text/plain; charset=UTF-8
< 
* Connection #0 to host localhost left intact
1,2,3,4,5* Closing connection 0
user_11519290

You are missing a higher-level class. It can't be an Int and compose a row with more than one column. Creating such a class as the following MyBO does the trick.

    case class MyBO(a: Int, b: Int, c: Int, d: Int, e: Int)
    
    implicit val myBOAsCsv = Marshaller.strict[MyBO, ByteString] { t =>
      Marshalling.WithFixedContentType(ContentTypes.`text/csv(UTF-8)`, () => {
        ByteString(List(t.a, t.b, t.c, t.d, t.e).mkString(","))
        })
      }

    implicit val csvStreaming = EntityStreamingSupport.csv()

    val route: Route =
      path("ping") {
        get {
          complete(Source(Stream(MyBO(1,3,5,7,11))))
        }
      }

 curl localhost:8080/ping -v                                                                                                                               [3f8df8e]
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to localhost (127.0.0.1) port 8080 (#0)
> GET /ping HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.64.1
> Accept: */*
>
< HTTP/1.1 200 OK
< Server: akka-http/10.2.4
< Date: Tue, 20 Jul 2021 07:10:18 GMT
< Transfer-Encoding: chunked
< Content-Type: text/csv; charset=UTF-8
<
1,3,5,7,11
* Connection #0 to host localhost left intact
* Closing connection 0
user_24998700
Related