diff --git a/lib/postgrex/protocol.ex b/lib/postgrex/protocol.ex index 10cc76cd..64c0ce4e 100644 --- a/lib/postgrex/protocol.ex +++ b/lib/postgrex/protocol.ex @@ -1363,6 +1363,11 @@ defmodule Postgrex.Protocol do err = Postgrex.Error.exception(postgres: fields) error_ready(s, status, err, buffer) + {:ok, msg, buffer} -> + status = new_status([], mode: :transaction) + s = handle_msg(s, status, msg) + recv_streaming(s, buffer) + {:disconnect, _, _} = dis -> dis end diff --git a/test/protocol_test.exs b/test/protocol_test.exs new file mode 100644 index 00000000..d6ba8c23 --- /dev/null +++ b/test/protocol_test.exs @@ -0,0 +1,40 @@ +defmodule Postgrex.ProtocolTest do + use ExUnit.Case, async: true + + alias Postgrex.Protocol + + defmodule Socket do + def send(pid, data) do + Kernel.send(pid, {:sent, IO.iodata_to_binary(data)}) + :ok + end + end + + test "streaming startup handles asynchronous messages before the copy response" do + responses = + IO.iodata_to_binary([ + backend_message(?S, ["application_name", 0, "postgrex", 0]), + backend_message(?N, [?S, "NOTICE", 0, ?M, "replication starting", 0, 0]), + backend_message(?A, [<<123::32>>, "events", 0, "ready", 0]), + backend_message(?W, <<0, 0::16>>) + ]) + + state = %Protocol{ + sock: {Socket, self()}, + buffer: responses, + parameters: %{}, + messages: [] + } + + assert {:ok, state} = Protocol.handle_streaming("START_REPLICATION", state) + assert state.parameters["application_name"] == "postgrex" + assert [%{message: "replication starting", severity: "NOTICE"}] = state.messages + assert state.buffer == "" + + assert_receive {:sent, <>} + end + + defp backend_message(type, data) do + [type, <>, data] + end +end