initial client lb implementation

Author: sleipnir

lib/grpc/client/resolver.ex

@@ -0,0 +1,72 @@
+defmodule GRPC.Client.Resolver do
+  @moduledoc """
+  Behaviour for gRPC client resolvers.
+  """
+  @type service_config :: GRPC.Client.ServiceConfig.t() | nil
+
+  @callback resolve(String.t()) ::
+              {:ok, %{addresses: list(map()), service_config: service_config()}}
+              | {:error, term()}
+
+  @behaviour GRPC.Client.Resolver
+
+  @doc """
+  Resolves a gRPC target string into a list of connection endpoints and an optional ServiceConfig.
+
+  The `target` string can use one of the supported URI schemes:
+
+    * `dns://[authority/]host[:port]` – resolves via DNS; looks up both A/AAAA records and optional `_grpc_config.<host>` TXT record.
+    * `ipv4:addr[:port][,addr[:port],...]` – uses a fixed list of IPv4 addresses.
+    * `ipv6:[addr][:port][,[addr][:port],...]` – uses a fixed list of IPv6 addresses.
+    * `unix:/absolute_path` – connects via Unix domain socket.
+    * `unix-abstract:name` – connects via abstract Unix socket (Linux only).
+    * `vsock:cid:port` – connects via VSOCK (Linux only).
+    * `xds:///name` – resolves via xDS control plane (Envoy/Istio/Traffic Director).
+
+  If no scheme is specified, `dns` is assumed. Default ports:
+
+    * `dns`, `ipv4`, `ipv6` → 50051
+    * `xds` → 443
+
+  Returns:
+
+    * `{:ok, %{addresses: list(map()), service_config: GRPC.Client.ServiceConfig.t() | nil}}` on success
+    * `{:error, reason}` on failure
+
+  Each `address` map includes at least:
+
+    * `:address` – host, IP, or socket path
+    * `:port` – TCP port (if applicable)
+    * additional fields may be present depending on the scheme (e.g., `:socket`, `:cid` for vsock).
+
+  This function abstracts the resolution mechanism, allowing the gRPC client to obtain endpoints and service configuration regardless of the underlying target type.
+  """
+  @impl GRPC.Client.Resolver
+  @spec resolve(String.t()) ::
+          {:ok, %{addresses: list(map()), service_config: GRPC.Client.ServiceConfig.t()}}
+          | {:error, term()}
+  def resolve(target) do
+    uri = URI.parse(target)
+    scheme = uri.scheme || "dns"
+
+    case scheme do
+      "dns" ->
+        GRPC.Client.Resolver.DNS.resolve(target)
+
+      "ipv4" ->
+        GRPC.Client.Resolver.IPv4.resolve(target)
+
+      "ipv6" ->
+        GRPC.Client.Resolver.IPv6.resolve(target)
+
+      "unix" ->
+        GRPC.Client.Resolver.Unix.resolve(target)
+
+      "xds" ->
+        GRPC.Client.Resolver.XDS.resolve(target)
+
+      _ ->
+        {:error, {:unknown_scheme, scheme}}
+    end
+  end
+end

lib/grpc/client/resolver/dns.ex

@@ -0,0 +1,38 @@
+defmodule GRPC.Client.Resolver.DNS do
+  @behaviour GRPC.Client.Resolver
+
+  alias GRPC.Client.ServiceConfig
+
+  @impl GRPC.Client.Resolver
+  def resolve(target) do
+    uri = URI.parse(target)
+    host = uri.host || target
+    port = uri.port || 50051
+
+    # resolve A/AAAA
+    {:ok, addresses} = :inet_res.lookup(String.to_charlist(host), :in, :a)
+
+    addrs =
+      Enum.map(addresses, fn ip ->
+        %{address: :inet.ntoa(ip) |> to_string(), port: port}
+      end)
+
+    # tries to resolve TXT to service config
+    service_config_json =
+      case :inet_res.lookup(~c"_grpc_config." ++ String.to_charlist(host), :in, :txt) do
+        [txt | _] ->
+          str = List.to_string(txt)
+
+          # TXT may have prefix "grpc_config="
+          case String.split(str, "grpc_config=") do
+            [_, json] -> json
+            _ -> nil
+          end
+
+        _ ->
+          nil
+      end
+
+    {:ok, %{addresses: addrs, service_config: ServiceConfig.parse(service_config_json)}}
+  end
+end

lib/grpc/client/resolver/ipv4.ex

