Class: Karafka::Web::Processing::Consumers::Aggregators::State::Steps::EvictExpiredProcesses

Inherits:
Base
  • Object
show all
Defined in:
lib/karafka/web/processing/consumers/aggregators/state/steps/evict_expired_processes.rb

Overview

Note:

We do not evict based on status (stopped), because we want to report the stopped processes for extra time within the ttl limitations. This makes tracking of things from UX perspective nicer.

Evicts expired processes from the current state.

We consider processes dead if they do not report often enough.

Instance Method Summary collapse

Methods inherited from Base

#initialize

Constructor Details

This class inherits a constructor from Karafka::Web::Processing::Consumers::Aggregators::State::Steps::Base

Instance Method Details

#call ⇒ Object

Deletes ttl-expired entries from context.state[:processes] and context.active_reports



20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/karafka/web/processing/consumers/aggregators/state/steps/evict_expired_processes.rb', line 20

def call
  max_ttl = context.aggregated_from - (::Karafka::Web.config.ttl / 1_000.0)

  context.state[:processes].delete_if do |_id, details|
    details[:dispatched_at] < max_ttl
  end

  context.active_reports.delete_if do |_id, details|
    details[:dispatched_at] < max_ttl
  end
end