I’ve solved this problem. Here’s how I did it:
Whenever an event is pushed into the producer, start a timer using Process.send_after(self(), :trigger, @inactivity_time_ms). Add the timer to the state of the producer so that we cancel the existing timer and start a new one every time we push a new event. Use handle_info to handle the message that will be sent from the timer, and inside it, use GenStage.async_notify(self(), {:producer, :done}) which will force the window to finish.
Here’s the code for the producer: Inactivity based trigger for Flow · GitHub .
