Contents

require "metacrunch/redis"

module Metacrunch
  class Redis::QueueSource

    DEFAULT_OPTIONS = {
      blocking_mode: false
    }

    def initialize(redis, queue_name, options = {})
      @redis = redis
      @queue_name = queue_name
      @options = DEFAULT_OPTIONS.merge(options)
    end

    def each(&block)
      return enum_for(__method__) unless block_given?

      if @options[:blocking_mode]
        while true
          list, result = @redis.blpop(@queue_name, timeout: 0)
          if result.present?
            yield result
          else
            yield nil
          end
        end
      else
        while result = @redis.lpop(@queue_name)
          yield result
        end
      end

      self
    end

  end
end

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
metacrunch-redis-1.1.0	lib/metacrunch/redis/queue_source.rb