时间戳

首页 > 解决方案 > 无法在 Akka Stream 中使用 GraphStage 类运行 SourceShape

scala - 无法在 Akka Stream 中使用 GraphStage 类运行 SourceShape

问题描述

我正在尝试使用 GraphStage 构造创建一个 Redis Akka 流源。这个想法是，每当我从 subscribe 方法获得更新时，我都会将其推送到下一个组件。此外，如果没有拉动信号，则组件应背压。这是代码：

class SubscriberSourceShape(channel: String, subscriber: Subscriber) 
    extends GraphStage[SourceShape[String]] {

  private val outlet: Outlet[String] = Outlet("SubscriberSource.Out")

  override def shape: SourceShape[String] = SourceShape(outlet)

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = {
    new GraphStageLogic(shape) {
      val callback = getAsyncCallback((msg: String) => push(outlet, msg)
      val handler = (msg: String) => callback.invoke(msg)

      override def preStart(): Unit = subscriber.subscribe(channel)(handler)
    }
  }
}

然而，当我用一个简单的接收器运行它时，我得到了这个错误：

Error in stage [akka.http.impl.engine.server.HttpServerBluePrint$ProtocolSwitchStage@58344854]: No handler defined in stage [...SubscriberSourceShape@6a9193dd] for out port [RedisSubscriberSource.Out(1414886352). All inlets and outlets must be assigned a handler with setHandler in the constructor of your graph stage logic.

怎么了？

标签： scalaakkaakka-stream

解决方案

您收到此错误是因为您尚未为 Source 设置任何输出处理程序，因此当下游组件（Flow 或 Sink）向此 Source 发送拉信号时，没有处理程序来处理拉信号。

您可以添加一个 OutputHandler 来消除错误。将 onPull 方法留空，因为您在 asyncCallback 上生成元素。只需将其添加到 GraphStageLogic 正文中：

      setHandler(outlet, new OutHandler {
        override def onPull(): Unit = {}
      })

推荐阅读