Contents

module Turnpike
  # A queue.
  class Queue
    # Returns a String name.
    attr :name

    # Creates a new queue.
    #
    # name - A queue name that responds to to_s.
    def initialize(name = 'queue')
      @name = "turnpike:#{name}"
    end

    # Removes all items from the queue.
    #
    # Returns Integer number of items that were removed.
    def clear
      redis.del name
    end

    # Returns whether the queue is empty.
    def empty?
      size == 0
    end

    # Return an Array of all queued items.
    def peek
      redis.lrange(name, 0, -1).map { |i| Marshal.load i }
    end

    # Pops one or more items from the front of the queue.
    #
    # n - Integer number of items to pop.
    #
    # Return a String item, an Array of items, or nil if the queue is empty.
    def pop(n = 1)
      items = []
      n.times do
        break unless item = redis.lpop(name)
        items << Marshal.load(item)
      end

      n == 1 ? items.first : items
    end

    # Pushes items to the end of the queue.
    #
    # items - A splat Array of items.
    #
    # Returns the Integer size of the queue after the operation.
    def push(*items)
      if redis_version >= '2.4'
        redis.rpush name, items.map { |i| Marshal.dump i }
      else
        items.each { |i| redis.rpush name, Marshal.dump(i) }
      end
    end

    # Syntactic sugar.
    alias << push

    # Returns the Integer size of the queue.
    def size
      redis.llen name
    end

    # Pushes items to the front of the queue.
    #
    # items - A splat Array of items.
    #
    # Returns the Integer size of the queue after the operation.
    def unshift(*items)
      if redis_version >= '2.4'
        redis.lpush name, items.map { |i| Marshal.dump i }
      else
        items.each { |i| redis.lpush name, Marshal.dump(i) }
      end
    end

    private

    def redis
      Redis.current ||= Redis.connect Turnpike.options
    end

    def redis_version
      @redis_version ||= redis.info['redis_version']
    end
  end
end

Version data entries

2 entries across 2 versions & 1 rubygems

Version	Path
turnpike-0.3.2	lib/turnpike/queue.rb
turnpike-0.3.1	lib/turnpike/queue.rb