BulkJob helpers

collection.rb

module BulkJob
  class Collection
    def with_worker(worker, &block)
      begin
        @orig_worker = @worker
        @worker = worker
        block.call
      ensure
        @worker = @orig_worker
      end
    end

    def add_job(*args, worker: nil, **hash_args)
      worker = @worker if worker.nil?
      raise ArgumentError, 'worker required' if worker.nil?

      jobs[worker.to_s] << (hash_args.empty? ? args : args + [hash_args])
    end

    def to_h
      {
        'class' => BulkJob::ProxyWorker,
        'args' => jobs_args
      }
    end

    def jobs
      @jobs ||= Hash.new do |hash, key|
        hash[key] = []
      end
    end

    private
    def jobs_args
      args = []
      jobs.each do |worker, job_ary|
        job_ary.each do |job|
          args << [worker.to_s, *job]
        end
      end
      args
    end
  end
end

collection_spec.rb

require 'spec_helper'

describe BulkJob::Collection do
  let(:collection) { BulkJob::Collection.new }

  describe '#add_job' do
    it 'adds the job with the specified worker' do
      collection.add_job(1, 2, worker: 'FooBar')
      expect(collection.jobs['FooBar'].first).to eq [1, 2]
    end

    it 'requires a worker' do
      expect { collection.add_job(1) }.to raise_error(ArgumentError, 'worker required')
    end

    it 'accepts a hash' do
      collection.add_job(foo: :bar, worker: 'FooBar')
      expect(collection.jobs['FooBar'].first).to eq [{foo: :bar}]
    end

    it 'adds a job with no arguments' do
      collection.add_job(worker: 'FooBar')
      expect(collection.jobs['FooBar'].first).to eq []
    end

    context 'using #with_worker' do
      around do |example|
        collection.with_worker('FooBar') do
          example.run
        end
      end

      it 'adds the job' do
        collection.add_job(1, 2)
        expect(collection.jobs['FooBar'].first).to eq [1, 2]
      end

      it 'overrides the worker when passed as an argument' do
        collection.add_job(1, 2, worker: 'BestWorker')
        expect(collection.jobs['FooBar']).to be_empty
        expect(collection.jobs['BestWorker'].first).to eq [1, 2]
      end
    end
  end

  describe '#to_h' do
    it 'returns a hash suitable for sidekiq #push_bulk' do
      collection.add_job(1, 2, worker: 'FooBar')
      collection.add_job('asdf', worker: 'FooBar')
      collection.add_job(foo: :bar, worker: 'AwesomeWorker')

      result = collection.to_h
      expect(result['class']).to eq BulkJob::ProxyWorker
      expect(result['args']).to eq [['FooBar', 1, 2], ['FooBar', 'asdf'], ['AwesomeWorker', {foo: :bar}]]
    end
  end
end

proxy_worker.rb

module BulkJob
  class ProxyWorker
    include Sidekiq::Worker

    def perform(klass, *args)
      klass.constantize.new.send(:perform, *args)
    end
  end
end

proxy_worker_spec.rb

require 'spec_helper'

describe BulkJob::ProxyWorker, sidekiq: :inline do
  let(:worker_class) do
    Class.new do
      include Sidekiq::Worker
      class_attribute :perform_called
      class_attribute :perform_args
      self.perform_called = false

      def perform(*args)
        self.class.perform_called = true
        self.class.perform_args = args
      end
    end
  end

  let(:worker1_class) { Class.new(worker_class) }
  let(:worker2_class) { Class.new(worker_class) }

  let(:worker1) { worker1_class.new }
  let(:worker2) { worker2_class.new }

  before do
    # Define temporary constants to support sidekiq's use of #constantize
    stub_const('BulkProxyTest1Worker', worker1_class)
    stub_const('BulkProxyTest2Worker', worker2_class)
  end

  it 'proxies to different workers for push_bulk' do
    result_ary = Sidekiq::Client.push_bulk({
      'class' => BulkJob::ProxyWorker,
      'args' => [
        [BulkProxyTest1Worker.to_s, 1, 2],
        [BulkProxyTest2Worker.to_s, {foo: :bar}]
      ]
    })

    expect(result_ary).to_not be_nil
    expect(result_ary.size).to eq 2

    expect(worker1_class.perform_called).to be_true
    expect(worker2_class.perform_called).to be_true

    expect(worker1_class.perform_args).to eq [1, 2]
    expect(worker2_class.perform_args).to eq [{'foo' => 'bar'}]
  end
end