ActiveJobsでキューを検査するにはどうすればよいですか？

Question

ジョブをエンキューする前に、キューを調べて、同じ引数を指定してジョブがキューにすでに存在するかどうかを確認したい場合は、そのジョブをエンキューしません。しかし、私はどのようにこれを行うことができなければならないのか分かりません。出来ますか？

私はTestHelperを使用して自分のテストで簡単に行うことができます。 TestHelperは本番環境では当然使用していないTestAdapterに依存しています。

少し背景。 APIでは、リクエストごとにクライアントのバージョン番号を取得します。私たちはサポートのためにIntercomを使用しており、サポートの問題に対処する際にお客様がどのバージョンを使用しているかを見ることができるように、Intercomにアプリケーションバージョンを提示したいと考えています。しかし、Intercomへの呼び出し回数を制限するために、Intercomへの各投稿を数分遅らせ、投稿がエンキューされている間は同じデータで新しいものをエンキューしたくありません。

私の質問はList queued tasks with ActiveJob AsyncAdapterに関連していますが、その質問はエンキューされたジョブの数のみを扱っています。

Efficiently reschedule ActiveJob (resque/sidekiq)は、これは不可能であることを示し、ソリューションを個別に実装する必要があります。

ActiveJobsを使用して待ち行列とその中のジョブを何とか調べることができますか、キューに入れられているものと実行されたものを把握する必要がありますか？

Answer 1

以下は未実装です。現時点ではSidekiq（Sidekiqを使用していると仮定しています）のみサポートしています。

注：部分的にしかテストしていないため、以下の項目を更新する必要があります。私は時間があればこれを完全にチェックします。

class ActiveJob::Base 
    def self.where(jid: nil, arguments: []) 
    found_jobs = [] 

    job_adapter = Rails.application.config.active_job.queue_adapter 

    case job_adapter 

    when :sidekiq 
     queue = Sidekiq::Queue.new(queue_name) 

     if jid.present? 
     job = queue.find_job(jid) 

     if job 
      # if arguments supplied, check also if args match 
      if arguments.present? 
      should_disregard_job_class = self == ActiveJob::Base 
      job_has_same_class = self.to_s == job.args[0]['job_class'] 
      job_has_same_arguments = arguments == job.args[0]['arguments'][3..-1] 

      if (should_disregard_job_class || job_has_same_class) && job_has_same_arguments 
       found_jobs << job 
      end 

      else 
      found_jobs << job 
      end 
     end 

     else 
     # TODO: optimise below as it is slow 
     queue.each do |job| 
      should_disregard_job_class = self == ActiveJob::Base 
      job_has_same_class = self.to_s == job.args[0]['job_class'] 
      job_has_same_arguments = arguments == job.args[0]['arguments'][3..-1] 

      if (should_disregard_job_class || job_has_same_class) && job_has_same_arguments 
      found_jobs << job 
      end 
     end 
     end 

    when :resque 
     # TODO 
    else 
     raise "TODO: missing Adapter implementation for #{job_adapter}" 
    end 

    found_jobs 
    end 
end 

使用

jobs = ActiveJob::Base.where(jid: '12345678abcdef') 
jobs = MyCustomJob.where(jid: '12345678abcdef') 
jobs = MyCustomJob.where(arguments: ['firstjobargumentvalue', 'secondjobargumentvalue'])