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

- (Queue) initialize

A new instance of Queue



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

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

Instance Method Details

- (Boolean) empty?

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

- (Integer) num_waiting

Note:

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

Waiting size

Returns:

  • (Integer) —

    Waiting size



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

def num_waiting
  @popq.size
end

- (NilClass) pop(*a, &b)

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

- (Object) push(*items) 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

- (Integer) size

Note:

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

Queue size

Returns:

  • (Integer) —

    Queue size



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

def size
  @items.size
end