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.

Direct Known Subclasses

PersistentWorkQueue

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
 
# File 'lib/fractor/work_queue.rb', line 18

def <<(work_item)
  enqueue(work_item)
end

#dequeue ⇒ Fractor::Work?

Retrieve a single work item from the queue (blocking if empty)

Returns:



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

def dequeue
  @queue.pop
end

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

Retrieve multiple work items from the queue in a single operation (standardized API)

Parameters:

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

    Maximum number of items to retrieve

Returns:



44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/fractor/work_queue.rb', line 44

def dequeue_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

#empty? ⇒ Boolean

Check if the queue is empty

Returns:

  • (Boolean) —

    true if the queue is empty



67
68
69
 
# File 'lib/fractor/work_queue.rb', line 67

def empty?
  @queue.empty?
end

#enqueue(work_item) ⇒ void

This method returns an undefined value.

Add a work item to the queue (standardized API)

Parameters:



25
26
27
28
29
30
31
32
 
# File 'lib/fractor/work_queue.rb', line 25

def enqueue(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

#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:



37
38
39
 
# File 'lib/fractor/work_queue.rb', line 37

def pop_batch(max_items = 10)
  dequeue_batch(max_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



81
82
83
84
85
86
 
# File 'lib/fractor/work_queue.rb', line 81

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



73
74
75
 
# File 'lib/fractor/work_queue.rb', line 73

def size
  @queue.size
end