barytherium

defmodule Barytherium.DefaultCallbackHandler do
  @moduledoc """
  This module serves as a more-or-less self-contained example of how to use Barytherium in
  your own project.

  ```
  (1)> {:ok, bdcbh_pid} = Barytherium.DefaultCallbackHandler.start_link('example.com', 61613, "/", [{"login", "grimAuxiliatrix"}, {"passcode", "hunter2"}, {"heart-beat", "5000,5000"}], [%Barytherium.Frame{command: :subscribe, headers: [{"id", "0"}, {"destination", "/amq/queue/darwin-v16-gzip"}, {"ack", "client"}]}])
  {:ok, #PID<0.270.0>}
  (2)>
  20:44:51.229 [info] Connection to example.com:61613 succeeded, remote end has picked up

  20:44:51.406 [info] Received connected frame: %Barytherium.Frame{command: :connected, headers: [{"server", "RabbitMQ/3.10.5"}, {"session", "session-1uw6oN9eUYg8MyRacmDcnw"}, {"heart-beat", "5000,5000"}, {"version", "1.2"}], body: ""}

  20:44:51.710 [info] Unpacked final frame in list: %Barytherium.Frame{command: :message,...
  ```
  """
  use GenServer
  require Logger
  alias Barytherium.Frame
  alias Barytherium.Network
  alias Barytherium.Network.Sender

  def init(state = %{host: host, port: port, virtual_host: _virtual_host, opts: opts}) do
    network_pid =
      Network.start_link(
        self(),
        host,
        port,
        opts ++
          [
            # These are optional (and close to defaults, omitting them wouldn't have much effect)
            send_timeout: 10_000,
            buffer: 0x400_000,
            receive_buffer: 0x400_000,
            send_buffer: 0x400_000,
            receiver_options: [
              first_frame_timeout: 5_000,
              heartbeat_receive_allowance: 1_000
            ]
          ]
      )

    {:ok, Map.put(state, :network_pid, network_pid)}
  end

  def start_link(
        host,
        port,
        virtual_host,
        connect_headers \\ [],
        connected_frames \\ [],
        opts \\ [],
        link_opts \\ []
      ) do
    GenServer.start_link(
      __MODULE__,
      %{
        opts: opts,
        host: host,
        port: port,
        virtual_host: virtual_host,
        connect_headers: connect_headers,
        connected_frames: connected_frames
      },
      link_opts
    )
  end

  def handle_cast({:barytherium, :connect, {:error, error}}, state = %{host: host, port: port}) do
    Logger.error("Connection to #{host}:#{port} failed, error: #{inspect(error)}")
    {:stop, :connect_failed, state}
  end

  def handle_cast(
        {:barytherium, :connect, {:ok, sender_pid}},
        state = %{
          host: host,
          port: port,
          virtual_host: virtual_host,
          connect_headers: connect_headers
        }
      ) do
    Logger.info("Connection to #{host}:#{port} succeeded, remote end has picked up")

    Sender.write(sender_pid, [
      %Frame{
        command: :connect,
        headers: connect_headers ++ [{"accept-version", "1.2"}, {"host", virtual_host}]
      }
    ])

    {:noreply, state}
  end

  def handle_cast(
        {:barytherium, :frames, {[frame = %Frame{command: :connected}], sender_pid}},
        state = %{connected_frames: connected_frames}
      ) do
    Logger.info("Received connected frame: " <> inspect(frame, binaries: :as_strings))
    Sender.write(sender_pid, connected_frames)
    {:noreply, state}
  end

  def handle_cast({:barytherium, :disconnect, reason}, state = %{host: host, port: port}) do
    Logger.error("Connection to #{host}:#{port} disconnected for reason #{reason}")
    {:stop, :connect_disconnected, state}
  end

  def handle_cast({:barytherium, :frames, {frames, sender_pid}}, state) do
    # Logger.info("Received frames: " <> inspect(frames, binaries: :as_strings))

    handle_frame(frames, sender_pid)
    {:noreply, state}
  end

  def handle_frame([frame | []], sender_pid) do
    Logger.info("Unpacked final frame in list: " <> inspect(frame, binaries: :as_strings))
    headers = Frame.headers_to_map(frame.headers)

    Sender.write(
      sender_pid,
      [%Frame{command: :ack, headers: [{"id", Map.get(headers, "ack")}]}]
    )
  end

  def handle_frame([frame | rest], sender_pid) do
    Logger.info("Unpacked frame: " <> inspect(frame, binaries: :as_strings))
    handle_frame(rest, sender_pid)
  end
end