Phase 1: read a whole message off a socket

lib/mysqlex/client.ex

@@ -0,0 +1,30 @@
+defmodule Mysqlex.Client do
+  @moduledoc """
+  Reads and writes packets over a `Mysqlex.Transport`.
+
+  This is where the socket lives. Everything it hands upwards is a whole
+  message, so no parser above it ever sees a partial read.
+  """
+
+  alias Mysqlex.Packet
+
+  @doc """
+  Reads one message, reading again for as long as the buffer holds less than a
+  whole message.
+
+  `buffer` carries over whatever a previous read left unconsumed, and the
+  returned `rest` is what this read left behind.
+  """
+  def recv_message(transport, socket, buffer, timeout) do
+    case Packet.decode_message(buffer) do
+      {:ok, sequence_id, message, rest} ->
+        {:ok, sequence_id, message, rest}
+
+      :more ->
+        case transport.recv(socket, 0, timeout) do
+          {:ok, data} -> recv_message(transport, socket, buffer <> data, timeout)
+          {:error, reason} -> {:error, reason}
+        end
+    end
+  end
+end

lib/mysqlex/tcp.ex

@@ -3,12 +3,15 @@ defmodule Mysqlex.Tcp do
   Thin wrapper around `:gen_tcp` for the connection's socket.
   """
 
+  @behaviour Mysqlex.Transport
+
   @socket_opts [packet: :raw, mode: :binary, active: false]
 
   def connect(host, port, opts \\ [], timeout \\ 5_000) do
     :gen_tcp.connect(host, port, @socket_opts ++ opts, timeout)
   end
 
+  @impl true
   def recv(socket, bytes, timeout \\ :infinity) do
     :gen_tcp.recv(socket, bytes, timeout)
   end

lib/mysqlex/transport.ex

@@ -0,0 +1,15 @@
+defmodule Mysqlex.Transport do
+  @moduledoc """
+  The socket operations the client needs.
+
+  Reading is expressed against this behaviour rather than `:gen_tcp` directly so
+  that framing and, later, the connection phase can be driven by a scripted
+  transport in tests. `Mysqlex.Tcp` is the real implementation.
+
+  Callbacks are added as the client comes to need them.
+  """
+
+  @type socket :: term
+
+  @callback recv(socket, bytes :: non_neg_integer, timeout) :: {:ok, binary} | {:error, term}
+end

mix.exs

@@ -9,6 +9,7 @@ defmodule Mysqlex.MixProject do
       app: :mysqlex,
       version: @version,
       elixir: "~> 1.17",
+      elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,
       deps: deps(),
       name: "Mysqlex",
@@ -18,6 +19,9 @@ defmodule Mysqlex.MixProject do
     ]
   end
 
+  defp elixirc_paths(:test), do: ["lib", "test/support"]
+  defp elixirc_paths(_), do: ["lib"]
+
   # Run "mix help compile.app" to learn about applications.
   def application do
     [

test/mysqlex/client_test.exs

@@ -0,0 +1,18 @@
+defmodule Mysqlex.ClientTest do
+  use ExUnit.Case, async: true
+
+  alias Mysqlex.{Client, FakeTransport}
+
+  test "reads a message delivered across several reads" do
+    socket = FakeTransport.script([<<5::little-24>>, <<0, "hel">>, "lo"])
+
+    assert {:ok, 0, "hello", ""} =
+             Client.recv_message(FakeTransport, socket, "", 1_000)
+  end
+
+  test "propagates a transport error when the message never completes" do
+    socket = FakeTransport.script([<<5::little-24>>, <<0, "hel">>])
+
+    assert {:error, :closed} = Client.recv_message(FakeTransport, socket, "", 1_000)
+  end
+end

test/support/fake_transport.ex

@@ -0,0 +1,24 @@
+defmodule Mysqlex.FakeTransport do
+  @moduledoc """
+  A transport that replays a scripted list of byte chunks, so framing and the
+  connection phase can be driven without a server.
+
+  Each `recv/3` returns the next chunk, which lets a test deliver one packet
+  across several reads, or several packets in one read.
+  """
+
+  @behaviour Mysqlex.Transport
+
+  def script(chunks) do
+    {:ok, agent} = Agent.start_link(fn -> chunks end)
+    agent
+  end
+
+  @impl true
+  def recv(agent, _bytes, _timeout) do
+    Agent.get_and_update(agent, fn
+      [chunk | rest] -> {{:ok, chunk}, rest}
+      [] -> {{:error, :closed}, []}
+    end)
+  end
+end