如何通过ZIO环境在ZIO任务之间共享ZIO队列

Question

我对Scala和ZIO有些陌生，遇到了一些奇怪的谜团。

我希望设置一个包含ZIO队列的ZIO环境，并在以后具有与这个共享队列不同的ZIO任务offer和take。

我试着像这样定义我的环境

    trait MainEnv extends Console with Clock
    {
      val mainQueue = Queue.unbounded[String]
    }

并从以下不同的任务访问队列

    for {
      env      <- ZIO.environment[MainEnv]
      queue    <- env.mainQueue
      ...

但是在我的测试中，我观察到我的每个单独的任务都有一个单独的队列实例。

查看unbounded的签名

  def unbounded[A]: UIO[Queue[A]]

我观察到它不会立即返回一个队列，而是返回一个返回队列的效果，所以虽然观察到的行为是有意义的，但这完全不是我所希望的，我也没有看到一个明确的方法来得到我想要的行为。

对于如何实现通过存储在环境中的共享队列进行通信的不同任务的目标，我将不胜感激。

这里供参考的是我的代码和输出。

样本执行

bash-3.2$ sbt run
[info] Loading project definition from /private/tmp/env-q/project
[info] Loading settings for project root from build.sbt ...
[info] Set current project to example (in build file:/private/tmp/env-q/)
[info] Compiling 1 Scala source to /private/tmp/env-q/target/scala-2.12/classes ...
[info] Done compiling.
[info] Packaging /private/tmp/env-q/target/scala-2.12/example_2.12-0.0.1-SNAPSHOT.jar ...
[info] Done packaging.
[info] Running example.Main 
env example.Main$$anon$1@36fbcafd queue zio.Queue$$anon$1@65b9a444
env example.Main$$anon$1@36fbcafd queue zio.Queue$$anon$1@7c050764

(挂起-注意env对象是相同的，但是队列对象不同，所以第二个任务被卡住了)

/tmp/env-q/test.scala

以下是我的完整测试，它基于https://www.slideshare.net/jdegoes/zio-queue幻灯片37中的示例

    package example
    import zio.{App, Queue, ZIO}
    import zio.blocking.Blocking
    import zio.clock.Clock
    import zio.console._

    trait MainEnv extends Console with Clock    // environment with queue
    {
      val mainQueue = Queue.unbounded[String]
    }

    object Main extends App                     // main test
    {
      val task1 = for {                         // task to add something to the queue
        env      <- ZIO.environment[MainEnv]
        queue    <- env.mainQueue
        _        <- putStrLn(s"env $env queue $queue")
        _        <- queue.offer("Give me Coffee!")
      } yield ()

      val task2 = for {                         // task to remove+print stuff from queue
        env      <- ZIO.environment[MainEnv]
        queue    <- env.mainQueue
        _        <- putStrLn(s"env $env queue $queue")
        _        <- queue.take.flatMap(putStrLn)
      } yield ()

      val program = ZIO.runtime[MainEnv]        // top level to run both tasks
        .flatMap {
          implicit rts =>
            for {
              _ <- task1.fork
              _ <- task2
            } yield ()
        }

      val runEnv = new MainEnv with Console.Live with Clock.Live

      def run(args: List[String]) =
        program.provide(runEnv).fold(_ => 1, _ => 0)
    }

/tmp/env-q/build.sbt

这是我使用的build.sbt

val ZioVersion = "1.0.0-RC13"

lazy val root = (project in file("."))
  .settings(
    organization := "example",
    name := "example",
    version := "0.0.1-SNAPSHOT",
    scalaVersion := "2.12.8",
    scalacOptions ++= Seq("-Ypartial-unification"),
    libraryDependencies ++= Seq(
      "dev.zio"                 %% "zio"                 % ZioVersion,
    ),
    addCompilerPlugin("org.spire-math" %% "kind-projector"     % "0.9.6"),
    addCompilerPlugin("com.olegpy"     %% "better-monadic-for" % "0.2.4")
  )

scalacOptions ++= Seq(
  "-deprecation",               // Emit warning and location for usages of deprecated APIs.
  "-encoding", "UTF-8",         // Specify character encoding used by source files.
  "-language:higherKinds",      // Allow higher-kinded types
  "-language:postfixOps",       // Allows operator syntax in postfix position (deprecated since Scala 2.10)
  "-feature",                   // Emit warning and location for usages of features that should be imported explicitly.
  "-Ypartial-unification",      // Enable partial unification in type constructor inference
  "-Xfatal-warnings",           // Fail the compilation if there are any warnings
)

Answer 1

在ZIO核心的官方Gitter通道中，Adam建议

您可能希望您的环境只有一个Queue[String]，然后您希望使用provideM和Queue.unbounded这样的方法来创建一个队列，并将其提供给整个应用程序。这就是provideM (相对于provide )的优势所在。它让我们通过提供一个A来满足需要ZIO[A]的环境。

对ZIO源代码的深入研究揭示了DefaultTestReporterSpec.scala中一个有用的例子。

将环境定义为

  trait MainEnv extends Console with Clock    // environment with queue
  {
    val mainQueue: Queue[String]
  }

将任务更改为使用=访问=，而不是使用<- (因为mainQueue现在是Queue[String]，而不是UIO[Queue[String]]，删除runEnv并更改测试中的run方法以使用provideSomeM )

  def run(args: List[String]) =
    program.provideSomeM(
      for {
        q <- Queue.unbounded[String]
      } yield new MainEnv with Console.Live with Clock.Live {
        override val mainQueue = q
      }
    ).fold(_ => 1, _ => 0)

我得到了预期的结果：

sbt:example> run
[info] Running example.Main 
env example.Main$$anon$1@45bfc0da queue zio.Queue$$anon$1@13b73d56
env example.Main$$anon$1@45bfc0da queue zio.Queue$$anon$1@13b73d56
Give me Coffee!
[success] Total time: 1 s, completed Oct 1, 2019 7:41:47 AM