Class: Fractor::WorkQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/fractor/work_queue.rb

Overview

Thread-safe work queue for continuous mode applications. Provides a simple interface for adding work items and retrieving them in batches, with automatic integration with Fractor::Supervisor.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ WorkQueue

Returns a new instance of WorkQueue.



10
11
12
13
 
# File 'lib/fractor/work_queue.rb', line 10

def initialize
  @queue = Thread::Queue.new
  @mutex = Mutex.new
end

Instance Attribute Details

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



8
9
10
 
# File 'lib/fractor/work_queue.rb', line 8

def queue
  @queue
end

Instance Method Details

#<<(work_item) ⇒ void

This method returns an undefined value.

Add a work item to the queue (thread-safe)

Parameters:



18
19
20
21
22
23
24
25
 
# File 'lib/fractor/work_queue.rb', line 18

def <<(work_item)
  unless work_item.is_a?(Fractor::Work)
    raise ArgumentError,
          "#{work_item.class} must be an instance of Fractor::Work"
  end

  @queue << work_item
end

#empty? ⇒ Boolean

Check if the queue is empty

Returns:

  • (Boolean) —

    true if the queue is empty



47
48
49
 
# File 'lib/fractor/work_queue.rb', line 47

def empty?
  @queue.empty?
end

#pop_batch(max_items = 10) ⇒ Array<Fractor::Work>

Retrieve multiple work items from the queue in a single operation

Parameters:

  • max_items (Integer) (defaults to: 10) —

    Maximum number of items to retrieve

Returns:



30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/fractor/work_queue.rb', line 30

def pop_batch(max_items = 10)
  items = []
  max_items.times do
    break if @queue.empty?

    begin
      items << @queue.pop(true)
    rescue ThreadError
      # Queue became empty between check and pop
      break
    end
  end
  items
end

#register_with_supervisor(supervisor, batch_size: 10) ⇒ void

This method returns an undefined value.

Register this work queue as a work source with a supervisor

Parameters:

  • supervisor (Fractor::Supervisor) —

    The supervisor to register with

  • batch_size (Integer) (defaults to: 10) —

    Number of items to retrieve per poll



61
62
63
64
65
66
 
# File 'lib/fractor/work_queue.rb', line 61

def register_with_supervisor(supervisor, batch_size: 10)
  supervisor.register_work_source do
    items = pop_batch(batch_size)
    items.empty? ? nil : items
  end
end

#size ⇒ Integer

Get the current size of the queue

Returns:

  • (Integer) —

    Number of items in the queue



53
54
55
 
# File 'lib/fractor/work_queue.rb', line 53

def size
  @queue.size
end