floof/lib/floof.ex

defmodule Floof do
  @moduledoc """
  Documentation for `Floof`.
  """

  require Logger

  def accept(port) do
    {:ok, socket} = :gen_tcp.listen(port, [
      :binary,
      packet: 4,
      active: false,
      reuseaddr: true
    ])
    Logger.info("Accepting connections on #{port}")
    loop_acceptor(socket)
  end

  defp loop_acceptor(socket) do
    {:ok, client} = :gen_tcp.accept(socket)
    {:ok, pid} = Task.Supervisor.start_child(
      Floof.TaskSupervisor, fn -> serve(client, %{}) end)
    :ok = :gen_tcp.controlling_process(client, pid)
    # switch to active now that we handed over the client
    :inet.setopts(client, [{:active, true}])
    send(Floof.Distributor, {:register, pid})
    loop_acceptor(socket)
  end

  defp serve(client, backlog) do
    receive do
      {:tcp, _, message} ->
        {:ok, dcd} = :FloofProtocol.decode(:ProtoMessage, message)
        backlog = handle_incoming(client, backlog, dcd)
        serve(client, backlog)
      {:fwdxfer, {dcdid, dcd}} ->
        handle_outgoing(client, dcdid)
        serve(client, Map.put(backlog, dcdid, dcd))
      {:havewe, id, res} ->
        dat = {:Summary, if res do :rejectpush else :pull end, id}
        {:ok, encd} = :FloofProtocol.encode(:ProtoMessage, {:summary, dat})
        :ok = client.send(encd)
        serve(client, backlog)
      {:SessionPushed, key} ->
        {backlog, added_ids} = extract_backlog(key, backlog, MapSet.new())
        for id <- added_ids do
          dat = {:Summary, :requestpull, id}
          {:ok, encd} = :FloofProtocol.encode(:ProtoMessage, {:summary, dat})
          :ok = client.send(encd)
        end
        serve(client, backlog)
      x -> Logger.warn("unable to handle request: #{inspect(x)}")
        serve(client, backlog)
    end
  end

  defp handle_incoming(client, backlog, dcd) do
    Logger.debug("got packet #{inspect(dcd)}")
    case dcd do
      {:session, {:SessionModify, {:attach, key}}} ->
        if Floof.SessionManager.attach(key, self()) do
          send self(), {:SessionPushed, key}
        end
        backlog
      {:summary, {:Summary, direction, id}} ->
        case direction do
          :requestpull ->
            send Floof.Distributor, {:havewe, self(), id}
            backlog
          :pull ->
            {item, backlog2} = Map.pop(backlog, id)
            if item != nil do
              {:ok, encd} = :FloofProtocol.encode(:ProtoMessage, {:xfer, {id, item}})
              :ok = client.send(encd)
            end
            backlog2
         :rejectpush ->
            Map.delete(backlog, id)
        end
      {:xfer, xf} ->
        {:XferBlob, _, _, _, _, _, xfinner} = xf
        {:ok, {:XferInner, id, _, _, _, _}} = :FloofProtocol.decode(:XferInner, xfinner)
        send Floof.Distributor, {:xfer, {id, xf}}
        backlog
    end
  end

  defp handle_outgoing(client, dcdid) do
    dat = {:Summary, :requestpull, dcdid}
    {:ok, encd} = :FloofProtocol.encode(:ProtoMessage, {:summary, dat})
    :ok = client.send(encd)
  end

  defp extract_backlog(key, backlog, added_ids) do
    case Floof.SessionManager.pop(key) do
      :empty -> {backlog, added_ids}
      {:value, {id, entry}} ->
        extract_backlog(key, Map.put(backlog, id, entry), MapSet.put(added_ids, id))
    end
  end
end