如何通过 ZStream 读取来自 ZHub 的消息？

【问题标题】：How to read messages from ZHub via ZStream?如何通过 ZStream 读取来自 ZHub 的消息？
【发布时间】：2021-05-06 10:22:29
【问题描述】：

我是 ZHub 和 ZStream 的新手，想熟悉一下他们的 API。

不幸的是，我无法让这个简单的示例工作：

for
    hub <- Hub.bounded[String](4)
    stream = ZStream.fromHub(hub)
    _ <- hub.publish("Hello")
    _ <- hub.publish("World")
    collected <- stream.runCollect
    _ <- ZIO.foreach(collected) { msg => console.putStrLn(msg) }
yield
    ()

我怀疑这个程序不会终止，因为我正在尝试收集无限流。我还尝试使用stream.tap(...) 打印消息或关闭集线器。没有任何帮助。

我在这里缺少什么？任何帮助表示赞赏，谢谢。

【问题讨论】：

标签： scala zio zio-streams

【解决方案1】：

@adamgfraser 在 GitHub 上提供了一个工作示例：

import zio._
import zio.stream._

object Example extends App {

  def run(args: List[String]): URIO[ZEnv, ExitCode] =
    for {
      promise <- Promise.make[Nothing, Unit]
      hub     <- Hub.bounded[String](2)
      stream = ZStream.managed(hub.subscribe).flatMap { queue =>
                 ZStream.fromEffect(promise.succeed(())) *>
                   ZStream.fromQueue(queue)
               }
      fiber     <- stream.take(2).runCollect.fork
      _         <- promise.await
      _         <- hub.publish("Hello")
      _         <- hub.publish("World")
      collected <- fiber.join
      _         <- ZIO.foreach(collected)(console.putStrLn(_)).orDie
    } yield ExitCode.success
}

我的错误是在等待订阅完成之前将值发布到中心。

【讨论】：