Gecko [ fernandoquevedodop.com ]

Name	Size	Permission
deferrable	[ DIR ]	drwxr-xr-x
protocols	[ DIR ]	drwxr-xr-x
buftok.rb	2.13 KB	-rw-r--r--
callback.rb	2.14 KB	-rw-r--r--
channel.rb	1.53 KB	-rw-r--r--
completion.rb	9.71 KB	-rw-r--r--
connection.rb	33.41 KB	-rw-r--r--
deferrable.rb	7.8 KB	-rw-r--r--
file_watch.rb	2 KB	-rw-r--r--
future.rb	1.83 KB	-rw-r--r--
io_streamer.rb	1.86 KB	-rw-r--r--
iterator.rb	7.04 KB	-rw-r--r--
messages.rb	2.77 KB	-rw-r--r--
pool.rb	4.1 KB	-rw-r--r--
process_watch.rb	1.24 KB	-rw-r--r--
processes.rb	3.66 KB	-rw-r--r--
protocols.rb	1.5 KB	-rw-r--r--
pure_ruby.rb	34.87 KB	-rw-r--r--
queue.rb	2.06 KB	-rw-r--r--
resolver.rb	4.76 KB	-rw-r--r--
spawnable.rb	2.12 KB	-rw-r--r--
streamer.rb	3.59 KB	-rw-r--r--
threaded_resource.rb	2.8 KB	-rw-r--r--
tick_loop.rb	2.09 KB	-rw-r--r--
timers.rb	1.29 KB	-rw-r--r--
version.rb	50 B	-rw-r--r--

Code Editor : queue.rb

module EventMachine
  # A cross thread, reactor scheduled, linear queue.
  #
  # This class provides a simple queue abstraction on top of the reactor
  # scheduler. It services two primary purposes:
  #
  # * API sugar for stateful protocols
  # * Pushing processing onto the reactor thread
  #
  # @example
  #
  #  q = EM::Queue.new
  #  q.push('one', 'two', 'three')
  #  3.times do
  #    q.pop { |msg| puts(msg) }
  #  end
  #
  class Queue
    def initialize
      @sink  = []
      @drain = []
      @popq  = []
    end

# Pop items off the queue, running the block on the reactor thread. The pop
    # will not happen immediately, but at some point in the future, either in
    # the next tick, if the queue has data, or when the queue is populated.
    #
    # @return [NilClass] nil
    def pop(*a, &b)
      cb = EM::Callback(*a, &b)
      EM.schedule do
        if @drain.empty?
          @drain = @sink
          @sink = []
        end
        if @drain.empty?
          @popq << cb
        else
          cb.call @drain.shift
        end
      end
      nil # Always returns nil
    end

# Push items onto the queue in the reactor thread. The items will not appear
    # in the queue immediately, but will be scheduled for addition during the
    # next reactor tick.
    def push(*items)
      EM.schedule do
        @sink.push(*items)
        unless @popq.empty?
          @drain = @sink
          @sink = []
          @popq.shift.call @drain.shift until @drain.empty? || @popq.empty?
        end
      end
    end
    alias :<< :push

# @return [Boolean]
    # @note This is a peek, it's not thread safe, and may only tend toward accuracy.
    def empty?
      @drain.empty? && @sink.empty?
    end

# @return [Integer] Queue size
    # @note This is a peek, it's not thread safe, and may only tend toward accuracy.
    def size
      @drain.size + @sink.size
    end

# @return [Integer] Waiting size
    # @note This is a peek at the number of jobs that are currently waiting on the Queue
    def num_waiting
      @popq.size
    end

end # Queue
end # EventMachine

${this.title}

Code Editor : queue.rb