@@ -0,0 +1,18 @@
+defmodule GRPC.Client.Resolver.IPv4 do
+  @behaviour GRPC.Client.Resolver
+
+  @impl GRPC.Client.Resolver
+  def resolve(target) do
+    # target exemplo: "ipv4:10.0.0.1:50051,10.0.0.2:50052"
+    [_scheme, addrs_str] = String.split(target, ":", parts: 2)
+
+    addresses =
+      String.split(addrs_str, ",")
+      |> Enum.map(fn addr ->
+        [ip, port] = String.split(addr, ":")
+        %{address: ip, port: String.to_integer(port)}
+      end)
+
+    {:ok, %{addresses: addresses, service_config: nil}}
+  end
+end

lib/grpc/client/resolver/ipv6.ex

@@ -0,0 +1,23 @@
+defmodule GRPC.Client.Resolver.IPv6 do
+  @behaviour GRPC.Client.Resolver
+
+  @impl GRPC.Client.Resolver
+  def resolve(target) do
+    # target example: "ipv6:[2607:f8b0:400e:c00::ef]:443,[::1]:50051"
+    [_scheme, addrs_str] = String.split(target, ":", parts: 2)
+
+    addresses =
+      String.split(addrs_str, ",")
+      |> Enum.map(fn entry ->
+        [ip, port] =
+          case Regex.run(~r/\[(.*?)\]:(\d+)/, entry) do
+            [_, ip, port] -> [ip, port]
+            _ -> [entry, "443"]
+          end
+
+        %{address: ip, port: String.to_integer(port)}
+      end)
+
+    {:ok, %{addresses: addresses, service_config: nil}}
+  end
+end

lib/grpc/client/resolver/unix.ex

@@ -0,0 +1,12 @@
+defmodule GRPC.Client.Resolver.Unix do
+  @behaviour GRPC.Client.Resolver
+
+  @impl GRPC.Client.Resolver
+  def resolve(target) do
+    # E.g.: "unix:///var/run/my.sock"
+    uri = URI.parse(target)
+    path = uri.path
+
+    {:ok, %{addresses: [%{address: path, port: nil, socket: :unix}], service_config: nil}}
+  end
+end

lib/grpc/client/resolver/xds.ex

@@ -0,0 +1,9 @@
+defmodule GRPC.Client.Resolver.XDS do
+  @behaviour GRPC.Client.Resolver
+
+  @impl GRPC.Client.Resolver
+  def resolve(_target) do
+    # E.g.: "xds:///myservice"
+    {:error, :not_implemented}
+  end
+end

lib/grpc/client/service_config.ex

@@ -0,0 +1,38 @@
+defmodule GRPC.Client.ServiceConfig do
+  @moduledoc """
+  Represents the gRPC ServiceConfig parsed from JSON (from DNS TXT or xDS).
+  """
+
+  defstruct load_balancing_policy: :pick_first,
+            method_configs: [],
+            raw: %{}
+
+  @type t :: %__MODULE__{
+          load_balancing_policy: atom(),
+          method_configs: list(),
+          raw: map()
+        }
+
+  def parse(nil), do: {:ok, %__MODULE__{}}
+
+  def parse(json) when is_binary(json) do
+    case Jason.decode(json) do
+      {:ok, map} -> {:ok, from_map(map)}
+      error -> error
+    end
+  end
+
+  defp from_map(map) do
+    lb =
+      map
+      |> Map.get("loadBalancingPolicy", "pick_first")
+      |> String.downcase()
+      |> String.to_existing_atom()
+
+    %__MODULE__{
+      load_balancing_policy: lb,
+      method_configs: Map.get(map, "methodConfig", []),
+      raw: map
+    }
+  end
+end

lib/grpc/server/supervisor.ex

@@ -55,7 +55,7 @@ defmodule GRPC.Server.Supervisor do
   end
 
   def init(opts) when is_list(opts) do
-    unless is_nil(Application.get_env(:grpc, :start_server)) do
+    if not is_nil(Application.get_env(:grpc, :start_server)) do
       raise "the :start_server config key has been deprecated.\
       The currently supported way is to configure it\
       through the :start_server option for the GRPC.Server.Supervisor"

lib/grpc/stream.ex

@@ -148,7 +148,7 @@ defmodule GRPC.Stream do
   """
   @spec run(t()) :: any()
   def run(%__MODULE__{flow: flow, options: opts}) do
-    unless Keyword.get(opts, :unary, false) do
+    if !Keyword.get(opts, :unary, false) do
       raise ArgumentError, "run/2 is not supported for non-unary streams"
     end
 

lib/grpc/stub.ex

@@ -231,7 +231,7 @@ defmodule GRPC.Stub do
 
     adapter_opts = opts[:adapter_opts]
 
-    unless is_list(adapter_opts) do
+    if not is_list(adapter_opts) do
       raise ArgumentError, ":adapter_opts must be a keyword list if present"
     end