Class: Qpid::Proton::Schedule

Inherits:
Object
  • Object
show all
Includes:
TimeCompare
Defined in:
lib/util/schedule.rb

Overview

A sorted, thread-safe list of scheduled Proc. Note: calls to #process are always serialized, but calls to #add may be concurrent.

Defined Under Namespace

Classes: Item

Instance Method Summary collapse

Methods included from TimeCompare

#before_eq, #earliest

Constructor Details

#initialize ⇒ Schedule

Returns a new instance of Schedule.



36
37
38
39
40
 
# File 'lib/util/schedule.rb', line 36

def initialize()
  @lock = Mutex.new
  @items = []
  @closed = false
end

Instance Method Details

#add(time, non_block = false, &proc) ⇒ Object

Returns true if the Schedule was previously empty.

Returns:

  • true if the Schedule was previously empty

Raises:

  • EOFError if schedule is closed

  • ThreadError if non_block and operation would block



49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/util/schedule.rb', line 49

def add(time, non_block=false, &proc)
  # non_block ignored for now, but we may implement a bounded schedule in future.
  @lock.synchronize do
    raise EOFError if @closed
    if at = (0...@items.size).bsearch { |i| @items[i].time > time }
      @items.insert(at, Item.new(time, proc))
    else
      @items << Item.new(time, proc)
    end
    return @items.size == 1
  end
end

#close ⇒ Object

#add raises EOFError after #close. #process can still be called to drain the schedule.



75
76
77
 
# File 'lib/util/schedule.rb', line 75

def close()
  @lock.synchronize { @closed = true }
end

#next_tick ⇒ Object 



42
43
44
 
# File 'lib/util/schedule.rb', line 42

def next_tick()
  @lock.synchronize { @items.first.time unless @items.empty? }
end

#process(now) ⇒ Object

Returns true if the Schedule became empty as a result of this call.

Returns:

  • true if the Schedule became empty as a result of this call



63
64
65
66
67
68
69
70
71
 
# File 'lib/util/schedule.rb', line 63

def process(now)
  due = []
  empty = @lock.synchronize do
    due << @items.shift while (!@items.empty? && before_eq(@items.first.time, now))
    @items.empty?
  end
  due.each { |i| i.proc.call() }
  return empty && !due.empty?
end