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 = 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 = Monitor.new, queue_cv = queue_mon.new_cond) ⇒ OnMemorySortedQueue

Returns a new instance of OnMemorySortedQueue.

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

def initialize(policy, queue_mon = 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 1585

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 1592

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 1563

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 1573

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