Playframework 和 Twitter Streaming API [英] Playframework and Twitter Streaming API

问题描述

如何从 Twitter Streaming API - POST statuses/filter 读取响应数据?我已建立连接并收到 200 状态代码，但我不知道如何阅读推文.我只想在推文到来时打印出来.

How to read response data from Twitter Streaming API - POST statuses/filter ? I have established connection and I receive 200 status code, but I don't know how to read tweets. I just want to println tweets as they coming.

ws.url(url)
.sign(OAuthCalculator(consumerKey, requestToken))
.withMethod("POST")
.stream()
.map { response =>
  if(response.headers.status == 200)
    println(response.body)
} 

我找到了这个解决方案

ws.url(url)
.sign(OAuthCalculator(consumerKey, requestToken))
.withMethod("POST")
.stream()
.map { response => 
  if(response.headers.status == 200){
    response.body
      .scan("")((acc, curr) => if (acc.contains("
")) curr.utf8String else acc + curr.utf8String)
      .filter(_.contains("
"))
      .map(json => Try(parse(json).extract[Tweet]))
      .runForeach {
        case Success(tweet) =>
          println("-----")
          println(tweet.text)
        case Failure(e) =>
          println("-----")
          println(e.getStackTrace)
      }
  }
}

推荐答案

流式 WS 请求的响应主体是 Akka Streams Source 字节.由于 Twitter Api 响应以换行符分隔(通常)，您可以使用 Framing.delimiter 将它们拆分为字节块，将块解析为 JSON，然后对它们执行您想要的操作.这样的事情应该可以工作:

The body of the response for a streaming WS request is an Akka Streams Source of bytes. Since Twitter Api responses are newline delimited (usually) you can use Framing.delimiter to split them up into byte chunks, parse the chunks to JSON, and do what you want with them. Something like this should work:

import akka.stream.scaladsl.Framing
import scala.util.{Success, Try}
import akka.util.ByteString
import play.api.libs.json.{JsSuccess, Json, Reads}
import play.api.libs.oauth.{ConsumerKey, OAuthCalculator, RequestToken}

case class Tweet(id: Long, text: String)
object Tweet {
  implicit val reads: Reads[Tweet] = Json.reads[Tweet]
}

def twitter = Action.async { implicit request =>
  ws.url("https://stream.twitter.com/1.1/statuses/filter.json?track=Rio2016")
      .sign(OAuthCalculator(consumerKey, requestToken))
      .withMethod("POST")
      .stream().flatMap { response =>
    response.body
      // Split up the byte stream into delimited chunks. Note
      // that the chunks are quite big
      .via(Framing.delimiter(ByteString.fromString("
"), 20000))
      // Parse the chunks into JSON, and then to a Tweet.
      // A better parsing strategy would be to account for all
      // the different possible responses, but here we just
      // collect those that match a Tweet.
      .map(bytes => Try(Json.parse(bytes.toArray).validate[Tweet]))
      .collect {
        case Success(JsSuccess(tweet, _)) => tweet.text
      }
      // Print out each chunk
      .runForeach(println).map { _ =>
        Ok("done")
    }
  }
}

注意:要实现流，您需要将隐式 Materializer 注入控制器.

Note: to materialize the stream you'll need to inject an implicit Materializer into your controller.

这篇关于Playframework 和 Twitter Streaming API的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！