Class: Fairy::OnMemorySortedQueue

Inherits:

Object

• Object
• Fairy::OnMemorySortedQueue

Defined in:

lib/fairy/share/port.rb

Direct Known Subclasses

SortedQueue1

Instance Method Summary

• #initialize(policy, queue_mon = XThread::Monitor.new, queue_cv = queue_mon.new_cond) ⇒ OnMemorySortedQueue constructor

  A new instance of OnMemorySortedQueue.

• #pop ⇒ Object
• #pop_all ⇒ Object
• #push(e) ⇒ Object
• #push_on_eos ⇒ Object

Constructor Details

#initialize(policy, queue_mon = XThread::Monitor.new, queue_cv = queue_mon.new_cond) ⇒ OnMemorySortedQueue

Returns a new instance of OnMemorySortedQueue.

# File 'lib/fairy/share/port.rb', line 1553

def initialize(policy, queue_mon = XThread::Monitor.new, queue_cv = queue_mon.new_cond)
  @policy = policy

  @pool_threshold = policy[:pool_threshold]
  @pool_threshold ||= CONF.SORTEDQUEUE_POOL_THRESHOLD

  @push_queue = []
  @pop_queue = nil

  @queue_mutex = queue_mon
  @queue_cv = queue_cv

  @sort_by = policy[:sort_by]
  @sort_by ||= CONF.SORTEDQUEUE_SORTBY   

  if @sort_by.kind_of?(String)
  @sort_by = eval("proc{#{@sort_by}}")
  end
end

Instance Method Details

#pop ⇒ Object

# File 'lib/fairy/share/port.rb', line 1595

def pop
  @queue_mon.synchronize do
  @queue_cv.wait_while{@pop_queue.nil?}
  @pop_queue.shift
  end
end

#pop_all ⇒ Object

# File 'lib/fairy/share/port.rb', line 1602

def pop_all
  @queue_mon.synchronize do
  @queue_cv.wait_while{@pop_queue.nil?}
  @pop_queue.shift(@pool_threshold)
  end
end

#push(e) ⇒ Object

# File 'lib/fairy/share/port.rb', line 1573

def push(e)
  @queue_mon.synchronize do
  @push_queue.push e
  if e == :END_OF_STREAM
    @push_queue.pop
    push_on_eos
  end
  end
end

#push_on_eos ⇒ Object

# File 'lib/fairy/share/port.rb', line 1583

def push_on_eos
  begin
  @pop_queue = @push_queue.sort_by{|e| @sort_by.call(e)}
  @pop_queue.push :END_OF_STREAM
  @push_queue.clear
  @push_queue = nil
  rescue
  Log::debug_exception
  end
  @queue_cv.broadcast
end