scala - 使用 fs2.Stream 对事件进行分组

Question

我有如下事件流：

sealed trait Event

val eventStream: fs2.Stream[IO, Event] = //...

我想对在一分钟内收到的这些事件进行分组（即从每分钟的 0 秒到 59 秒）。这听起来很简单fs2

val groupedEventsStream = eventStream groupAdjacentBy {event => 
    TimeUnit.MILLISECONDS.toMinutes(System.currentTimeMillis())
}

问题是分组功能不纯。它使用currentTimeMillis. 我可以按如下方式解决这个问题：

stream.evalMap(t => IO(System.currentTimeMillis(), t))
  .groupAdjacentBy(t => TimeUnit.MILLISECONDS.toMinutes(t._1))

问题是我想避免使用带有元组的笨拙样板。还有其他解决方案吗？

或者，对于这种情况，使用不纯函数可能不是那么糟糕？

Answer 1

您可以使用cat.effect.Clock删除一些样板：

def groupedEventsStream[A](stream: fs2.Stream[IO, A])
                          (implicit clock: Clock[IO], eq: Eq[Long]): fs2.Stream[IO, (Long, Chunk[(Long, A)])] =
  stream.evalMap(t => clock.realTime(TimeUnit.MINUTES).map((_, t)))
    .groupAdjacentBy(_._1)