Skip to content
This repository

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Fetching contributors…

Octocat-spinner-32-eaf2f5

Cannot retrieve contributors at this time

file 95 lines (75 sloc) 2.007 kb
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94
class Redis
  class SubscribedClient
    def initialize(client)
      @client = client
    end

    # Starting with 2.2.1, assume that this method is called with a single
    # array argument. Check its size for backwards compat.
    def call(*args)
      if args.first.is_a?(Array) && args.size == 1
        command = args.first
      else
        command = args
      end

      @client.process([command])
    end

    # Assume that this method is called with a single array argument. No
    # backwards compat here, since it was introduced in 2.2.2.
    def call_without_reply(command)
      @commands.push command
      nil
    end

    def subscribe(*channels, &block)
      subscription("subscribe", "unsubscribe", channels, block)
    end

    def psubscribe(*channels, &block)
      subscription("psubscribe", "punsubscribe", channels, block)
    end

    def unsubscribe(*channels)
      call [:unsubscribe, *channels]
    end

    def punsubscribe(*channels)
      call [:punsubscribe, *channels]
    end

  protected

    def subscription(start, stop, channels, block)
      sub = Subscription.new(&block)

      begin
        @client.call_loop([start, *channels]) do |line|
          type, *rest = line
          sub.callbacks[type].call(*rest)
          break if type == stop && rest.last == 0
        end
      ensure
        send(stop)
      end
    end
  end

  class Subscription
    attr :callbacks

    def initialize
      @callbacks = Hash.new do |hash, key|
        hash[key] = lambda { |*_| }
      end

      yield(self)
    end

    def subscribe(&block)
      @callbacks["subscribe"] = block
    end

    def unsubscribe(&block)
      @callbacks["unsubscribe"] = block
    end

    def message(&block)
      @callbacks["message"] = block
    end

    def psubscribe(&block)
      @callbacks["psubscribe"] = block
    end

    def punsubscribe(&block)
      @callbacks["punsubscribe"] = block
    end

    def pmessage(&block)
      @callbacks["pmessage"] = block
    end
  end
end
Something went wrong with that request. Please try again.