Class: SendQ

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

Overview

SendQ - Easy access to time-constrained send queues.

Example:

output = [] 
# process a maximum of 5 within 3 seconds
s = SendQ.new(5, 3) { |x| output.push(x) }
(0..9).each { |x| s.add(x) }
s.run
output # => [0,1,2,3,4]
s.queue # => [5,6,7,8,9]
s.run # nothing changes, we're still waiting to run
sleep 3 # wait for the constraint to expire
s.run
output # => [0,1,2,3,4,5,6,7,8,9] 
s.queue # => []

Note: The calculation is very simple and can probably be gamed. It simply tests if the last run, subtracted from the current time, is greater than the interval.

The clock is by default the unix epoch and therefore is restricted to a minimum granularity of 1 second. If you wish to change this or otherwise manipulate the constraint, see SendQ.clock.

Simple accessors (all can be modified):

  • queue: this is the internal queue that will be exhausted on runs. See SendQ#add.

  • filled: this is the number filled in the current interval. guaranteed to never be greater than max_per_interval

  • action: this is the block you supplied to the constructor.

  • max_per_interval: the maximum number of queue elements to exhaust in the timeframe of the interval.

–

Please see the COPYING file in the source distribution for copyright information.

++

Constant Summary collapse

VERSION = 
"0.0.1"

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(max_per_interval, interval = 1, &action) ⇒ SendQ

Constructor.

  • max_per_interval (required): the maximum amount of queue elements to process in a given interval.

  • interval (optional, default 1 second): if this much time has passed, reset how many can be filled in the next run to max_per_interval. See SendQ.clock for more information on how to manipulate the clock.

  • action (required): A block that describes what you want to do with each element in the queue. This block must take one argument - the element of the queue to process.

Raises:

  • (ArgumentError) 


57
58
59
60
61
62
63
64
65
66
 
# File 'lib/sendq.rb', line 57

def initialize(max_per_interval, interval = 1, &action)
    raise ArgumentError, "Requires an action block to construct" unless action

    @interval = interval
    @last_run = 0
    @max_per_interval = max_per_interval
    @queue    = []
    @filled   = 0
    @action   = action
end

Instance Attribute Details

#action ⇒ Object

Returns the value of attribute action.



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

def action
  @action
end

#filled ⇒ Object

Returns the value of attribute filled.



46
47
48
 
# File 'lib/sendq.rb', line 46

def filled
  @filled
end

#max_per_interval ⇒ Object

Returns the value of attribute max_per_interval.



48
49
50
 
# File 'lib/sendq.rb', line 48

def max_per_interval
  @max_per_interval
end

#queue ⇒ Object

Returns the value of attribute queue.



45
46
47
 
# File 'lib/sendq.rb', line 45

def queue
  @queue
end

Instance Method Details

#add(item) ⇒ Object

Add an item to the queue.

Like any normal queue, items are processed in FIFO order.



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

def add(item)
    @queue.push(item)
end

#run ⇒ Object

Evaluate how many items to process in the queue and then process them. Update the clock with its current value.



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

def run
    fill     = [how_many, @queue.size].min
    @filled += fill
    fill.times do @action.call(@queue.shift) end
    @last_run = clock
end