Skip to content

Add a fail-safe ensuring that we don't act on old timers in groupedWithin #363

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
adamw merged 1 commit into from
Sep 24, 2025
Merged

Add a fail-safe ensuring that we don't act on old timers in groupedWithin #363

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 13 additions & 6 deletions core/src/main/scala/ox/flow/FlowOps.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -677,7 +677,7 @@ class FlowOps[+T]:
*/
def groupedWithin(n: Int, duration: FiniteDuration)(using BufferCapacity): Flow[Seq[T]] = groupedWeightedWithin(n, duration)(_ => 1)

private case object GroupingTimeout
private case class GroupingTimeout(generation: Long)

/** Chunks up the emitted elements into groups, within a time window, or limited by the cumulative weight being greater or equal to the
* `minWeight`, whatever happens first. The timeout is reset after a group is emitted. If timeout expires and the buffer is empty,
Expand All @@ -698,14 +698,18 @@ class FlowOps[+T]:
unsupervised:
val c = outer.runToChannel()
val c2 = BufferCapacity.newChannel[Seq[T]]
val timerChannel = BufferCapacity.newChannel[GroupingTimeout.type]
val timerChannel = BufferCapacity.newChannel[GroupingTimeout]
forkPropagate(c2):
var buffer = Vector.empty[T]
var accumulatedCost: Long = 0
var currentGeneration = 0L

def forkTimeout() = forkCancellable:
sleep(duration)
timerChannel.sendOrClosed(GroupingTimeout).discard
def forkTimeout() =
currentGeneration += 1
val gen = currentGeneration
forkCancellable:
sleep(duration)
timerChannel.sendOrClosed(GroupingTimeout(gen)).discard

var timeoutFork: Option[CancellableFork[Unit]] = Some(forkTimeout())

Expand All @@ -727,10 +731,13 @@ class FlowOps[+T]:
timeoutFork.foreach(_.cancelNow())
c2.error(r)
false
case timerChannel.Received(GroupingTimeout) =>
case timerChannel.Received(GroupingTimeout(gen)) if gen == currentGeneration =>
timeoutFork = None // enter 'timed out state', may stay in this state if buffer is empty
if buffer.nonEmpty then sendBufferAndForkNewTimeout()
true
case timerChannel.Received(GroupingTimeout(_)) =>
// different (older) generation, ignore - must have been sent concurrently with cancelling the timer
true
case c.Received(t) =>
buffer = buffer :+ t

Expand Down