application_worker.rb 1.34 KB
Newer Older
1 2
# frozen_string_literal: true

3 4 5 6 7
Sidekiq::Worker.extend ActiveSupport::Concern

module ApplicationWorker
  extend ActiveSupport::Concern

8
  include Sidekiq::Worker # rubocop:disable Cop/IncludeSidekiqWorker
9 10

  included do
11
    set_queue
12 13 14
  end

  module ClassMethods
15 16 17 18 19 20 21
    def inherited(subclass)
      subclass.set_queue
    end

    def set_queue
      queue_name = [queue_namespace, base_queue_name].compact.join(':')

22
      sidekiq_options queue: queue_name # rubocop:disable Cop/SidekiqOptionsQueue
23 24
    end

25 26 27 28 29 30 31 32
    def base_queue_name
      name
        .sub(/\AGitlab::/, '')
        .sub(/Worker\z/, '')
        .underscore
        .tr('/', '_')
    end

33 34 35 36 37 38 39 40 41 42
    def queue_namespace(new_namespace = nil)
      if new_namespace
        sidekiq_options queue_namespace: new_namespace

        set_queue
      else
        get_sidekiq_options['queue_namespace']&.to_s
      end
    end

43 44 45
    def queue
      get_sidekiq_options['queue'].to_s
    end
46 47 48 49 50 51 52 53 54 55 56 57 58 59 60

    def bulk_perform_async(args_list)
      Sidekiq::Client.push_bulk('class' => self, 'args' => args_list)
    end

    def bulk_perform_in(delay, args_list)
      now = Time.now.to_i
      schedule = now + delay.to_i

      if schedule <= now
        raise ArgumentError, 'The schedule time must be in the future!'
      end

      Sidekiq::Client.push_bulk('class' => self, 'args' => args_list, 'at' => schedule)
    end
61 62
  end
end