Sha256: c12e1cca5bf947d69a3eaf9c32096dddbc101bc9b36d1224da0959a16815a6b5
Contents?: true
Size: 1.05 KB
Versions: 19
Compression:
Stored size: 1.05 KB
Contents
# frozen_string_literal: true
module PgEventstore
# This class pulls events from db and feeds given SubscriptionRunners
# @!visibility private
class SubscriptionRunnersFeeder
# @param config_name [Symbol]
def initialize(config_name)
@config_name = config_name
end
# @param runners [Array<PgEventstore::SubscriptionRunner>]
# @return [void]
def feed(runners)
runners = runners.select(&:running?).select(&:time_to_feed?)
return if runners.empty?
runners_query_options = runners.to_h { |runner| [runner.id, runner.next_chunk_query_opts] }
grouped_events = subscription_queries.subscriptions_events(runners_query_options)
runners.each do |runner|
runner.feed(grouped_events[runner.id]) if grouped_events[runner.id]
end
end
private
# @return [PgEventstore::Connection]
def connection
PgEventstore.connection(@config_name)
end
# @return [PgEventstore::SubscriptionQueries]
def subscription_queries
SubscriptionQueries.new(connection)
end
end
end
Version data entries
19 entries across 19 versions & 1 rubygems