Akka-Http + Twitter 流媒体 API

Question

我最近对尝试 Scala 世界中的一些流媒体功能产生了兴趣。这发生在我阅读 Play 2 中的 Iteratee API 时。

但是，人们似乎认为 Iteratee API 已接近弃用，并向我推荐了以下库之一：

• scalaz 流
• akka-streams，或者更具体地说，akka-http

我真的不想进入 scalaz 世界，所以我决定查看 akka-http。

不幸的是，目前关于 akka-http 主题的文档似乎非常稀少，我在让一切正常工作时遇到了很多麻烦。

像往常一样，我选择了大家最喜欢的流数据来源：Twitter。

谷歌搜索主题大多会导致

• Matthias Nehlsens 在他的BirdWatch project 上的出色工作。不幸的是，他仍然使用迭代器。
• 人们在 Twitter 客户端上使用 akka-streams，但我不太喜欢使用这些，因为我不会从这些中真正学到很多东西。

使用 akka-http 执行基本的 GET 请求的结构似乎有点像这样：

object AkkaHttpExample extends App {

    implicit val system = ActorSystem("akka-http-example")
    implicit val materializer = ActorMaterializer()
    import system.dispatcher 

    val connectionFlow: Flow[HttpRequest, HttpResponse, Future[Http.OutgoingConnection]] = Http().outgoingConnection("www.stackoverflow.com", 80)

    val request: HttpRequest = HttpRequest(
        HttpMethods.GET,
        uri = "/"
    )

    val future: Future[HttpResponse] =
        Source.single(request)
            .via(connectionFlow)
            .runWith(Sink.head)

    val result: HttpResponse = Await.result(future, 5 seconds)
}

上面的代码有效（尽管由于某种原因解析结果的主体非常烦人）。

当我尝试做同样的事情但指向 /1.1/statuses/sample.json 端点（它应该发出一个示例流）时，我的 Future 只是坐在那里并超时。尽管考虑到数据的流式传输性质，这似乎合乎逻辑，但这应该返回 404，因为此时我什至没有进行正确的 OAuth。

作为参考，这是我到目前为止的代码：

object AkkaHttpExample extends App {

    implicit val system = ActorSystem("akka-http-example")
    implicit val materializer = ActorMaterializer()
    import system.dispatcher    

    val connectionFlow: Flow[HttpRequest, HttpResponse, Future[Http.OutgoingConnection]] = Http().outgoingConnection("stream.twitter.com", 80)

    val request: HttpRequest = HttpRequest(
        HttpMethods.GET,
        uri = "/1.1/statuses/sample.json"
    )

    val future: Future[HttpResponse] =
        Source.single(request)
            .via(connectionFlow)
            .runWith(Sink.head)

    val result: HttpResponse = Await.result(future, 50 seconds)
}

就像我说的那样，我认为这可能是流式传输的性质无论如何都会导致问题，因此我尝试根据我能找到的唯一示例更改我的代码以逐个处理块，如下所示：

object AkkaHttpExample extends App {

    implicit val system = ActorSystem("akka-http-example")
    implicit val materializer = ActorMaterializer()
    import system.dispatcher

    val connectionFlow: Flow[HttpRequest, HttpResponse, Future[Http.OutgoingConnection]] = Http().outgoingConnection("stream.twitter.com", 80)

    val request: HttpRequest = HttpRequest(
        HttpMethods.GET,
        uri = "/1.1/statuses/sample.json"
    )

    Source.single(request)
        .via(connectionFlow)
        .map(_.entity.dataBytes)
        .flatten(FlattenStrategy.concat)
        .map(_.decodeString("UTF-8"))
        .runForeach(println _)
        .onComplete(_ => system.terminate())
}

这会导致应用程序立即终止。删除 .onComplete 子句使 ActorSystem 和应用程序保持运行，但实际上似乎什么也没发生 :'(

有人有这方面的经验吗？到目前为止，图书馆一直是一个令人头疼的问题。我应该回到 Play + WS + Iteratees 吗？

Answer 1

我认为您遇到的第一个问题是错误地使用http（端口80）与https（端口443）作为您的连接协议。当我切换到 https 并通过有效的 oauth 授权标头时，我能够让事情正常工作。我也稍微修改了您的示例以匹配最新的 http api。这是我的工作：

implicit val system = ActorSystem("akka-http-example")
implicit val materializer = ActorMaterializer()
import system.dispatcher

val connectionFlow = Http().cachedHostConnectionPoolTls[Long]("stream.twitter.com", 443)

//Note myAuth omitted here.  Also using a RawHeader because I was
//too lazy to use the real Authorization model header
val authHeader = RawHeader("Authorization", myAuth)
Source.single((request, 1l))
    .via(connectionFlow)
    .map{
      case (util.Success(resp), id) =>
        resp.entity.dataBytes
      case other =>
        println(s"Got unexpected response: $other")            
        Source.empty
     }
    .flatten(FlattenStrategy.concat)
    .map(_.decodeString("UTF-8"))
    .runForeach(println _)
    .onComplete{tr =>
      system.shutdown
    }