defmodule Object.NetworkTransport do @moduledoc """ Network transport layer abstraction for Object communication. Provides a unified interface for multiple transport protocols including TCP, UDP, WebSockets, and gRPC. Handles connection management, pooling, multiplexing, and encryption. ## Features - Multiple transport protocol support - Connection pooling and reuse - Automatic reconnection with exponential backoff - TLS/SSL encryption - Message framing and fragmentation - Bandwidth throttling and QoS - Circuit breaker pattern """ use GenServer require Logger @type transport_type :: :tcp | :udp | :websocket | :grpc @type connection_opts :: [ host: String.t(), port: non_neg_integer(), transport: transport_type(), ssl: boolean(), pool_size: pos_integer(), timeout: timeout(), reconnect_interval: pos_integer(), max_reconnect_attempts: pos_integer() ] @type connection_id :: String.t() @type transport_state :: %{ connections: %{connection_id() => connection()}, pools: %{String.t() => [connection()]}, config: connection_opts(), metrics: transport_metrics() } @type connection :: %{ id: connection_id(), socket: port() | pid(), transport: transport_type(), host: String.t(), port: non_neg_integer(), state: :connected | :disconnected | :connecting, last_activity: DateTime.t(), bytes_sent: non_neg_integer(), bytes_received: non_neg_integer(), message_count: non_neg_integer(), error_count: non_neg_integer(), circuit_breaker: circuit_breaker_state() } @type circuit_breaker_state :: %{ state: :closed | :open | :half_open, failure_count: non_neg_integer(), last_failure: DateTime.t() | nil, success_count: non_neg_integer() } @type transport_metrics :: %{ total_connections: non_neg_integer(), active_connections: non_neg_integer(), total_bytes_sent: non_neg_integer(), total_bytes_received: non_neg_integer(), total_messages: non_neg_integer(), avg_latency_ms: float() } # Client API @doc """ Starts the network transport service. """ def start_link(opts \\ []) do GenServer.start_link(__MODULE__, opts, name: __MODULE__) end @doc """ Establishes a connection to a remote endpoint. """ @spec connect(String.t(), non_neg_integer(), Keyword.t()) :: {:ok, connection_id()} | {:error, term()} def connect(host, port, opts \\ []) do GenServer.call(__MODULE__, {:connect, host, port, opts}) end @doc """ Sends data over a specific connection. """ @spec send_data(connection_id(), binary()) :: :ok | {:error, term()} def send_data(conn_id, data) do GenServer.call(__MODULE__, {:send, conn_id, data}) end @doc """ Sends data to a specific endpoint, using connection pooling. """ @spec send_to(String.t(), non_neg_integer(), binary(), Keyword.t()) :: :ok | {:error, term()} def send_to(host, port, data, opts \\ []) do GenServer.call(__MODULE__, {:send_to, host, port, data, opts}) end @doc """ Closes a specific connection. """ @spec disconnect(connection_id()) :: :ok def disconnect(conn_id) do GenServer.cast(__MODULE__, {:disconnect, conn_id}) end @doc """ Gets metrics for the transport layer. """ @spec get_metrics() :: transport_metrics() def get_metrics do GenServer.call(__MODULE__, :get_metrics) end # Server Callbacks @impl true def init(opts) do config = Keyword.merge(default_config(), opts) state = %{ connections: %{}, pools: %{}, config: config, metrics: %{ total_connections: 0, active_connections: 0, total_bytes_sent: 0, total_bytes_received: 0, total_messages: 0, avg_latency_ms: 0.0 } } # Start connection health checker Process.send_after(self(), :check_connection_health, 5_000) {:ok, state} end @impl true def handle_call({:connect, host, port, opts}, _from, state) do transport = Keyword.get(opts, :transport, :tcp) ssl = Keyword.get(opts, :ssl, false) case establish_connection(host, port, transport, ssl, opts) do {:ok, connection} -> conn_id = connection.id new_state = put_in(state.connections[conn_id], connection) new_state = update_metrics(new_state, :connection_established) {:reply, {:ok, conn_id}, new_state} {:error, reason} = error -> Logger.error("Failed to connect to #{host}:#{port} - #{inspect(reason)}") {:reply, error, state} end end @impl true def handle_call({:send, conn_id, data}, _from, state) do case Map.get(state.connections, conn_id) do nil -> {:reply, {:error, :connection_not_found}, state} %{state: :connected} = conn -> case do_send_data(conn, data) do :ok -> new_conn = update_connection_metrics(conn, byte_size(data)) new_state = put_in(state.connections[conn_id], new_conn) new_state = update_metrics(new_state, :message_sent, byte_size(data)) {:reply, :ok, new_state} {:error, reason} -> new_conn = handle_send_error(conn, reason) new_state = put_in(state.connections[conn_id], new_conn) {:reply, {:error, reason}, new_state} end %{state: status} -> {:reply, {:error, {:connection_not_ready, status}}, state} end end @impl true def handle_call({:send_to, host, port, data, opts}, _from, state) do pool_key = "#{host}:#{port}" # Get or create connection from pool {conn, new_state} = get_pooled_connection(state, host, port, opts) case conn do nil -> {:reply, {:error, :no_available_connection}, new_state} _ -> case do_send_data(conn, data) do :ok -> updated_conn = update_connection_metrics(conn, byte_size(data)) new_state = update_pool_connection(new_state, pool_key, updated_conn) new_state = update_metrics(new_state, :message_sent, byte_size(data)) {:reply, :ok, new_state} {:error, reason} -> {:reply, {:error, reason}, new_state} end end end @impl true def handle_call(:get_metrics, _from, state) do {:reply, state.metrics, state} end @impl true def handle_cast({:disconnect, conn_id}, state) do case Map.get(state.connections, conn_id) do nil -> {:noreply, state} conn -> close_connection(conn) new_state = %{state | connections: Map.delete(state.connections, conn_id)} new_state = update_metrics(new_state, :connection_closed) {:noreply, new_state} end end @impl true def handle_info(:check_connection_health, state) do now = DateTime.utc_now() timeout_threshold = 60_000 # 60 seconds new_state = Enum.reduce(state.connections, state, fn {conn_id, conn}, acc -> last_activity = DateTime.to_unix(conn.last_activity, :millisecond) current_time = DateTime.to_unix(now, :millisecond) if current_time - last_activity > timeout_threshold do Logger.warning("Connection #{conn_id} timed out, closing") close_connection(conn) %{acc | connections: Map.delete(acc.connections, conn_id)} else # Send keepalive for TCP connections if conn.transport == :tcp and conn.state == :connected do :inet.setopts(conn.socket, [{:send_timeout, 5000}]) :gen_tcp.send(conn.socket, <<0::8>>) # Null byte keepalive end acc end end) Process.send_after(self(), :check_connection_health, 5_000) {:noreply, new_state} end @impl true def handle_info({:tcp, socket, data}, state) do case find_connection_by_socket(state.connections, socket) do {conn_id, conn} -> # Handle incoming data new_conn = %{conn | last_activity: DateTime.utc_now(), bytes_received: conn.bytes_received + byte_size(data) } new_state = put_in(state.connections[conn_id], new_conn) new_state = update_metrics(new_state, :message_received, byte_size(data)) # Forward to message handler send(self(), {:handle_message, conn_id, data}) {:noreply, new_state} nil -> Logger.warning("Received data from unknown socket") {:noreply, state} end end @impl true def handle_info({:tcp_closed, socket}, state) do case find_connection_by_socket(state.connections, socket) do {conn_id, conn} -> Logger.info("Connection #{conn_id} closed by remote") new_state = handle_connection_closed(state, conn_id, conn) {:noreply, new_state} nil -> {:noreply, state} end end @impl true def handle_info({:handle_message, conn_id, data}, state) do # This is where we'd integrate with the message protocol handler # For now, just log it Logger.debug("Received #{byte_size(data)} bytes on connection #{conn_id}") {:noreply, state} end # Private Functions defp default_config do [ pool_size: 10, timeout: 5_000, reconnect_interval: 1_000, max_reconnect_attempts: 5, ssl: false, transport: :tcp ] end defp establish_connection(host, port, :tcp, ssl, opts) do tcp_opts = [ :binary, packet: 4, # 4-byte length header active: true, nodelay: true, keepalive: true, send_timeout: Keyword.get(opts, :timeout, 5_000) ] connect_fun = if ssl, do: &:ssl.connect/3, else: &:gen_tcp.connect/3 case connect_fun.(to_charlist(host), port, tcp_opts) do {:ok, socket} -> conn = %{ id: generate_connection_id(), socket: socket, transport: :tcp, host: host, port: port, state: :connected, last_activity: DateTime.utc_now(), bytes_sent: 0, bytes_received: 0, message_count: 0, error_count: 0, circuit_breaker: %{ state: :closed, failure_count: 0, last_failure: nil, success_count: 0 } } {:ok, conn} {:error, reason} -> {:error, reason} end end defp establish_connection(host, port, :udp, _ssl, _opts) do udp_opts = [ :binary, active: true, reuseaddr: true ] case :gen_udp.open(0, udp_opts) do {:ok, socket} -> conn = %{ id: generate_connection_id(), socket: socket, transport: :udp, host: host, port: port, state: :connected, last_activity: DateTime.utc_now(), bytes_sent: 0, bytes_received: 0, message_count: 0, error_count: 0, circuit_breaker: %{ state: :closed, failure_count: 0, last_failure: nil, success_count: 0 } } {:ok, conn} {:error, reason} -> {:error, reason} end end defp establish_connection(_host, _port, transport, _ssl, _opts) do {:error, {:unsupported_transport, transport}} end defp do_send_data(%{transport: :tcp, socket: socket} = conn, data) do case check_circuit_breaker(conn) do :ok -> case :gen_tcp.send(socket, data) do :ok -> :ok {:error, reason} -> {:error, reason} end {:error, :circuit_open} = error -> error end end defp do_send_data(%{transport: :udp, socket: socket, host: host, port: port} = conn, data) do case check_circuit_breaker(conn) do :ok -> case :gen_udp.send(socket, to_charlist(host), port, data) do :ok -> :ok {:error, reason} -> {:error, reason} end {:error, :circuit_open} = error -> error end end defp check_circuit_breaker(%{circuit_breaker: %{state: :open}} = conn) do # Check if we should transition to half-open if should_attempt_reset?(conn.circuit_breaker) do :ok else {:error, :circuit_open} end end defp check_circuit_breaker(_conn), do: :ok defp should_attempt_reset?(%{last_failure: nil}), do: true defp should_attempt_reset?(%{last_failure: last_failure}) do DateTime.diff(DateTime.utc_now(), last_failure, :second) > 30 end defp close_connection(%{transport: :tcp, socket: socket}) do :gen_tcp.close(socket) end defp close_connection(%{transport: :udp, socket: socket}) do :gen_udp.close(socket) end defp close_connection(_conn), do: :ok defp update_connection_metrics(conn, bytes_sent) do %{conn | bytes_sent: conn.bytes_sent + bytes_sent, message_count: conn.message_count + 1, last_activity: DateTime.utc_now() } end defp handle_send_error(conn, _reason) do cb = update_circuit_breaker(conn.circuit_breaker, :failure) %{conn | error_count: conn.error_count + 1, circuit_breaker: cb } end defp update_circuit_breaker(cb, :failure) do failure_count = cb.failure_count + 1 if failure_count >= 5 do %{cb | state: :open, failure_count: failure_count, last_failure: DateTime.utc_now() } else %{cb | failure_count: failure_count} end end defp update_circuit_breaker(cb, :success) do %{cb | state: :closed, failure_count: 0, success_count: cb.success_count + 1 } end defp get_pooled_connection(state, host, port, opts) do pool_key = "#{host}:#{port}" pool = Map.get(state.pools, pool_key, []) # Find available connection case Enum.find(pool, fn conn -> conn.state == :connected and check_circuit_breaker(conn) == :ok end) do nil -> # Create new connection case establish_connection(host, port, Keyword.get(opts, :transport, :tcp), Keyword.get(opts, :ssl, false), opts) do {:ok, conn} -> new_pool = [conn | pool] |> Enum.take(state.config[:pool_size]) new_state = put_in(state.pools[pool_key], new_pool) {conn, new_state} {:error, _reason} -> {nil, state} end conn -> {conn, state} end end defp update_pool_connection(state, pool_key, updated_conn) do pool = Map.get(state.pools, pool_key, []) new_pool = Enum.map(pool, fn conn -> if conn.id == updated_conn.id, do: updated_conn, else: conn end) put_in(state.pools[pool_key], new_pool) end defp find_connection_by_socket(connections, socket) do Enum.find(connections, fn {_id, conn} -> conn.socket == socket end) end defp handle_connection_closed(state, conn_id, conn) do close_connection(conn) new_state = %{state | connections: Map.delete(state.connections, conn_id)} update_metrics(new_state, :connection_closed) end defp update_metrics(state, event, bytes \\ 0) do metrics = case event do :connection_established -> %{state.metrics | total_connections: state.metrics.total_connections + 1, active_connections: state.metrics.active_connections + 1 } :connection_closed -> %{state.metrics | active_connections: max(0, state.metrics.active_connections - 1) } :message_sent -> %{state.metrics | total_bytes_sent: state.metrics.total_bytes_sent + bytes, total_messages: state.metrics.total_messages + 1 } :message_received -> %{state.metrics | total_bytes_received: state.metrics.total_bytes_received + bytes } _ -> state.metrics end %{state | metrics: metrics} end defp generate_connection_id do :crypto.strong_rand_bytes(16) |> Base.encode16(case: :lower) end end