Class: ActionCable::SubscriptionAdapter::Redis::Listener

# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 206

def retry_connecting?
  @reconnect_attempt += 1

  return false if @reconnect_attempt > @reconnect_attempts.size

  sleep_t = @reconnect_attempts[@reconnect_attempt - 1]

  sleep(sleep_t) if sleep_t > 0

  true
end

# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 167

def ensure_listener_running
  # The internal sentinel subscription keeps the listener alive in
  # normal operation, but the thread can still die for other reasons
  # (e.g. exhausting the Redis reconnect attempts). Since this method
  # memoizes with `||=`, a dead thread would never be replaced and
  # every later subscribe would queue forever. Drop a dead thread so
  # the next subscribe spawns a fresh listener.
  if @thread && !@thread.alive?
    @thread = nil
    @reconnect_attempt = 0
  end

  @thread ||= Thread.new do
    Thread.current.abort_on_exception = true

    begin
      conn = @adapter.redis_connection_for_subscriptions
      listen conn
    rescue RedisClient::ConnectionError => e
      reset
      if retry_connecting?
        logger&.warn "Redis connection failed: #{e.message}. Trying to reconnect..."
        when_connected { resubscribe }
        retry
      else
        logger&.error "Failed to reconnect to Redis after #{@reconnect_attempt} attempts."
      end
    end
  end
end

# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 105

def listen(conn)
  pubsub_client = conn.pubsub

  @reconnect_attempt = 0
  @subscribed_client = pubsub_client

  pubsub_client.call("subscribe", INTERNAL_CHANNEL)

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

  loop do
    type, chan, message = pubsub_client.next_event(60)
    case type
    when "subscribe", "psubscribe"
      if callbacks = @subscribe_callbacks[chan]
        next_callback = callbacks.shift
        @executor.post(&next_callback) if next_callback
        @subscribe_callbacks.delete(chan) if callbacks.empty?
      end
    when "message", "pmessage"
      broadcast(chan, message)
    when "unsubscribe", "punsubscribe"
      if message == 0
        @subscription_lock.synchronize do
          @subscribed_client = nil
        end
        break
      end
    end
  end
end

# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 160

def remove_channel(channel)
  @subscription_lock.synchronize do
    when_connected { @subscribed_client.call("unsubscribe", channel) }
  end
end

# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 225

def reset
  @subscription_lock.synchronize do
    @subscribed_client = nil
    @when_connected.clear
  end
end

# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 218

def resubscribe
  channels = @sync.synchronize do
    @subscribers.keys
  end
  @subscribed_client.call("subscribe", *channels) unless channels.empty?
end

# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 139

def shutdown
  @subscription_lock.synchronize do
    return if @thread.nil?

    when_connected do
      @subscribed_client.call("unsubscribe")
      @subscribed_client = nil
    end
  end

  Thread.pass while @thread.alive?
end

# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 198

def when_connected(&block)
  if @subscribed_client
    block.call
  else
    @when_connected << block
  end
end