Milvex.Connection (milvex v0.14.1)

Copy Markdown

State machine managing gRPC channel lifecycle with automatic reconnection.

Each connection maintains a gRPC channel to a Milvus server and monitors the underlying connection process for failures.

States

  • :connecting - Attempting to establish initial connection
  • :connected - Channel active, connection monitored
  • :reconnecting - Lost connection, attempting to restore

Usage

# Start a connection
{:ok, conn} = Milvex.Connection.start_link(host: "localhost", port: 19530)

# Get the gRPC channel for making calls
{:ok, channel} = Milvex.Connection.get_channel(conn)

# Disconnect
:ok = Milvex.Connection.disconnect(conn)

Named Connections

# Start a named connection
{:ok, _} = Milvex.Connection.start_link([host: "localhost"], name: :milvus)

# Use the named connection
{:ok, channel} = Milvex.Connection.get_channel(:milvus)

Reconnection Behavior

The connection monitors the underlying gRPC connection process. When the connection dies, it automatically reconnects using exponential backoff with jitter to prevent thundering herd problems.

Configuration options:

  • :reconnect_base_delay - Base delay in ms (default: 1000)
  • :reconnect_max_delay - Maximum delay cap in ms (default: 60000)
  • :reconnect_multiplier - Exponential multiplier (default: 2.0)
  • :reconnect_jitter - Jitter factor 0.0-1.0 (default: 0.1)

Connection Pooling

HTTP/2 multiplexes all RPCs over a single connection, and servers cap the number of concurrent streams per connection (SETTINGS_MAX_CONCURRENT_STREAMS). Strict clients such as the Mint adapter reject requests over that limit with :too_many_concurrent_requests. Set :pool_size to open multiple connections and distribute RPCs round-robin across them:

{:ok, pool} = Milvex.Connection.start_link(host: "localhost", pool_size: 4)

The pool answers the same call protocol as a single connection, so the returned pid (or registered name) can be passed anywhere a connection is expected. With pool_size: 1 (the default) a single connection is started and behavior is unchanged. See Milvex.ConnectionPool for details.

Summary

Types

channel()

gRPC channel used by Milvex connections.

state()
t()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

connected(arg1, arg2, data)
connected?(conn, opts \\ [])

Checks if the connection is currently established.

connecting(arg1, arg2, data)
disconnect(conn)

Disconnects from the Milvus server and stops the connection process.

get_channel(conn, opts \\ [])

Gets the gRPC channel and connection config from the connection.

get_config(conn, opts \\ [])

Gets the connection config without picking a channel.

reconnecting(arg1, arg2, data)
start_link(opts)

Starts a connection to a Milvus server.

Types

channel()

@type channel() :: %GRPC.Channel{
  accepted_compressors: term(),
  adapter: term(),
  adapter_payload: term(),
  codec: term(),
  compressor: term(),
  cred: term(),
  headers: term(),
  host: term(),
  interceptors: term(),
  port: term(),
  ref: term(),
  scheme: term()
}

gRPC channel used by Milvex connections.

state()

@type state() :: :connecting | :connected | :reconnecting

t()

@type t() :: %Milvex.Connection{
  channel: channel() | nil,
  config: Milvex.Config.t(),
  conn_monitor_ref: reference() | nil,
  registry: {:ets.tid(), pos_integer()} | nil,
  retry_count: non_neg_integer()
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

connected(arg1, arg2, data)

connected?(conn, opts \\ [])

@spec connected?(
  GenServer.server(),
  keyword()
) :: boolean()

Checks if the connection is currently established.

connecting(arg1, arg2, data)

disconnect(conn)

@spec disconnect(GenServer.server()) :: :ok

Disconnects from the Milvus server and stops the connection process.

get_channel(conn, opts \\ [])

@spec get_channel(
  GenServer.server(),
  keyword()
) :: {:ok, channel(), Milvex.Config.t()} | {:error, Milvex.Error.t()}

Gets the gRPC channel and connection config from the connection.

Returns {:ok, channel, config} if connected, or {:error, error} if not connected. The config map contains the full Milvex.Config.t() used by this connection.

get_config(conn, opts \\ [])

@spec get_config(
  GenServer.server(),
  keyword()
) :: {:ok, Milvex.Config.t()} | {:error, Milvex.Error.t()}

Gets the connection config without picking a channel.

The config is static after startup, so unlike get_channel/2 this does not consume a round-robin slot on pooled connections and succeeds even while the connection is (re)connecting.

reconnecting(arg1, arg2, data)

start_link(opts)

@spec start_link(keyword()) :: GenStateMachine.on_start()

Starts a connection to a Milvus server.

Options

  • :name - Optional name to register the connection process
  • :pool_size - When greater than 1, starts a Milvex.ConnectionPool of that many connections instead of a single connection
  • All other options are passed to Milvex.Config.parse/1

Examples

{:ok, conn} = Milvex.Connection.start_link(host: "localhost", port: 19530)
{:ok, conn} = Milvex.Connection.start_link([host: "localhost"], name: :milvus)
{:ok, pool} = Milvex.Connection.start_link(host: "localhost", pool_size: 4)