代码之家 › 专栏 › 技术社区 › stefanobaghino

使用流中的最新项完成请求

reactive-streams akka-http akka-stream akka scala

stefanobaghino · 技术社区 · 6 年前

我想完成一个 GET 使用流中最新可用项的请求。这个流特别聚集了由参与者生成并已被websocket单独使用的事件。

假设事件可以表示为:

final case class Event(id: String, value: Double)

我做的第一件事就是创造一个 SourceQueue 其中,参与者将推送事件和中心,以便不同的客户端可以独立接收这些事件:

val (queue, hub) =
  Source.queue[Event](256, OverflowStrategy.dropHead).
    toMat(BroadcastHub.sink(bufferSize = 256))(Keep.both).run()

然后我就可以创造一个演员来推动事件 queue 及格 hub 对于通过websocket服务事件的服务:

extractUpgradeToWebSocket { upgrade =>
  complete(upgrade.handleMessagesWithSinkSource(
    inSink = Sink.ignore,
    outSource =
      hub.map(in => TextMessage(fmt.write(in).toString()))
  ))
}

这很好,同时也适用于多个消费者。

我接下来要做的是有一个服务,它使用来自 枢纽 并生成每个ID的最新事件列表,通过 得到 端点。

我尝试了几种方法来解决这个问题。我尝试的两种方法是:

运行更新私有变量的流
配有一个水槽 last 要素

运行更新私有变量的流

实际上,这是我最后一次尝试。奇怪的(或者是?)我想我注意到事实上没有任何记录 log Combinator被记录了?).

使用这种方法的结果是 latest 总是 null 因此反应总是空的。

final class Service(hub: Source[Event, NotUsed])(implicit s: ActorSystem, m: ActorMaterializer, t: Timeout) extends Directives with JsonSupport {

  implicit private val executor = system.dispatcher

  @volatile private[this] var latest: List[Event] = _

  hub.
    log("hub", identity).
    groupBy(Int.MaxValue, { case Event(id, _) => id }).
    map { case event @ Event(id, _) => Map(id -> event) }.
    reduce(_ ++ _).
    mergeSubstreams.
    map(_.values.toList).
    toMat(Sink.foreach(latest = _))(Keep.none).run()

  val definition = get { complete(Option(latest)) }

}

我还尝试了一种类似的方法,使用“box”actor并将聚合传递给它,但效果是相同的。

配有一个水槽 `最后的` 要素

这是我尝试的第一种方法。其结果是,响应将挂起,直到达到超时,并且akka http将返回500到浏览器。

final class Service(hub: Source[Event, NotUsed])(implicit s: ActorSystem, m: ActorMaterializer, t: Timeout) extends Directives with JsonSupport {

  implicit private val executor = system.dispatcher
  private[this] val currentLocations =
    hub.
      groupBy(Int.MaxValue, { case Event(id, _) => id }).
      map { case event @ Event(id, _) => Map(id -> event) }.
      reduce(_ ++ _).
      mergeSubstreams.
      map(_.values.toList).
      runWith(Sink.reduce((_, next) => next))

  val definition = get { complete(currentLocations) }

}

1 回复 | 直到 6 年前

Ramón J Romero y Vigil 6 年前

ActorRef 作为一个接收器

您可以创建 Actor 一直在跑 Map 属于 id 到 Event :

import scala.collection.immutable

object QueryMap

class MapKeeperActor() extends Actor {

  var internalMap = immutable.Map.empty[String, Event]

  override def receive = {
    case e : Event    => internalMap = internalMap + (e.id -> e)
    case _ : QueryMap => sender ! internalMap
  }
}

然后可以在 Sink 将附加到 BroadcastHub :

object OnCompleteMessage

val system : ActorSystem = ???

val mapKeeperRef = system.actorOf(Props[MapKeeperActor])

val mapKeeperSink : Sink[Event, _] = Sink.actorRef[Event](mapKeeperRef, OnCompleteMessage)

路由中的查询参与者

我们现在可以创建 Route 它将使用指令查询地图管理员。但是,您必须决定如何序列化 地图 变成一个 ResponseEntity 对于 HttpResponse :

val serializeMap : Map[String, Event] => ResponseEntity = ???

val route = 
  get {
    onComplete( (mapKeeperRef ? QueryMap).mapTo[Map[String, Event]]) {
      case Success(map) => complete(HttpResponse(entity=serializeMap(map))
      case Failure(ex)  => complete((InternalServerError, s"An error occurred: ${ex.getMessage}"))
    }
  }

使用流中的最新项完成请求

运行更新私有变量的流

配有一个水槽 最后的 要素

配有一个水槽 `最后的` 要素