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 def connect(host, port) do {:ok, socket} = :gen_tcp.connect(host, port, [ packet: 4, active: true ]) send(Floof.Distributor, {:register, self()}) Logger.info("Established connection to #{host}:#{port}") serve(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}} -> send_summary(client, :requestpull, dcdid) serve(client, Map.put(backlog, dcdid, dcd)) {:havewe, dcdid, res} -> send_summary(client, if res do :rejectpush else :pull end, dcdid) serve(client, backlog) {:SessionPushed, key} -> {backlog, added_ids} = extract_backlog(key, backlog, MapSet.new()) for id <- added_ids, do: send_summary(client, :requestpull, id) 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 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 defp send_summary(client, direction, dcdid) do dat = {:Summary, direction, dcdid} {:ok, encd} = :FloofProtocol.encode(:ProtoMessage, {:summary, dat}) :ok = client.send(encd) end end