connection is active now, tests added

Author: cybernetlab

lib/thrift/binary/framed/client.ex

@@ -21,7 +21,7 @@ defmodule Thrift.Binary.Framed.Client do
   alias Thrift.TApplicationException
   alias Thrift.Transport.SSL
 
-  @immutable_tcp_opts [active: false, packet: 4, mode: :binary]
+  @immutable_tcp_opts [active: true, packet: 4, mode: :binary]
 
   @type error :: {:error, atom} | {:error, {:exception, struct}}
   @type success :: {:ok, binary}
@@ -168,10 +168,13 @@ defmodule Thrift.Binary.Framed.Client do
   end
 
   @impl Connection
-  def disconnect(info, %{sock: {transport, sock}}) do
+  def disconnect(info, %{sock: {transport, sock}} = s) do
     :ok = transport.close(sock)
 
     case info do
+      {:reconnect, _} = reconnect ->
+        {:connect, info, %{s | sock: nil}}
+
       {:close, from} ->
         Connection.reply(from, :ok)
         {:stop, :normal, nil}
@@ -268,13 +271,13 @@ defmodule Thrift.Binary.Framed.Client do
     timeout = Keyword.get(tcp_opts, :timeout, default_timeout)
 
     with :ok <- transport.send(sock, [message | serialized_args]),
-         {:ok, message} <- transport.recv(sock, 0, timeout) do
+         {:ok, message} <- receive_message(sock, timeout) do
       reply = deserialize_message_reply(message, rpc_name, seq_id)
       {:reply, reply, s}
     else
       {:error, :closed} = error ->
         if reconnect do
-          {:connect, {:reconnect, :call, msg, from}, %{s | sock: nil}}
+          {:disconnect, {:reconnect, {:call, msg, from}}, s}
         else
           {:disconnect, error, error, s}
         end
@@ -310,7 +313,7 @@ defmodule Thrift.Binary.Framed.Client do
 
       {:error, :closed} = error ->
         if reconnect do
-          {:connect, {:reconnect, :cast, msg}, %{s | sock: nil}}
+          {:disconnect, {:reconnect, {:cast, msg}}, s}
         else
           {:disconnect, error, s}
         end
@@ -320,10 +323,32 @@ defmodule Thrift.Binary.Framed.Client do
     end
   end
 
+  @impl Connection
+  def handle_info({:tcp_closed, sock}, %{reconnect: true, sock: {_transport, sock}} = s) do
+    {:disconnect, {:reconnect, nil}, s}
+  end
+
+  def handle_info(_, s) do
+    {:noreply, s}
+  end
+
   def deserialize_message_reply(message, rpc_name, seq_id) do
     handle_message(Binary.deserialize(:message_begin, message), seq_id, rpc_name)
   end
 
+  defp receive_message(sock, timeout) do
+    receive do
+      {:tcp, ^sock, data} -> {:ok, data}
+      {:tcp_closed, ^sock} -> {:error, :closed}
+      {:tcp_error, ^sock, error} -> {:error, error}
+      {:ssl, ^sock, data} -> {:ok, data}
+      {:ssl_closed, ^sock} -> {:error, :closed}
+      {:ssl_error, ^sock, error} -> {:error, error}
+    after
+      timeout -> {:error, :timeout}
+    end
+  end
+
   defp handle_message({:ok, {:reply, seq_id, rpc_name, serialized_response}}, seq_id, rpc_name) do
     {:ok, serialized_response}
   end
@@ -402,7 +427,7 @@ defmodule Thrift.Binary.Framed.Client do
     end
   end
 
-  defp maybe_resend_data({:ok, s}, {:reconnect, :call, msg, from}) do
+  defp maybe_resend_data({:ok, s}, {:reconnect, {:call, msg, from}}) do
     case handle_call(msg, from, s) do
       {:reply, reply, s} ->
         GenServer.reply(from, reply)
@@ -417,7 +442,7 @@ defmodule Thrift.Binary.Framed.Client do
     end
   end
 
-  defp maybe_resend_data({:ok, s}, {:reconnect, :cast, msg}) do
+  defp maybe_resend_data({:ok, s}, {:reconnect, {:cast, msg}}) do
     case handle_cast(msg, s) do
       {:noreply, s} ->
         {:ok, s}

mix.lock

@@ -14,7 +14,7 @@
   "makeup_elixir": {:hex, :makeup_elixir, "0.14.0", "cf8b7c66ad1cff4c14679698d532f0b5d45a3968ffbcbfd590339cb57742f1ae", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm"},
   "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm"},
   "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm"},
-  "nimble_parsec": {:hex, :nimble_parsec, "0.5.1", "c90796ecee0289dbb5ad16d3ad06f957b0cd1199769641c961cfe0b97db190e0", [], [], "hexpm"},
+  "nimble_parsec": {:hex, :nimble_parsec, "0.5.1", "c90796ecee0289dbb5ad16d3ad06f957b0cd1199769641c961cfe0b97db190e0", [:mix], [], "hexpm"},
   "parse_trans": {:hex, :parse_trans, "3.3.0", "09765507a3c7590a784615cfd421d101aec25098d50b89d7aa1d66646bc571c1", [:rebar3], [], "hexpm"},
   "ranch": {:hex, :ranch, "1.7.1", "6b1fab51b49196860b733a49c07604465a47bdb78aa10c1c16a3d199f7f8c881", [:rebar3], [], "hexpm"},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.4", "f0eafff810d2041e93f915ef59899c923f4568f4585904d010387ed74988e77b", [:make, :mix, :rebar3], [], "hexpm"},

test/thrift/binary/framed/server_test.exs

@@ -177,4 +177,24 @@ defmodule Servers.Binary.Framed.IntegrationTest do
   thrift_test "client methods can be called by name instead of pid", %{client_name: name} do
     assert {:ok, true} == Client.ping(name)
   end
+
+  @ping_reply <<128, 1, 0, 2, 0, 0, 0, 4, 112, 105, 110, 103, 0, 0, 0, 0, 2, 0, 0, 1, 0>>
+  thrift_test "client can reconnect when connection closed by server", ctx do
+    {:ok, sock} = :gen_tcp.listen(0, [:binary, packet: 4, active: false])
+    {:ok, port} = :inet.port(sock)
+    first_conn = Task.async(fn ->
+      {:ok, conn} = :gen_tcp.accept(sock)
+      :ok = :gen_tcp.close(conn)
+    end)
+    name = String.to_atom("#{ctx.client_name}_1")
+    {:ok, client} = Client.start_link("localhost", port, name: name, reconnect: true)
+    second_conn = Task.async(fn ->
+      {:ok, conn} = :gen_tcp.accept(sock)
+      {:ok, _} = :gen_tcp.recv(conn, 0)
+      :ok = :gen_tcp.send(conn, @ping_reply)
+    end)
+    assert {:ok, true} == Client.ping(client)
+    Task.await(first_conn)
+    Task.await(second_conn)
+  end
 end