scala - 如何登录 fs2 Stream.takeWhile

Question

我们有一个使用 fs2-kafka 从 kafka 中提取的事件流，并且当事件比给定的截止日期更新或偏移量位于分区末尾时，我们正在完成处理（这对于问题来说并不重要，但要提供有关该程序的一些上下文）。

理想情况下，我们希望在满足这些条件时记录，但takeWhile需要takeThrough纯函数O => Boolean。

我们的流是：

partitionStream
  .takeWhile(shouldProcess(processingDelay, _))
  .takeThrough(!atLogEndOffset(_, endOffsets))

显然我们可以做一个log.info内部shouldProcess，atLongEndOffset但这意味着在一个纯函数内部产生副作用，我们不希望这样做。

如果不调用两次函数（一种用于记录，另一种用于条件评估），哪种方法会更好？

谢谢！