使用AKKA-HTTP流式CSV源

Question

我想使用reactivemongo-akkastream 0.12.1從Mongodb流式傳輸數據，並將結果返回到其中一條路徑（使用Akka-http）的CSV流中。 我沒有實現，這裏繼爲例：

http://doc.akka.io/docs/akka-http/10.0.0/scala/http/routing-dsl/source-streaming-support.html#simple-csv-streaming-example

，它看起來做工精細。

我現在面臨的唯一問題是如何將標題添加到輸出CSV文件。有任何想法嗎？

感謝

Answer 1

除了一個事實，即這個例子是不是真的生成CSV的穩健方法（不提供適當的轉義），則需要返工有點添加標題。這是我會做：

1. 做出Flow到Source[Tweet]轉換爲CSV行，例如源一個Source[List[String]]
2. 連擊它包含您的標題作爲一個單一的List[String]
3. 源適應編組渲染行，而不是推特

這裏的來源是一些示例代碼：

case class Tweet(uid: String, txt: String) 

def getTweets: Source[Tweet, NotUsed] = ??? 

val tweetToRow: Flow[Tweet, List[String], NotUsed] = 
    Flow[Tweet].map { t => 
    List(
     t.uid, 
     t.txt.replaceAll(",", ".")) 
    } 

// provide a marshaller from a row (List[String]) to a ByteString 
implicit val tweetAsCsv = Marshaller.strict[List[String], ByteString] { row => 
    Marshalling.WithFixedContentType(ContentTypes.`text/csv(UTF-8)`,() => 
    ByteString(row.mkString(",")) 
) 
} 

// enable csv streaming 
implicit val csvStreaming = EntityStreamingSupport.csv() 

val route = path("tweets") { 
    val headers = Source.single(List("uid", "text")) 
    val tweets: Source[List[String], NotUsed] = getTweets.via(tweetToRow) 
    complete(headers.concat(tweets)) 
} 

更新：如果您的getTweets方法返回Future，則可以僅通過其源值映射並預先設置標題，例如：

val route = path("tweets") { 
    val headers = Source.single(List("uid", "text")) 
    val rows: Future[Source[List[String], NotUsed]] = getTweets 
     .map(tweets => headers.concat(tweets.via(tweetToRow))) 
    complete(rows) 
}