Class: Shoryuken::Polling::StrictPriority

Inherits:
BaseStrategy show all
Defined in:
lib/shoryuken/polling/strict_priority.rb

Overview

A polling strategy that processes queues in strict priority order. Higher priority queues are always processed before lower priority queues. Queues are temporarily paused when no messages are found.

Instance Method Summary collapse

Methods inherited from BaseStrategy

#==, #delay

Methods included from Util

#elapsed, #fire_event, #logger, #unparse_queues, #worker_name

Constructor Details

#initialize(queues, delay = nil) ⇒ StrictPriority

Initializes a new StrictPriority polling strategy

Parameters:

  • queues (Array<String>) —

    array of queue names, with higher priority queues appearing more frequently

  • delay (Float, nil) (defaults to: nil) —

    delay in seconds before unpausing empty queues



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/shoryuken/polling/strict_priority.rb', line 13

def initialize(queues, delay = nil)
  # Priority ordering of the queues, highest priority first
  @queues = queues
            .group_by { |q| q }
            .sort_by { |_, qs| -qs.count }
            .map(&:first)

  # Pause status of the queues, default to past time (unpaused)
  @paused_until = queues
                  .each_with_object({}) { |queue, h| h[queue] = Time.at(0) }

  @delay = delay
  # next_queue/messages_found run on the dispatch thread while
  # message_processed runs on processor-completion threads (for FIFO
  # queues), so all access to the shared state is serialized.
  @mutex = Mutex.new
  # Start queues at 0
  reset_next_queue
end

Instance Method Details

#active_queues ⇒ Array<Array>

Returns the list of active (non-paused) queues with their priorities

Returns:

  • (Array<Array>) —

    array of [queue_name, priority] pairs



61
62
63
64
65
66
67
68
69
 
# File 'lib/shoryuken/polling/strict_priority.rb', line 61

def active_queues
  @mutex.synchronize do
    @queues
      .reverse
      .map.with_index(1)
      .reject { |q, _| queue_paused?(q) }
      .reverse
  end
end

#message_processed(queue) ⇒ void

This method returns an undefined value.

Called when a message from a queue has been processed

Parameters:

  • queue (String) —

    the queue name



75
76
77
78
79
80
81
82
 
# File 'lib/shoryuken/polling/strict_priority.rb', line 75

def message_processed(queue)
  @mutex.synchronize do
    if queue_paused?(queue)
      logger.debug "Unpausing #{queue}"
      @paused_until[queue] = Time.at(0)
    end
  end
end

#messages_found(queue, messages_found) ⇒ void

This method returns an undefined value.

Handles the result of polling a queue

Parameters:

  • queue (String) —

    the queue name

  • messages_found (Integer) —

    number of messages found



48
49
50
51
52
53
54
55
56
 
# File 'lib/shoryuken/polling/strict_priority.rb', line 48

def messages_found(queue, messages_found)
  @mutex.synchronize do
    if messages_found == 0
      pause(queue)
    else
      reset_next_queue
    end
  end
end

#next_queue ⇒ QueueConfiguration?

Returns the next queue to poll based on strict priority

Returns:



36
37
38
39
40
41
 
# File 'lib/shoryuken/polling/strict_priority.rb', line 36

def next_queue
  @mutex.synchronize do
    next_queue = next_active_queue
    next_queue.nil? ? nil : QueueConfiguration.new(next_queue, {})
  end
end