Client SDKs Elixir Reference Client KubeMQ.Client GenServer API reference — connection, lifecycle, and all messaging operations for the Elixir SDK.
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.
Requires Elixir 1.15+.
{ :ok , client} = KubeMQ . Client . start_link ( address: "localhost:50000" , client_id: "my-app" ) address is optional — omit it to use the default:
{ :ok , client} = KubeMQ . Client . start_link ( client_id: "my-app" ) Option Type Required Description addressString.t()No KubeMQ server endpoint (host:port) — defaults to "localhost:50000" client_idString.t()Yes Unique client identifier auth_tokenString.t()No Authentication bearer token tlskeyword()No SSL options: cacertfile, certfile, keyfile, verify rpc_timeoutpos_integer()No Default RPC timeout in ms (default 10_000) reconnect_policykeyword()No initial_delay, max_delay, max_attempts, multiplierdefault_cache_ttlpos_integer()No Default cache TTL for queries in ms nameatom()No Register the GenServer under a name
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) :: :okClose connection and stop the GenServer
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
Option Type Description on_event(EventReceive.t() -> any())Callback for received events on_error(Error.t() -> any())Callback for subscription errors groupString.t()Consumer group name
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
Option Type Description start_atatom or tuple :start_new_only, :start_from_first, :start_from_last, {:start_at_sequence, n}, {:start_at_time, unix}, {:start_at_time_delta, ms}groupString.t()Consumer group name on_event(EventStoreReceive.t() -> any())Callback for received events
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
Option Type Description on_command(CommandReceive.t() -> CommandReply.t())Handler that returns a reply on_error(Error.t() -> any())Error callback groupString.t()Consumer group name
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
Option Type Description on_query(QueryReceive.t() -> QueryReply.t())Handler that returns a reply on_error(Error.t() -> any())Error callback groupString.t()Consumer group name
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
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
{ :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) Was this page helpful?
Yes No