我未能實作Sink.seq，當需要實作時，我因此例外而失敗
akka.stream.SubscriptionWithCancelException$StageWasCompleted$:
這是github上的完整源代碼：https : //github.com/Christewart/bitcoin-s-core/blob/aaecc7c180e5cc36ec46d73d6b2b0b0da87ab260/app/server-test/src/test/scala/org/bitcoins/server/WebsocketTests.scala #L51
我試圖將所有從 websocket 推出的元素聚合到一個Sink.seq. 我必須先進行一些 json 轉換，然后才能將Sink.seq.
val endSink: Sink[WalletNotification[_], Future[Seq[WalletNotification[_]]]] =
Sink.seq[WalletNotification[_]]
val sink: Sink[Message, Future[Seq[WalletNotification[_]]]] = Flow[Message]
.map {
case message: TextMessage.Strict =>
//we should be able to parse the address message
val text = message.text
val notification: WalletNotification[_] = {
upickle.default.read[WalletNotification[_]](text)(
WsPicklers.walletNotificationPickler)
}
logger.info(s"Notification=$notification")
notification
case msg =>
logger.error(s"msg=$msg")
sys.error("")
}
.log(s"@@@ endSink @@@")
.toMat(endSink)(Keep.right)
val f: Flow[
Message,
Message,
(Future[Seq[WalletNotification[_]]], Promise[Option[Message]])] = {
Flow
.fromSinkAndSourceMat(sink, Source.maybe[Message])(Keep.both)
}
val tuple: (
Future[WebSocketUpgradeResponse],
(Future[Seq[WalletNotification[_]]], Promise[Option[Message]])) = {
Http()
.singleWebSocketRequest(req, f)
}
val walletNotificationsF: Future[Seq[WalletNotification[_]]] =
tuple._2._1
val promise: Promise[Option[Message]] = tuple._2._2
logger.info(s"Requesting new address for expectedAddrStr")
val expectedAddressStr = ConsoleCli
.exec(CliCommand.GetNewAddress(labelOpt = None), cliConfig)
.get
val expectedAddress = BitcoinAddress.fromString(expectedAddressStr)
promise.success(None)
logger.info(s"before notificationsF")
//hangs here, as the future never gets completed, fails with an exception
for {
notifications <- walletNotificationsF
_ = logger.info(s"after notificationsF")
} yield {
//assertions in here...
}
我究竟做錯了什么？
uj5u.com熱心網友回復：
要保持客戶端連接打開，您需要“更多代碼”，例如：
val sourceKickOff = Source
.single(TextMessage("kick off msg"))
// Keeps the connection open
.concatMat(Source.maybe[Message])(Keep.right)
請參閱完整的作業示例，它使用來自服務器的訊息：https : //github.com/pbernet/akka_streams_tutorial/blob/b6d4c89a14bdc5d72c557d8cede59985ca8e525f/src/main/scala/akkahttp/WebsocketEcho.scala#L28
uj5u.com熱心網友回復：
問題是這條線
Flow.fromSinkAndSourceMat(sink, Source.maybe[Message])(Keep.both)
它需要是
Flow.fromSinkAndSourceCoupledMat(sink, Source.maybe[Message])(Keep.both)
當流終止時，Coupled物化流的一部分將確保終止下游的 Sink。
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/394713.html