Sha256: 0f61a49a8a838b3836bb06931852cf6b57d515be7ff1a85ee0c70443589ce16a
Contents?: true
Size: 968 Bytes
Versions: 17
Compression:
Stored size: 968 Bytes
Contents
# frozen_string_literal: true
require "activerecord-postgres_pub_sub"
module SidekiqPublisher
class Runner
extend PrivateAttr
LISTENER_TIMEOUT_SECONDS = 60
CHANNEL_NAME = "sidekiq_publisher_job"
private_attr_reader :publisher
def self.run
new.run
end
def initialize
@publisher = Publisher.new
end
def run
ActiveRecord::PostgresPubSub::Listener.listen(
CHANNEL_NAME,
listen_timeout: LISTENER_TIMEOUT_SECONDS
) do |listener|
listener.on_start { publisher.publish }
listener.on_notify { publisher.publish }
listener.on_timeout { listener_timeout }
end
end
private
def listener_timeout
if Job.unpublished.exists?
SidekiqPublisher.logger&.warn(
"#{self.class.name}: msg='publishing pending jobs at timeout'"
)
publisher.publish
else
Job.purge_expired_published!
end
end
end
end
Version data entries
17 entries across 17 versions & 1 rubygems