defmodule Memcache.Receiver do @moduledoc false use GenServer alias Memcache.Protocol defmodule State do @moduledoc false defstruct sock: nil, parent: nil, buffer: <<>> end def start_link(args) do GenServer.start_link(__MODULE__, args) end def stop(receiver) do GenServer.stop(receiver) end def read(receiver, client, command, opts) do GenServer.cast(receiver, {:read, client, command, opts}) end def read_quiet(receiver, client, commands) do GenServer.cast(receiver, {:read_quiet, client, commands}) end def init([sock, parent]) do {:ok, %State{sock: sock, parent: parent}} end def handle_cast({:read, client, command, opts}, %State{sock: sock, buffer: buffer} = s) do reply_or_disconnect(recv_response(command, sock, buffer, opts), client, s) end def handle_cast({:read_quiet, client, commands}, %State{sock: sock, buffer: buffer} = s) do reply_or_disconnect(recv_response_quiet(commands, sock, [], buffer), client, s) end defp reply_or_disconnect({:ok, response, buffer}, client, s) do Connection.reply(client, response) send(s.parent, {:receiver, :done, client, self()}) {:noreply, %{s | buffer: buffer}} end defp reply_or_disconnect(error, _client, s) do send(s.parent, {:receiver, :disconnect, error, self()}) {:stop, :normal, s} end def recv_response(:STAT, sock, buffer, _opts) do recv_stat(sock, buffer, Map.new) end def recv_response(_command, sock, buffer, %{cas: cas}) do with {:ok, header, buffer} <- recv_header(sock, buffer), {:ok, response, buffer} <- recv_body(header, sock, buffer) do if cas do { :ok, append_cas_version(response, header), buffer } else { :ok, response, buffer } end end end defp recv_header_and_body(sock, buffer) do with {:ok, header, buffer} <- recv_header(sock, buffer) do recv_body(header, sock, buffer) end end defp recv_header(sock, buffer) do with {:ok, raw_header, buffer} <- read(sock, buffer, 24) do { :ok, Protocol.parse_header(raw_header), buffer } end end defp recv_body(header, sock, buffer) do body_size = Protocol.total_body_size(header) if body_size > 0 do with {:ok, body, buffer} <- read(sock, buffer, body_size) do response = Protocol.parse_body(header, body) |> elem(1) { :ok, response, buffer} end else response = Protocol.parse_body(header, :empty) |> elem(1) { :ok, response, buffer} end end defp recv_stat(sock, buffer, results) do case recv_header_and_body(sock, buffer) do { :ok, { :ok, :done }, buffer } -> { :ok, { :ok, results }, buffer } { :ok, { :ok, key, val }, buffer } -> recv_stat(sock, buffer, Map.put(results, key, val)) err -> err end end defp append_cas_version({:ok}, %{cas: cas_version}), do: {:ok, cas_version} defp append_cas_version({:ok, value}, %{cas: cas_version}), do: {:ok, value, cas_version} defp append_cas_version(error, %{cas: _cas_version}), do: error defp recv_response_quiet([], _sock, results, buffer) do { :ok, { :ok, Enum.reverse(tl(results)) }, buffer } end defp recv_response_quiet(commands, sock, results, buffer) do with {:ok, header_raw, rest} <- read(sock, buffer, 24) do header = Protocol.parse_header(header_raw) body_size = Protocol.total_body_size(header) if body_size > 0 do case read(sock, rest, body_size) do { :ok, body, rest } -> { rest_commands, results } = match_response(commands, results, header, Protocol.parse_body(header, body)) recv_response_quiet(rest_commands, sock, results, rest) err -> err end else { rest_commands, results } = match_response(commands, results, header, Protocol.parse_body(header, :empty)) recv_response_quiet(rest_commands, sock, results, rest) end end end defp match_response([ { i, _command, _args, %{cas: true} } | rest ], results, header, { i, response }) do { rest, [append_cas_version(response, header) | results] } end defp match_response([ { i, _command, _args, _opts } | rest ], results, _header, { i, response }) do { rest, [response | results] } end defp match_response([ { _i , command, _args, _opts } | rest ], results, header, response_with_index) do match_response(rest, [Protocol.quiet_response(command) | results], header, response_with_index) end defp read(_sock, buffer, min_required) when byte_size(buffer) >= min_required do { requested, rest } = cut(buffer, min_required) { :ok, requested, rest} end defp read(sock, buffer, min_required) do case :gen_tcp.recv(sock, 0) do { :ok, data } -> read(sock, buffer <> data, min_required) { :error, reason } -> { :error, reason } end end defp cut(bin, at) do first = binary_part(bin, 0, at) rest = binary_part(bin, at, byte_size(bin) - at) { first, rest } end end