FS2 Stream with StateT[IO, _, _]，定期转储状态答案

【问题标题】：FS2 Stream with StateT[IO, _, _], periodically dumping stateFS2 Stream with StateT[IO, _, _]，定期转储状态
【发布时间】：2018-08-01 02:56:44
【问题描述】：

我有一个程序消耗无限的数据流。在此过程中，我想记录一些指标，它们形成一个幺半群，因为它们只是简单的总和和平均值。定期，我想在某处写出这些指标，清除它们，然后返回累积它们。我基本上有：

object Foo {
  type MetricsIO[A] = StateT[IO, MetricData, A]

  def recordMetric(m: MetricData): MetricsIO[Unit] = {
    StateT.modify(_.combine(m))
  }

  def sendMetrics: MetricsIO[Unit] = {
    StateT.modifyF { s =>
      val write: IO[Unit] = writeMetrics(s)
      write.attempt.map {
        case Left(_) => s
        case Right(_) => Monoid[MetricData].empty
      }
    }
  }
}

所以大部分执行直接使用IO，而提升使用StateT.liftF。在某些情况下，我会调用recordMetric。最后我有一个流：

val mainStream: Stream[MetricsIO, Bar] = ...

而且我想定期，每隔一分钟左右，转储指标，所以我尝试了：

val scheduler: Scheduler = ...
val sendStream =
  scheduler
    .awakeEvery[MetricsIO](FiniteDuration(1, TimeUnit.Minutes))
    .evalMap(_ => Foo.sendMetrics)

val result = mainStream.concurrently(sendStream).compile.drain

然后我执行通常的顶级程序，在启动状态下调用run，然后调用unsafeRunSync。

问题是，我只看到空指标！我怀疑这与我的 monoid 隐含地向 sendStream 提供空指标有关，但我不太明白为什么应该这样或如何解决它。也许有一种方法可以将这些sendMetrics 调用“交错”到主流中？

编辑：这是一个最小的完整可运行示例：

import fs2._
import cats.implicits._
import cats.data._
import cats.effect._
import java.util.concurrent.Executors
import scala.concurrent.ExecutionContext
import scala.concurrent.duration._

val sec = Executors.newScheduledThreadPool(4)
implicit val ec = ExecutionContext.fromExecutorService(sec)

type F[A] = StateT[IO, List[String], A]

val slowInts = Stream.unfoldEval[F, Int, Int](1) { n =>
  StateT(state => IO {
    Thread.sleep(500)
    val message = s"hello $n"
    val newState = message :: state
    val result = Some((n, n + 1))
    (newState, result)
  })
}

val ticks = Scheduler.fromScheduledExecutorService(sec).fixedDelay[F](FiniteDuration(1, SECONDS))

val slowIntsPeriodicallyClearedState = slowInts.either(ticks).evalMap[Int] {
  case Left(n) => StateT.liftF(IO(n))
  case Right(_) => StateT(state => IO {
    println(state)
    (List.empty, -1)
  })
}

现在如果我这样做：

slowInts.take(10).compile.drain.run(List.empty).unsafeRunSync

然后我得到了预期的结果——状态正确地累积到输出中。但如果我这样做：

slowIntsPeriodicallyClearedState.take(10).compile.drain.run(List.empty).unsafeRunSync

然后我看到一个空列表一直打印出来。我本来希望打印出部分列表（大约 2 个元素）。

【问题讨论】：

标签： scala monad-transformers scala-cats fs2 cats-effect

【解决方案1】：

StateT 与效果类型一起使用是不安全的，因为面对并发访问它是不安全的。相反，请考虑使用Ref（来自 fs2 或 cat-effect，具体取决于版本）。

类似这样的：

def slowInts(ref: Ref[IO, Int]) = Stream.unfoldEval[F, Int, Int](1) { n =>
  val message = s"hello $n"
  ref.modify(message :: _) *> IO {
    Thread.sleep(500)
    val result = Some((n, n + 1))
    result
  }
}

val ticks = Scheduler.fromScheduledExecutorService(sec).fixedDelay[IO](FiniteDuration(1, SECONDS))

def slowIntsPeriodicallyClearedState(ref: Ref[IO, Int] = 
  slowInts.either(ticks).evalMap[Int] {
    case Left(n) => IO.pure(n)
    case Right(_) =>
      ref.modify(_ => Nil).flatMap { case Change(previous, now) => 
        IO(println(now)).as(-1)
      }
  }

【讨论】：