ActionCable::SubscriptionAdapter::Redis::Listener < SubscriberMap

# File actioncable/lib/action_cable/subscription_adapter/redis.rb, line 65
          def initialize(adapter, config_options, event_loop)
            super()

            @adapter = adapter
            @event_loop = event_loop

            @subscribe_callbacks = Hash.new { |h, k| h[k] = [] }
            @subscription_lock = Mutex.new

            @reconnect_attempt = 0
            # Use the same config as used by Redis conn
            @reconnect_attempts = config_options.fetch(:reconnect_attempts, 1)
            @reconnect_attempts = Array.new(@reconnect_attempts, 0) if @reconnect_attempts.is_a?(Integer)

            @subscribed_client = nil

            @when_connected = []

            @thread = nil
          end

# File actioncable/lib/action_cable/subscription_adapter/redis.rb, line 138
          def add_channel(channel, on_success)
            @subscription_lock.synchronize do
              ensure_listener_running
              @subscribe_callbacks[channel] << on_success
              when_connected { @subscribed_client.subscribe(channel) }
            end
          end

# File actioncable/lib/action_cable/subscription_adapter/redis.rb, line 152
          def invoke_callback(*)
            @event_loop.post { super }
          end

# File actioncable/lib/action_cable/subscription_adapter/redis.rb, line 86
          def listen(conn)
            conn.without_reconnect do
              original_client = extract_subscribed_client(conn)

              conn.subscribe("_action_cable_internal") do |on|
                on.subscribe do |chan, count|
                  @subscription_lock.synchronize do
                    if count == 1
                      @reconnect_attempt = 0
                      @subscribed_client = original_client

                      until @when_connected.empty?
                        @when_connected.shift.call
                      end
                    end

                    if callbacks = @subscribe_callbacks[chan]
                      next_callback = callbacks.shift
                      @event_loop.post(&next_callback) if next_callback
                      @subscribe_callbacks.delete(chan) if callbacks.empty?
                    end
                  end
                end

                on.message do |chan, message|
                  broadcast(chan, message)
                end

                on.unsubscribe do |chan, count|
                  if count == 0
                    @subscription_lock.synchronize do
                      @subscribed_client = nil
                    end
                  end
                end
              end
            end
          end

# File actioncable/lib/action_cable/subscription_adapter/redis.rb, line 146
          def remove_channel(channel)
            @subscription_lock.synchronize do
              when_connected { @subscribed_client.unsubscribe(channel) }
            end
          end

# File actioncable/lib/action_cable/subscription_adapter/redis.rb, line 125
          def shutdown
            @subscription_lock.synchronize do
              return if @thread.nil?

              when_connected do
                @subscribed_client.unsubscribe
                @subscribed_client = nil
              end
            end

            Thread.pass while @thread.alive?
          end