Actorpublisher 作为 handleMessagesWithSinkSource 中的源答案

【问题标题】：Actorpublisher as source in handleMessagesWithSinkSourceActorpublisher 作为 handleMessagesWithSinkSource 中的源
【发布时间】：2016-04-23 20:04:00
【问题描述】：

我是 AKKA Streams 的新手。（使用 Akka v 2.4.4）我正在尝试创建一个可以将新通知推送给订阅客户端的 Websocket。我的策略是实现一个 ActorPublisher，稍后我可以向它发送消息，然后将其推送给客户端。

首先，我复制了一个 ActorPublisher 的示例：

case class Tick()

class TickActor extends ActorPublisher[Int] {
  import scala.concurrent.duration._

  implicit val ec = context.dispatcher

  val tick = context.system.scheduler.schedule(1 second, 1 second, self, `Tick())`

  var cnt = 0
  var buffer = Vector.empty[Int]

  override def receive: Receive = {
    case Tick() => {
      cnt = cnt + 1
      if (buffer.isEmpty && totalDemand > 0) {
        onNext(cnt)
      }
      else {
        buffer :+= cnt
        if (totalDemand > 0) {
          val (use,keep) = buffer.splitAt(totalDemand.toInt)
          buffer = keep
          use foreach onNext
        }
      }
    }
  }

  override def postStop() = tick.cancel()
}

我的问题是我不知道如何将它用作源代码。

我尝试了以下方法：

val source: Source[Strict, ActorRef] = Source.actorPublisher(Props[TickActor]).map(i => TextMessage(i.toString))
  optionalHeaderValueByType[akka.http.scaladsl.model.ws.UpgradeToWebSocket]() {
    case Some(upgrade) =>
      complete(
        upgrade.handleMessagesWithSinkSource(Sink.ignore,source))
    case None =>
      reject(akka.http.scaladsl.server.ExpectedWebSocketRequestRejection)
  }

但是当我与客户端连接时，我得到以下 ClassCastException: java.lang.ClassCastException: java.lang.Integer cannot be cast to scala.runtime.Nothing$

如果我将源更改为：

val src: Source[Strict, NotUsed] = Source.fromIterator(() => Iterator.continually(ThreadLocalRandom.current.nextInt()))
      .filter(i => i > 0 && i % 2 == 0).map(i => TextMessage(i.toString))

它运行得很好。

我有点费力地连接点，所以希望你能引导我走向正确的方向。

【问题讨论】：

标签： scala websocket stream akka

【解决方案1】：

我尝试了您的示例并且能够重现该问题。我只做了一项更改来解决问题。添加了类型参数，现在它很有意义，因为在 akka 流中的某处，有一个类似的代码 elem.asInstanceOf[T]。因此，当actorPublisher中缺少类型时，类型被推断为Nothing

val source = Source.actorPublisher[Int](Props[TickActor]).map(i => TextMessage(i.toString))

【讨论】：