<commit_before>defmodule Red.Connection do
  @moduledoc false

  use Connection

  alias Red.Protocol
  require Logger

  @initial_state %{
    socket: nil,
    tail: "",
    opts: nil,
    queue: :queue.new,
  }

  @socket_opts [:binary, active: false]

  ## Callbacks

  @doc false
  def init(opts) do
    {:connect, :init, Dict.merge(@initial_state, opts: opts)}
  end

  @doc false
  def connect(info, s)

  def connect(_info, %{opts: opts} = s) do
    {host, port, socket_opts} = tcp_connection_opts(opts)

    case :gen_tcp.connect(host, port, socket_opts) do
      {:ok, socket} ->
        setup_socket_buffers(socket)
        auth_and_select_db(%{s | socket: socket})
      {:error, reason} ->
        {:stop, reason, s}
    end
  end

  @doc false
  def disconnect(reason, s)

  def disconnect(:stop, %{socket: nil} = s) do
    {:stop, :normal, s}
  end

  def disconnect(:stop, %{socket: socket} = s) do
    :gen_tcp.close(socket)
    {:stop, :normal, %{s | socket: nil}}
  end

  def disconnect({:error, reason} = error, %{socket: socket, queue: queue} = s) do
    Logger.error "[Red] Disconnected from Redis (#{inspect reason})"

    queue
    |> :queue.to_list
    |> Stream.map(&extract_client_from_queued_item/1)
    |> Enum.map(&Connection.reply(&1, error))

    # Backoff with 0 to churn through all the commands in the mailbox before
    # reconnecting.
    s = %{s | socket: nil, queue: :queue.new, tail: ""}
    {:backoff, 0, s}
  end

  @doc false
  def handle_call(operation, from, s)

  def handle_call(_operation, _from, %{socket: nil} = s) do
    {:reply, {:error, :closed}, s}
  end

  def handle_call({:command, args}, from, %{queue: queue} = s) do
    s
    |> enqueue({:command, from})
    |> send_noreply(Protocol.pack(args))
  end

  def handle_call({:pipeline, commands}, from, %{queue: queue} = s) do
    s
    |> enqueue({:pipeline, from, length(commands)})
    |> send_noreply(Enum.map(commands, &Protocol.pack/1))
  end

  @doc false
  def handle_cast(operation, s)

  def handle_cast(:stop, s) do
    {:disconnect, :stop, s}
  end

  @doc false
  def handle_info(msg, s)

  def handle_info({:tcp, socket, data}, %{socket: socket} = s) do
    reactivate_socket(s)
    s = new_data(s, s.tail <> data)
    {:noreply, s}
  end

  def handle_info({:tcp_closed, socket}, %{socket: socket} = s) do
    {:disconnect, {:error, :tcp_closed}, s}
  end

  def handle_info({:tcp_error, socket, reason}, %{socket: socket} = s) do
    {:disconnect, {:error, reason}, s}
  end

  ## Helper functions<commit_msg>Remove a couple of "unused variable" warnings<commit_after>defmodule Red.Connection do
  @moduledoc false

  use Connection

  alias Red.Protocol
  require Logger

  @initial_state %{
    socket: nil,
    tail: "",
    opts: nil,
    queue: :queue.new,
  }

  @socket_opts [:binary, active: false]

  ## Callbacks

  @doc false
  def init(opts) do
    {:connect, :init, Dict.merge(@initial_state, opts: opts)}
  end

  @doc false
  def connect(info, s)

  def connect(_info, %{opts: opts} = s) do
    {host, port, socket_opts} = tcp_connection_opts(opts)

    case :gen_tcp.connect(host, port, socket_opts) do
      {:ok, socket} ->
        setup_socket_buffers(socket)
        auth_and_select_db(%{s | socket: socket})
      {:error, reason} ->
        {:stop, reason, s}
    end
  end

  @doc false
  def disconnect(reason, s)

  def disconnect(:stop, %{socket: nil} = s) do
    {:stop, :normal, s}
  end

  def disconnect(:stop, %{socket: socket} = s) do
    :gen_tcp.close(socket)
    {:stop, :normal, %{s | socket: nil}}
  end

  def disconnect({:error, reason} = error, %{socket: socket, queue: queue} = s) do
    Logger.error "[Red] Disconnected from Redis (#{inspect reason})"

    queue
    |> :queue.to_list
    |> Stream.map(&extract_client_from_queued_item/1)
    |> Enum.map(&Connection.reply(&1, error))

    # Backoff with 0 to churn through all the commands in the mailbox before
    # reconnecting.
    s = %{s | socket: nil, queue: :queue.new, tail: ""}
    {:backoff, 0, s}
  end

  @doc false
  def handle_call(operation, from, s)

  def handle_call(_operation, _from, %{socket: nil} = s) do
    {:reply, {:error, :closed}, s}
  end

  def handle_call({:command, args}, from, s) do
    s
    |> enqueue({:command, from})
    |> send_noreply(Protocol.pack(args))
  end

  def handle_call({:pipeline, commands}, from, s) do
    s
    |> enqueue({:pipeline, from, length(commands)})
    |> send_noreply(Enum.map(commands, &Protocol.pack/1))
  end

  @doc false
  def handle_cast(operation, s)

  def handle_cast(:stop, s) do
    {:disconnect, :stop, s}
  end

  @doc false
  def handle_info(msg, s)

  def handle_info({:tcp, socket, data}, %{socket: socket} = s) do
    reactivate_socket(s)
    s = new_data(s, s.tail <> data)
    {:noreply, s}
  end

  def handle_info({:tcp_closed, socket}, %{socket: socket} = s) do
    {:disconnect, {:error, :tcp_closed}, s}
  end

  def handle_info({:tcp_error, socket, reason}, %{socket: socket} = s) do
    {:disconnect, {:error, reason}, s}
  end

  ## Helper functions
{'fraction_non_alphanumeric': 0.15059288537549406, 'fraction_numerical': 0.001976284584980237, 'mean_word_length': 3.3072340425531914, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3354981', 'n_tokens_mistral': 1895, 'n_tokens_neox': 1771, 'n_words': 501}