mperham--sidekiq/test/test_actors.rb

# frozen_string_literal: true
require_relative 'helper'
require 'sidekiq/cli'
require 'sidekiq/fetch'
require 'sidekiq/scheduled'
require 'sidekiq/processor'

describe 'Actors' do
  class JoeWorker
    include Sidekiq::Worker
    def perform(slp)
      raise "boom" if slp == "boom"
      sleep(slp) if slp > 0
      $count += 1
    end
  end

  before do
    Sidekiq.redis {|c| c.flushdb}
  end

  describe 'scheduler' do
    it 'can start and stop' do
      f = Sidekiq::Scheduled::Poller.new
      f.start
      f.terminate
    end

    it 'can schedule' do
      ss = Sidekiq::ScheduledSet.new
      q = Sidekiq::Queue.new

      JoeWorker.perform_in(0.01, 0)

      assert_equal 0, q.size
      assert_equal 1, ss.size

      sleep 0.015
      s = Sidekiq::Scheduled::Poller.new
      s.enqueue
      assert_equal 1, q.size
      assert_equal 0, ss.size
      s.terminate
    end
  end

  describe 'processor' do
    before do
      $count = 0
    end

    it 'can start and stop' do
      m = Mgr.new
      f = Sidekiq::Processor.new(m, m.options)
      f.terminate
    end

    class Mgr
      attr_reader :latest_error
      attr_reader :mutex
      attr_reader :cond
      def initialize
        @mutex = ::Mutex.new
        @cond = ::ConditionVariable.new
      end
      def processor_died(inst, err)
        @latest_error = err
        @mutex.synchronize do
          @cond.signal
        end
      end
      def processor_stopped(inst)
        @mutex.synchronize do
          @cond.signal
        end
      end
      def options
        opts = { :concurrency => 3, :queues => ['default'] }
        opts[:fetch] = Sidekiq::BasicFetch.new(opts)
        opts
      end
    end

    it 'can process' do
      mgr = Mgr.new

      p = Sidekiq::Processor.new(mgr, mgr.options)
      JoeWorker.perform_async(0)

      a = $count
      p.process_one
      b = $count
      assert_equal a + 1, b
    end

    it 'deals with errors' do
      mgr = Mgr.new

      p = Sidekiq::Processor.new(mgr, mgr.options)
      JoeWorker.perform_async("boom")
      q = Sidekiq::Queue.new
      assert_equal 1, q.size

      a = $count
      mgr.mutex.synchronize do
        p.start
        mgr.cond.wait(mgr.mutex)
      end
      b = $count
      assert_equal a, b

      sleep 0.001
      assert_equal false, p.thread.status
      p.terminate(true)
      refute_nil mgr.latest_error
      assert_equal RuntimeError, mgr.latest_error.class
    end

    it 'gracefully kills' do
      mgr = Mgr.new

      p = Sidekiq::Processor.new(mgr, mgr.options)
      JoeWorker.perform_async(1)
      q = Sidekiq::Queue.new
      assert_equal 1, q.size

      a = $count
      p.start
      sleep(0.05)
      p.terminate
      p.kill(true)

      b = $count
      assert_equal a, b
      assert_equal false, p.thread.status
      refute mgr.latest_error, mgr.latest_error.to_s
    end
  end
end