How to do logging in fs2 Stream.takeWhile

Viewed 322

We have a Stream of events that are being pulled from kafka using fs2-kafka and we are finishing processing when the events are newer than a given deadline or the offset is at the end of the partition (this does not matter too much for the question, but to give a bit of context about the program).

We ideally would want to log when those conditions are met, but takeWhile and takeThrough require pure functions O => Boolean.

Our stream is:

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

Obviously we could do a log.info inside shouldProcess and atLongEndOffset but that would mean side effecting inside a pure function, something we would prefer not doing.

What approach would be better without calling the functions twice (one for logging and one for the condition evaluation)?

Thanks!

0 Answers
Related