Class: EventMachine::Queue

Inherits:
Object
  • Object
show all
Defined in:
lib/em/queue.rb

Overview

A cross thread, reactor scheduled, linear queue.

This class provides a simple queue abstraction on top of the reactor scheduler. It services two primary purposes:

  • API sugar for stateful protocols
  • Pushing processing onto the reactor thread

Examples:


q = EM::Queue.new
q.push('one', 'two', 'three')
3.times do
  q.pop { |msg| puts(msg) }
end

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Queue

Returns a new instance of Queue.



19
20
21
22
 
# File 'lib/em/queue.rb', line 19

def initialize
  @items = []
  @popq  = []
end

Instance Method Details

#empty? ⇒ Boolean

Note:

This is a peek, it's not thread safe, and may only tend toward accuracy.

Returns:

  • (Boolean) 


54
55
56
 
# File 'lib/em/queue.rb', line 54

def empty?
  @items.empty?
end

#num_waiting ⇒ Integer

Note:

This is a peek at the number of jobs that are currently waiting on the Queue

Returns Waiting size.

Returns:

  • (Integer) —

    Waiting size



66
67
68
 
# File 'lib/em/queue.rb', line 66

def num_waiting
  @popq.size
end

#pop(*a, &b) ⇒ NilClass

Pop items off the queue, running the block on the reactor thread. The pop will not happen immediately, but at some point in the future, either in the next tick, if the queue has data, or when the queue is populated.

Returns:

  • (NilClass) —

    nil



29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/em/queue.rb', line 29

def pop(*a, &b)
  cb = EM::Callback(*a, &b)
  EM.schedule do
    if @items.empty?
      @popq << cb
    else
      cb.call @items.shift
    end
  end
  nil # Always returns nil
end

#push(*items) ⇒ Object Also known as: <<

Push items onto the queue in the reactor thread. The items will not appear in the queue immediately, but will be scheduled for addition during the next reactor tick.



44
45
46
47
48
49
 
# File 'lib/em/queue.rb', line 44

def push(*items)
  EM.schedule do
    @items.push(*items)
    @popq.shift.call @items.shift until @items.empty? || @popq.empty?
  end
end

#size ⇒ Integer

Note:

This is a peek, it's not thread safe, and may only tend toward accuracy.

Returns Queue size.

Returns:

  • (Integer) —

    Queue size



60
61
62
 
# File 'lib/em/queue.rb', line 60

def size
  @items.size
end