# Client (/sdks/elixir/reference/client) `KubeMQ.Client` is an OTP GenServer that manages the gRPC connection and exposes all messaging operations. Start it with `start_link/1` or add it to your supervision tree. ## Package [#package] ```elixir title="mix.exs" {:kubemq, "~> 1.0"} ``` Requires Elixir 1.15+. ## Starting the Client [#starting-the-client] ```elixir {:ok, client} = KubeMQ.Client.start_link( address: "localhost:50000", client_id: "my-app" ) ``` `address` is optional — omit it to use the default: ```elixir {:ok, client} = KubeMQ.Client.start_link(client_id: "my-app") ``` ### Constructor Options [#constructor-options] | Option | Type | Required | Description | | ------------------- | --------------- | -------- | ---------------------------------------------------------------------- | | `address` | `String.t()` | No | KubeMQ server endpoint (`host:port`) — defaults to `"localhost:50000"` | | `client_id` | `String.t()` | Yes | Unique client identifier | | `auth_token` | `String.t()` | No | Authentication bearer token | | `tls` | `keyword()` | No | SSL options: `cacertfile`, `certfile`, `keyfile`, `verify` | | `rpc_timeout` | `pos_integer()` | No | Default RPC timeout in ms (default `10_000`) | | `reconnect_policy` | `keyword()` | No | `initial_delay`, `max_delay`, `max_attempts`, `multiplier` | | `default_cache_ttl` | `pos_integer()` | No | Default cache TTL for queries in ms | | `name` | `atom()` | No | Register the GenServer under a name | ## Connection Lifecycle [#connection-lifecycle] | Function | Spec | Description | | -------------------- | ---------------------------------------------------------- | --------------------------------------- | | `start_link/1` | `(keyword()) :: {:ok, pid()} \| {:error, term()}` | Start and connect the client | | `connected?/1` | `(client) :: boolean()` | Check if client is connected | | `connection_state/1` | `(client) :: atom()` | Current connection state | | `ping/1` | `(client) :: {:ok, ServerInfo.t()} \| {:error, Error.t()}` | Ping the server | | `close/1` | `(client) :: :ok` | Close connection and stop the GenServer | ## Events [#events] | Function | Spec | Description | | ----------------------- | --------------------------------------------------------------------------- | -------------------------------- | | `send_event/2` | `(client, Event.t()) :: :ok \| {:error, Error.t()}` | Send a fire-and-forget event | | `send_event_stream/1` | `(client) :: {:ok, EventStreamHandle.t()} \| {:error, Error.t()}` | Open a persistent event stream | | `subscribe_to_events/3` | `(client, channel, opts) :: {:ok, Subscription.t()} \| {:error, Error.t()}` | Subscribe to events on a channel | ### Subscribe Options [#subscribe-options] | Option | Type | Description | | ---------- | ----------------------------- | -------------------------------- | | `on_event` | `(EventReceive.t() -> any())` | Callback for received events | | `on_error` | `(Error.t() -> any())` | Callback for subscription errors | | `group` | `String.t()` | Consumer group name | ## Events Store [#events-store] | Function | Spec | Description | | ----------------------------- | -------------------------------------------------------------------------------- | ------------------------------------ | | `send_event_store/2` | `(client, EventStore.t()) :: {:ok, EventStoreResult.t()} \| {:error, Error.t()}` | Send a persistent event | | `send_event_store_stream/1` | `(client) :: {:ok, EventStoreStreamHandle.t()} \| {:error, Error.t()}` | Open a persistent event store stream | | `subscribe_to_events_store/3` | `(client, channel, opts) :: {:ok, Subscription.t()} \| {:error, Error.t()}` | Subscribe with replay options | ### Events Store Subscribe Options [#events-store-subscribe-options] | Option | Type | Description | | ---------- | ---------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------- | | `start_at` | atom or tuple | `:start_new_only`, `:start_from_first`, `:start_from_last`, `{:start_at_sequence, n}`, `{:start_at_time, unix}`, `{:start_at_time_delta, ms}` | | `group` | `String.t()` | Consumer group name | | `on_event` | `(EventStoreReceive.t() -> any())` | Callback for received events | ## Commands [#commands] | Function | Spec | Description | | ------------------------- | ---------------------------------------------------------------------------- | ------------------------------------ | | `send_command/2` | `(client, Command.t()) :: {:ok, CommandResponse.t()} \| {:error, Error.t()}` | Send a command and wait for response | | `subscribe_to_commands/3` | `(client, channel, opts) :: {:ok, Subscription.t()} \| {:error, Error.t()}` | Subscribe to handle commands | ### Command Subscribe Options [#command-subscribe-options] | Option | Type | Description | | ------------ | ------------------------------------------ | ---------------------------- | | `on_command` | `(CommandReceive.t() -> CommandReply.t())` | Handler that returns a reply | | `on_error` | `(Error.t() -> any())` | Error callback | | `group` | `String.t()` | Consumer group name | ## Queries [#queries] | Function | Spec | Description | | ------------------------ | --------------------------------------------------------------------------- | ---------------------------------- | | `send_query/2` | `(client, Query.t()) :: {:ok, QueryResponse.t()} \| {:error, Error.t()}` | Send a query and wait for response | | `subscribe_to_queries/3` | `(client, channel, opts) :: {:ok, Subscription.t()} \| {:error, Error.t()}` | Subscribe to handle queries | ### Query Subscribe Options [#query-subscribe-options] | Option | Type | Description | | ---------- | -------------------------------------- | ---------------------------- | | `on_query` | `(QueryReceive.t() -> QueryReply.t())` | Handler that returns a reply | | `on_error` | `(Error.t() -> any())` | Error callback | | `group` | `String.t()` | Consumer group name | ## Queues [#queues] | Function | Spec | Description | | -------------------------- | ------------------------------------------------------------------------------------ | ----------------------------------- | | `send_queue_message/2` | `(client, QueueMessage.t()) :: {:ok, QueueSendResult.t()} \| {:error, Error.t()}` | Send a single queue message | | `send_queue_messages/2` | `(client, [QueueMessage.t()]) :: {:ok, QueueBatchResult.t()} \| {:error, Error.t()}` | Send a batch of messages | | `receive_queue_messages/3` | `(client, channel, opts) :: {:ok, QueueReceiveResult.t()} \| {:error, Error.t()}` | Receive messages from a queue | | `ack_all_queue_messages/3` | `(client, channel, opts) :: {:ok, QueueAckAllResult.t()} \| {:error, Error.t()}` | Acknowledge all pending messages | | `poll_queue/2` | `(client, opts) :: {:ok, PollResponse.t()} \| {:error, Error.t()}` | Poll with transactional control | | `queue_upstream/1` | `(client) :: {:ok, QueueUpstreamHandle.t()} \| {:error, Error.t()}` | Open an upstream stream for sending | ## Channel Management [#channel-management] | Function | Spec | Description | | ----------------------- | --------------------------------------------------------------------------------- | ------------------------------- | | `create_channel/3` | `(client, name, type) :: :ok \| {:error, Error.t()}` | Create a channel of given type | | `delete_channel/3` | `(client, name, type) :: :ok \| {:error, Error.t()}` | Delete a channel | | `list_channels/3` | `(client, type, filter \\ "") :: {:ok, [ChannelInfo.t()]} \| {:error, Error.t()}` | List channels by type | | `purge_queue_channel/2` | `(client, channel) :: :ok \| {:error, Error.t()}` | Purge all messages from a queue | Channel types: `:events`, `:events_store`, `:commands`, `:queries`, `:queues` ## Quick Usage [#quick-usage] ```elixir title="main.exs" {:ok, client} = KubeMQ.Client.start_link( address: "localhost:50000", client_id: "elixir-demo" ) {:ok, info} = KubeMQ.Client.ping(client) IO.puts("Connected to #{info.host} v#{info.version}") KubeMQ.Client.close(client) ``` ## See Also [#see-also] * [Connection Examples](/sdks/elixir/how-to/connection/) * [Elixir SDK Getting Started](/sdks/elixir)