rails/activejob/test/support/integration/adapters/sneakers.rb

require "sneakers/runner"
require "sneakers/publisher"
require "timeout"

module Sneakers
  class Publisher
    def safe_ensure_connected
      @mutex.synchronize do
        ensure_connection! unless connected?
      end
    end
  end
end

module SneakersJobsManager
  def setup
    ActiveJob::Base.queue_adapter = :sneakers
    Sneakers.configure  heartbeat: 2,
                        amqp: "amqp://guest:guest@localhost:5672",
                        vhost: "/",
                        exchange: "active_jobs_sneakers_int_test",
                        exchange_type: :direct,
                        daemonize: true,
                        threads: 1,
                        workers: 1,
                        pid_path: Rails.root.join("tmp/sneakers.pid").to_s,
                        log: Rails.root.join("log/sneakers.log").to_s
    unless can_run?
      puts "Cannot run integration tests for sneakers. To be able to run integration tests for sneakers you need to install and start rabbitmq.\n"
      exit
    end
  end

  def clear_jobs
    bunny_queue.purge
  end

  def start_workers
    @pid = fork do
      queues = %w(integration_tests)
      workers = queues.map do |q|
        worker_klass = "ActiveJobWorker"+Digest::MD5.hexdigest(q)
        Sneakers.const_set(worker_klass, Class.new(ActiveJob::QueueAdapters::SneakersAdapter::JobWrapper) do
          from_queue q
        end)
      end
      Sneakers::Runner.new(workers).run
    end
    begin
      Timeout.timeout(10) do
        while bunny_queue.status[:consumer_count] == 0
          sleep 0.5
        end
      end
    rescue Timeout::Error
      stop_workers
      raise "Failed to start sneakers worker"
    end
  end

  def stop_workers
    Process.kill "TERM", @pid
    Process.kill "TERM", File.open(Rails.root.join("tmp/sneakers.pid").to_s).read.to_i
  rescue
  end

  def can_run?
    begin
      bunny_publisher
    rescue
      return false
    end
    true
  end

  protected
    def bunny_publisher
      @bunny_publisher ||= begin
        p = ActiveJob::QueueAdapters::SneakersAdapter::JobWrapper.send(:publisher)
        p.safe_ensure_connected
        p
      end
    end

    def bunny_queue
      @queue ||= bunny_publisher.exchange.channel.queue "integration_tests", durable: true
    end
end