Sha256: 24d6810c076ba0bf76db9b7702be4d0e38a4cfd0bd44c028bfb29922be96c9a5

Contents?: true

Size: 936 Bytes

Versions: 4

Compression:

Stored size: 936 Bytes

Contents

# frozen_string_literal: true

require "activerecord-postgres_pub_sub"

module SidekiqPublisher
  class Runner
    LISTENER_TIMEOUT_SECONDS = 60
    CHANNEL_NAME = "sidekiq_publisher_job"

    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

    attr_reader :publisher

    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

4 entries across 4 versions & 1 rubygems

Version Path
sidekiq_publisher-1.6.4 lib/sidekiq_publisher/runner.rb
sidekiq_publisher-1.6.3 lib/sidekiq_publisher/runner.rb
sidekiq_publisher-1.6.2 lib/sidekiq_publisher/runner.rb
sidekiq_publisher-1.6.1 lib/sidekiq_publisher/runner.rb