Class: Proco::Queue::BatchQueue

Inherits:
Base
  • Object
show all
Defined in:
lib/proco/queue/batch_queue.rb

Instance Method Summary collapse

Methods inherited from Base

#invalidate, #push, #take

Methods included from MT::Base

#broadcast, #do_when, #signal, #synchronize, #try_when, #wait_until

Constructor Details

#initialize(size, batch_size, delay) ⇒ BatchQueue 



7
8
9
10
11
 
# File 'lib/proco/queue/batch_queue.rb', line 7

def initialize size, batch_size, delay
  super size, delay
  @futures = []
  @batch_size = batch_size
end

Instance Method Details

#push_impl(item) ⇒ Object 



13
14
15
16
17
18
19
 
# File 'lib/proco/queue/batch_queue.rb', line 13

def push_impl item
  @items << item
  if @items.length % @batch_size == 1
    @futures << Future.new
  end
  @futures.last
end

#take_impl ⇒ Object 



21
22
23
24
25
26
 
# File 'lib/proco/queue/batch_queue.rb', line 21

def take_impl
  items  = @items[0, @batch_size]
  @items = @items[@batch_size..-1] || []

  [@futures.shift, items]
end