emilsoman
Flow window that triggers on inactivity
I am refactoring a time consuming part of our application using Flow. In my use case, I need a Flow pipeline in which the Flow.reduce step should be triggered when there are no incoming events for more than a specified duration of time. For example, prepare and update a report based on existing events when there are no new events for 5 minutes. Can this be done using Flow?
PS: I read about Flow.Window.session, but in session based windows, it looked like a new event has to arrive after the inactive period in order to identify a gap, which doesn’t seem to solve my problem.
Trending in Discussions
Other Trending Topics
Chat & Discussions>Discussions
Latest on Elixir Forum
Sponsor Spotlight
Build Elixir applications with speed and confidence.
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #code-sync
- #javascript
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #elixirconf-us
- #websockets
- #supervisor
- #advent-of-code
- #distillery
- #processes
- #api
- #forms
- #metaprogramming
- #performance
- #security
Our Sponsors
Build Elixir applications with speed and confidence.
Supporting innovation across the BEAM ecosystem.
We build reliable cloud platforms for business-critical systems.
Catch errors, track performance, monitor hosts and more.
Error tracking, logs, uptime, and Just Enough APM. Built for developers.
Practical resources that improve the lives of professional developers.
Bringing learning, work and compliance together since 1987.
Enabling companies to succeed by building software people love.
The team behind Membrane, Popcorn, LiveDebugger. Available for hire.
Courses that'll move you from confusion to "Aha, now I get it!"
Marked As Solved
emilsoman
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. Usehandle_infoto handle the message that will be sent from the timer, and inside it, useGenStage.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 .