Support Latest OTP

lib/statix.ex

@@ -359,11 +359,10 @@ defmodule Statix do
   @doc false
   def new(module, options) do
     config = get_config(module, options)
-    conn = Conn.new(config.host, config.port)
-    header = IO.iodata_to_binary([conn.header | config.prefix])
+    conn = Conn.new(config.host, config.port, config.prefix)
 
     %__MODULE__{
-      conn: %{conn | header: header},
+      conn: conn,
       pool: build_pool(module, config.pool_size),
       tags: config.tags
     }

lib/statix/conn.ex

@@ -1,21 +1,20 @@
 defmodule Statix.Conn do
   @moduledoc false
 
-  defstruct [:sock, :header]
+  defstruct [:sock, :address, :port, :prefix]
 
   alias Statix.Packet
 
   require Logger
 
-  def new(host, port) when is_binary(host) do
-    new(String.to_charlist(host), port)
+  def new(host, port, prefix) when is_binary(host) do
+    new(String.to_charlist(host), port, prefix)
   end
 
-  def new(host, port) when is_list(host) or is_tuple(host) do
+  def new(host, port, prefix) when is_list(host) or is_tuple(host) do
     case :inet.getaddr(host, :inet) do
       {:ok, address} ->
-        header = Packet.header(address, port)
-        %__MODULE__{header: header}
+        %__MODULE__{address: address, port: port, prefix: prefix}
 
       {:error, reason} ->
         raise(
@@ -30,34 +29,31 @@ defmodule Statix.Conn do
     %__MODULE__{conn | sock: sock}
   end
 
-  def transmit(%__MODULE__{header: header, sock: sock}, type, key, val, options)
+  def transmit(%__MODULE__{sock: sock, prefix: prefix} = conn, type, key, val, options)
       when is_binary(val) and is_list(options) do
     result =
-      header
+      prefix
       |> Packet.build(type, key, val, options)
-      |> transmit(sock)
+      |> transmit(conn)
 
-    if result == {:error, :port_closed} do
+    with {:error, error} <- result do
       Logger.error(fn ->
         if(is_atom(sock), do: "", else: "Statix ") <>
-          "#{inspect(sock)} #{type} metric \"#{key}\" lost value #{val}" <> " due to port closure"
+          "#{inspect(sock)} #{type} metric \"#{key}\" lost value #{val}" <>
+          " error=#{inspect(error)}"
       end)
     end
 
     result
   end
 
-  defp transmit(packet, sock) do
-    try do
-      Port.command(sock, packet)
-    rescue
-      ArgumentError ->
-        {:error, :port_closed}
+  defp transmit(packet, %__MODULE__{address: address, port: port, sock: sock_name}) do
+    sock = Process.whereis(sock_name)
+
+    if sock do
+      :gen_udp.send(sock, address, port, packet)
     else
-      true ->
-        receive do
-          {:inet_reply, _port, status} -> status
-        end
+      {:error, :port_closed}
     end
   end
 end

lib/statix/packet.ex

@@ -1,34 +1,8 @@
 defmodule Statix.Packet do
   @moduledoc false
 
-  use Bitwise
-
-  otp_release = :erlang.system_info(:otp_release)
-  @addr_family if(otp_release >= '19', do: [1], else: [])
-
-  def header({n1, n2, n3, n4}, port) do
-    true = Code.ensure_loaded?(:gen_udp)
-
-    anc_data_part =
-      if function_exported?(:gen_udp, :send, 5) do
-        [0, 0, 0, 0]
-      else
-        []
-      end
-
-    @addr_family ++
-      [
-        band(bsr(port, 8), 0xFF),
-        band(port, 0xFF),
-        band(n1, 0xFF),
-        band(n2, 0xFF),
-        band(n3, 0xFF),
-        band(n4, 0xFF)
-      ] ++ anc_data_part
-  end
-
-  def build(header, name, key, val, options) do
-    [header, key, ?:, val, ?|, metric_type(name)]
+  def build(prefix, name, key, val, options) do
+    [prefix, key, ?:, val, ?|, metric_type(name)]
     |> set_option(:sample_rate, options[:sample_rate])
     |> set_option(:tags, options[:tags])
   end

test/statix_test.exs

@@ -9,7 +9,11 @@ defmodule StatixTest do
 
   defp close_port() do
     %{pool: pool} = current_statix()
-    Enum.each(pool, &Port.close/1)
+
+    Enum.each(pool, fn module_name ->
+      sock = Process.whereis(module_name)
+      :gen_udp.close(sock)
+    end)
   end
 
   setup do
@@ -156,22 +160,22 @@ defmodule StatixTest do
 
     assert capture_log(fn ->
              assert {:error, :port_closed} == increment("sample")
-           end) =~ "counter metric \"sample\" lost value 1 due to port closure"
+           end) =~ "counter metric \"sample\" lost value 1 error=:port_closed\n\e[0m"
 
     assert capture_log(fn ->
              assert {:error, :port_closed} == decrement("sample")
-           end) =~ "counter metric \"sample\" lost value -1 due to port closure"
+           end) =~ "counter metric \"sample\" lost value -1 error=:port_closed\n\e[0m"
 
     assert capture_log(fn ->
              assert {:error, :port_closed} == gauge("sample", 2)
-           end) =~ "gauge metric \"sample\" lost value 2 due to port closure"
+           end) =~ "gauge metric \"sample\" lost value 2 error=:port_closed\n\e[0m"
 
     assert capture_log(fn ->
              assert {:error, :port_closed} == histogram("sample", 3)
-           end) =~ "histogram metric \"sample\" lost value 3 due to port closure"
+           end) =~ "histogram metric \"sample\" lost value 3 error=:port_closed\n\e[0m"
 
     assert capture_log(fn ->
              assert {:error, :port_closed} == timing("sample", 2.5)
-           end) =~ "timing metric \"sample\" lost value 2.5 due to port closure"
+           end) =~ "timing metric \"sample\" lost value 2.5 error=:port_closed\n\e[0m"
   end
 end

test/support/test_server.exs

@@ -8,6 +8,7 @@ defmodule Statix.TestServer do
   @impl true
   def init(port) do
     {:ok, socket} = :gen_udp.open(port, [:binary, active: true])
+    Process.flag(:trap_exit, true)
     {:ok, %{socket: socket, test: nil}}
   end
 
@@ -20,13 +21,23 @@ defmodule Statix.TestServer do
     end
   end
 
+  @impl true
+  def handle_info({:EXIT, _pid, reason}, state) do
+    {:stop, reason, state}
+  end
+
   @impl true
   def handle_info({:udp, socket, host, port, packet}, %{socket: socket, test: test} = state) do
     metadata = %{host: host, port: port, socket: socket}
     send(test, {:test_server, metadata, packet})
     {:noreply, state}
   end
 
+  @impl true
+  def terminate(_reason, %{socket: socket}) do
+    :gen_udp.close(socket)
+  end
+
   def setup(test_module) do
     :ok = set_current_test(test_module, self())
     ExUnit.Callbacks.on_exit(fn -> set_current_test(test_module, nil) end)