Contents

module Sidekiq
  module Grouping
    class Actor
      include Sidekiq::Grouping::Logging
      include ::Celluloid

      def initialize
        link_to_sidekiq_manager
      end

      private

      def start_polling
        interval = Sidekiq::Grouping::Config.poll_interval
        info "Start polling of queue batches every #{interval} seconds"
        every(interval) { flush_batches }
      end

      def flush_batches
        batches = []

        Sidekiq::Grouping::Batch.all.map do |batch|
          if batch.could_flush?
            batches << batch
          end
        end

        flush(batches)
      end

      def link_to_sidekiq_manager
        start_polling
      rescue
        info "Can't link #{self.class.name}. Sidekiq::Manager not running. Retrying in 5 seconds ..."
        after(5) { link_to_sidekiq_manager }
      end

      def flush(batches)
        if batches.any?
          names = batches.map { |batch| "#{batch.worker_class} in #{batch.queue}" }
          info "Trying to flush batched queues: #{names.join(',')}"
          batches.each { |batch| batch.flush }
        end
      end
    end
  end
end

Version data entries

2 entries across 2 versions & 1 rubygems

Version	Path
sidekiq-grouping-1.0.4	lib/sidekiq/grouping/actor.rb
sidekiq-grouping-1.0.3	lib/sidekiq/grouping/actor.rb