使用 Akka-HTTP、Twitter4j 从 API 获取推文

Question

我有通过用户 ID 获取推文的方法，为此我使用了 Twitter4J 和 Akka 流。我想以用户发送获取请求并接收一组推文的方式创建一个 API。请求如下所示： http://localhost:8080/tweets/534563976

目前，我只能获得 Ok 状态，但我不知道如何通过像 Postman 这样的客户端获取推文（当我发送 get 请求时）。因为我只在我的控制台上收到所有推文。我的 API 路由如下所示：

trait ApiRoute extends Databases {
  val twitterStream = TwitterStreamFilters.configureTwitterStream()
  val counter = new Counter
  twitterStream.addListener(counter)

  val routes = pathPrefix("tweets") {
    pathSingleSlash {
      complete("hello")
    }
  } ~
    pathPrefix("tweets") {
      pathPrefix(LongNumber) {
        userId =>
          get {
            onSuccess(Future.successful(TwitterStreamFilters.filterTwitterStreamByUserID(twitterStream, Array(userId)))) {
              complete(StatusCodes.OK)
            }
          }
      }
    }
}

但我想把它改成这样，。但是当我写这个时，我得到 TypeMismatch, expected:ToResponseMarshallable, actual:RequestContext 编译错误：

onSuccess(Future.successful(TwitterStreamFilters.filterTwitterStreamByUserID(twitterStream, Array(userId)))) {
                  result => complete(result)
                }

这是过滤推文的方法：

  def filterTwitterStreamByUserID(twitterStream: TwitterStream, userId: Array[Long]) = {
    twitterStream.filter(new FilterQuery().follow(userId))
  }

这是我获取推文流的代码：

class Counter extends StatusAdapter{

  implicit val system = ActorSystem("EmojiTrends")
  implicit val materializer = ActorMaterializer()
  implicit val executionContext = system.dispatcher
  implicit val LoggingAdapter =
    Logging(system, classOf[Counter])

  val overflowStrategy = OverflowStrategy.backpressure
  val bufferSize = 1000
  val statusSource = Source.queue[Status](
    bufferSize,
    overflowStrategy
  )

  val sink: Sink[Any, Future[Done]] = Sink.foreach(println)
  val flow: Flow[Status, String, NotUsed] = Flow[Status].map(status => status.getText)

  val graph = statusSource via flow to sink
  val queue = graph.run()

  override def onStatus(status: Status) =
    Await.result(queue.offer(status), Duration.Inf)
}

那么我如何更改我的 API 方法调用或过滤方法本身以获取一组推文作为对我的 get 请求的响应？或者更好地说如何将它们从控制台重定向到响应正文？我应该为此使用数据库吗？ 任何建议表示赞赏！

Answer 1

匹配onSuccess的结果：

onSuccess(Future.successful(TwitterStreamFilters.filterTwitterStreamByUserID(twitterStream, Array(userId)))) {
  case tweets: Set[Tweet] => complete(StatusCodes.OK, tweets)
}

当然，您必须提供响应编组器。如果你想要一个 JSON 作为响应，你可以使用Akka HTTP JSON + Circe。只需添加这两个导入，如果您在响应类中没有一些特殊类型（例如Tweet），它应该可以工作：

import de.heikoseeberger.akkahttpcirce.FailFastCirceSupport._
import io.circe.generic.auto._