Pulsar.Client (Pulsar v3.0.0)

Copy Markdown View Source

A client represents an isolated Pulsar connection context and owns the consumers and producers that use it.

Usage

The client is the only thing that belongs in the host application's supervision tree. Consumers and producers are declared on it and run underneath it:

children = [
  {Pulsar.Client,
   host: "pulsar://localhost:6650",
   consumers: [
     [topic: topic, subscription_name: "sub", callback_module: MyCallback]
   ]}
]

Supervisor.start_link(children, strategy: :one_for_one)

Several named clients can coexist. Consumers and producers select one with :client, defaulting to :default.

Declared and runtime resources

For sets only known at runtime — a consumer per tenant, say — Pulsar.Consumer.start/1 and Pulsar.Producer.start/1 add to a running client:

Pulsar.Consumer.start(
  topic: topic,
  subscription_name: "sub",
  callback_module: MyCallback,
  client: :analytics
)

Declared resources are recreated after their client or resource branch restarts. Runtime resources are not; their caller must restore them.

Both forms initialize asynchronously. Pulsar.Consumer.await_ready/2 and Pulsar.Producer.await_ready/2 provide a bounded topology-and-worker readiness barrier when one is needed.

consumers/1 and producers/1 list the logical resources currently running under a client. Partitioned resources still appear once: the returned pid is their stable root, not one entry per partition or worker.

See the architecture guide for the complete ownership and recovery model.

Summary

Functions

consumers(client_name \\ :default)

Returns the consumer resources currently running under a client.

lookup_broker(broker_url, opts \\ [])

Looks up an existing broker connection by broker URL.

producers(client_name \\ :default)

Returns the producer resources currently running under a client.

random_broker(client_name \\ :default)

Returns a random broker process registered to the specified client.

start_broker(broker_url, opts \\ [])

Starts a broker connection.

start_link(opts)

Starts a client with the given options.

stop(client_name, opts \\ [])

Stops a client, and with it every consumer, producer and broker connection it owns.

stop_broker(broker_url, opts \\ [])

Stops a broker connection by broker URL.

Functions

consumers(client_name \\ :default)

@spec consumers(atom()) :: [pid()]

Returns the consumer resources currently running under a client.

Each pid is the stable topology root returned by Pulsar.Consumer.start/1, regardless of how many partitions or workers that consumer has. Returns an empty list while the client or its consumer branch is unavailable. The order is unspecified.

lookup_broker(broker_url, opts \\ [])

@spec lookup_broker(
  String.t(),
  keyword()
) :: {:ok, pid()} | {:error, :not_found}

Looks up an existing broker connection by broker URL.

Returns {:ok, broker_pid} if found, {:error, :not_found} otherwise.

producers(client_name \\ :default)

@spec producers(atom()) :: [pid()]

Returns the producer resources currently running under a client.

Each pid is the stable topology root returned by Pulsar.Producer.start/1, regardless of how many partitions or workers that producer has. Returns an empty list while the client or its producer branch is unavailable. The order is unspecified.

random_broker(client_name \\ :default)

@spec random_broker(atom()) :: pid() | nil

Returns a random broker process registered to the specified client.

Defaults to the :default client if no client is specified.

This is useful for operations that need any broker from a client (e.g., service discovery).

start_broker(broker_url, opts \\ [])

@spec start_broker(
  String.t(),
  keyword()
) :: {:ok, pid()} | {:error, term()}

Starts a broker connection.

If a broker for the given URL already exists, returns the existing broker. Otherwise, starts a new broker connection with the provided options.

Returns {:ok, broker_pid} if successful, {:error, reason} otherwise.

start_link(opts)

Starts a client with the given options.

Options

  • :name (atom/0) - Name the client is registered under, and the name consumers and producers select it by. Defaults to :default, which is also their default :client. The default value is :default.

  • :host - Required. Bootstrap broker URL, e.g. pulsar://localhost:6650.

  • :consumers (list of keyword/0) - Consumers declared under this client, each a keyword list of Pulsar.Consumer options. Their :client is set to this one. See the module documentation for the lifecycle of declared resources. The default value is [].

  • :producers (list of keyword/0) - Producers declared under this client, each a keyword list of Pulsar.Producer options. Their :client is set to this one.

    Consumers and producers initialize independently, so callbacks that publish during startup must handle {:error, :not_found} and {:error, :not_ready}.

    The default value is [].

  • :auth (keyword/0) - Authentication configuration, as [type: module, opts: keyword]. The default value is [type: Pulsar.Auth.None, opts: []].

  • :conn_timeout (timeout/0) - Milliseconds to wait for a connection to a broker. :infinity waits indefinitely, which leaves the broker process blocked in connect with no reconnect timer and no way to answer calls until the network gives up. The default value is 1000.

  • :max_frame_size (pos_integer/0) - Largest frame accepted from this cluster, in bytes. Raise it to match a broker configured with a larger maxMessageSize. The default value is 5253120.

  • :ping_interval (pos_integer/0) - Milliseconds between keepalive pings to each broker in this cluster. The default value is 60000.

  • :cleanup_interval (pos_integer/0) - Milliseconds between sweeps for requests that never got a response. The default value is 30000.

  • :request_timeout (pos_integer/0) - Milliseconds after which a request without a response is failed. The default value is 60000.

  • :socket_opts (list of term/0) - Options passed to :gen_tcp.connect/4 or :ssl.connect/4. Defaults to verifying the broker's certificate against the CA bundle from :castore. Not a keyword list: bare atoms such as :inet6 and tuples such as {:raw, level, opt, value} are valid entries.

stop(client_name, opts \\ [])

@spec stop(
  atom(),
  keyword()
) :: :ok

Stops a client, and with it every consumer, producer and broker connection it owns.

For a client you started yourself, from a script or IEx. A client in a supervision tree is restarted by its supervisor whatever its exit reason, so this only cycles it; stop those by removing them from the tree.

Options

  • :timeout - Maximum time to wait for shutdown (default: 5000ms)

Examples

Pulsar.Client.stop(:my_client)

stop_broker(broker_url, opts \\ [])

@spec stop_broker(
  String.t(),
  keyword()
) :: :ok | {:error, :not_found}

Stops a broker connection by broker URL.