Method: Que::Locker#initialize

Defined in:

lib/que/locker.rb

#initialize(queues: [Que.default_queue], connection_url: nil, listen: true, poll: true, poll_interval: DEFAULT_POLL_INTERVAL, wait_period: DEFAULT_WAIT_PERIOD, maximum_buffer_size: DEFAULT_MAXIMUM_BUFFER_SIZE, minimum_buffer_size: DEFAULT_MINIMUM_BUFFER_SIZE, worker_priorities: DEFAULT_WORKER_PRIORITIES, on_worker_start: nil) ⇒ Locker

Returns a new instance of Locker.

# File 'lib/que/locker.rb', line 52

def initialize(
  queues:              [Que.default_queue],
  connection_url:      nil,
  listen:              true,
  poll:                true,
  poll_interval:       DEFAULT_POLL_INTERVAL,
  wait_period:         DEFAULT_WAIT_PERIOD,
  maximum_buffer_size: DEFAULT_MAXIMUM_BUFFER_SIZE,
  minimum_buffer_size: DEFAULT_MINIMUM_BUFFER_SIZE,
  worker_priorities:   DEFAULT_WORKER_PRIORITIES,
  on_worker_start:     nil
)

  # Sanity-check all our arguments, since some users may instantiate Locker
  # directly.
  Que.assert [TrueClass, FalseClass], listen
  Que.assert [TrueClass, FalseClass], poll

  Que.assert Numeric, poll_interval
  Que.assert Numeric, wait_period

  Que.assert Array, worker_priorities
  worker_priorities.each { |p| Que.assert([Integer, NilClass], p) }

  # We use a JobBuffer to track jobs and pass them to workers, and a
  # ResultQueue to receive messages from workers.
  @job_buffer = JobBuffer.new(
    maximum_size: maximum_buffer_size,
    minimum_size: minimum_buffer_size,
    priorities:   worker_priorities.uniq,
  )

  @result_queue = ResultQueue.new

  @stop = false

  Que.internal_log :locker_instantiate, self do
    {
      queues:              queues,
      listen:              listen,
      poll:                poll,
      poll_interval:       poll_interval,
      wait_period:         wait_period,
      maximum_buffer_size: maximum_buffer_size,
      minimum_buffer_size: minimum_buffer_size,
      worker_priorities:   worker_priorities,
    }
  end

  # Local cache of which advisory locks are held by this connection.
  @locks = Set.new

  @poll_interval = poll_interval

  if queues.is_a?(Hash)
    @queue_names = queues.keys
    @queues = queues.transform_values do |interval|
      interval || poll_interval
    end
  else
    @queue_names = queues
    @queues = queues.map do |queue_name|
      [queue_name, poll_interval]
    end.to_h
  end

  @wait_period = wait_period.to_f / 1000 # Milliseconds to seconds.

  @workers =
    worker_priorities.map do |priority|
      Worker.new(
        priority:       priority,
        job_buffer:     @job_buffer,
        result_queue:   @result_queue,
        start_callback: on_worker_start,
      )
    end

  # To prevent race conditions, let every worker get into a ready state
  # before starting up the locker thread.
  loop do
    break if job_buffer.waiting_count == workers.count
    sleep 0.001
  end

  # If we weren't passed a specific connection_url, borrow a connection from
  # the pool and derive the connection string from it.
  connection_args =
    if connection_url
      uri = URI.parse(connection_url)

      opts =
        {
          host:     uri.host,
          user:     uri.user,
          password: uri.password,
          port:     uri.port || 5432,
          dbname:   uri.path[1..-1],
        }

      if uri.query
        opts.merge!(Hash[uri.query.split("&").map{|s| s.split('=')}.map{|a,b| [a.to_sym, b]}])
      end

      opts
    else
      Que.pool.checkout do |conn|
        c = conn.wrapped_connection

        {
          host:     c.host,
          user:     c.user,
          password: c.pass,
          port:     c.port,
          dbname:   c.db,
        }
      end
    end

  @connection = Que::Connection.wrap(PG::Connection.open(connection_args))

  @thread =
    Thread.new do
      # An error causing this thread to exit is a bug in Que, which we want
      # to know about ASAP, so propagate the error if it happens.
      Thread.current.abort_on_exception = true

      # Give this thread priority, so it can promptly respond to NOTIFYs.
      Thread.current.priority = 1

      begin
        unless connection_args.has_key?(:application_name)
          @connection.execute(
            "SELECT set_config('application_name', $1, false)",
            ["Que Locker: #{@connection.backend_pid}"]
          )
        end

        Poller.setup(@connection)

        @listener =
          if listen
            Listener.new(connection: @connection)
          end

        @pollers =
          if poll
            @queues.map do |queue_name, interval|
              Poller.new(
                connection:    @connection,
                queue:         queue_name,
                poll_interval: interval,
              )
            end
          end

        work_loop
      ensure
        @connection.wrapped_connection.close
      end
    end
end