Sha256: fad0c60b3ddc343ec4e1f3198dccc91bb661fbf8a577fa947ce452a1ad535b97
Contents?: true
Size: 1.72 KB
Versions: 13
Compression:
Stored size: 1.72 KB
Contents
module EbDeployer
class EventPoller
include Utils
def initialize(app, env, eb_driver)
@app, @env, @eb_driver = app, env, eb_driver
end
def get_anchor
events, _ = fetch_events_from_eb(:max_records => 1)
events.first
end
def poll(from_anchor, &block)
handled = Set.new
loop do
fetch_events(from_anchor) do |events|
# events from api is latest first order
to_be_handled = []
reached_anchor = false
events.each do |event|
if digest(event) == digest(from_anchor)
reached_anchor = true
end
if !handled.include?(digest(event)) && !reached_anchor
to_be_handled << event
end
end
to_be_handled.reverse.each do |event|
yield(event)
handled << digest(event)
end
!reached_anchor
end
sleep 15
end
end
private
def digest(event)
return nil unless event
JSON.dump(event)
end
def fetch_events(from_anchor, &block)
options = {}
if from_anchor && from_anchor[:event_date]
options[:start_time] = from_anchor[:event_date].iso8601
end
events, next_token = fetch_events_from_eb(options)
should_continue = yield(events)
fetch_next(next_token, &block) if next_token && should_continue
end
def fetch_next(next_token, &block)
events, next_token = fetch_events_from_eb(:next_token => next_token)
should_continue = yield(events)
fetch_next(next_token, &block) if next_token && should_continue
end
def fetch_events_from_eb(options)
@eb_driver.fetch_events(@app, @env, options)
end
end
end
Version data entries
13 entries across 13 versions & 1 rubygems