Pulsar.Client (Pulsar v3.2.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 pool by URL and selects one of its connections.

producers(client_name \\ :default)

Returns the producer resources currently running under a client.

random_broker(client_name \\ :default)

Returns a connection from a random broker pool registered to the specified client.

start_broker(broker_url, opts \\ [])

Selects 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 and removes the entire pool for a broker URL, including all of its connections.

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 | :disconnected}

Looks up an existing broker pool by URL and selects one of its connections.

Returns {:ok, broker_pid} if found, or an error when the pool or requested connection is unavailable.

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 connection from a random broker pool 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()}

Selects a broker connection.

If a pool for the given URL already exists, returns one of its connections. Otherwise, starts a pool with the client-configured number of connections and returns one.

Broker connection options passed to this function override the client's settings when creating a new pool. Once a pool exists, later calls select from its stored configuration instead of reconfiguring it.

The internal :connection_slot option selects a numbered pool member. By default, any live connection process is returned, including one that is currently reconnecting.

Returns {:ok, broker_pid} if successful or {:error, reason} for runtime failures. Invalid broker connection options raise NimbleOptions.ValidationError.

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.

  • :connections_per_broker (pos_integer/0) - Connections this client opens to each broker. Consumer and producer workers are assigned slots round-robin and retain their slot across worker and group restarts. Each slot adds one process and TCP connection per discovered broker. Defaults to one. The default value is 1.

  • :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 [].

  • :worker_restart_intensity (keyword/0) - How often a consumer or producer worker under this client may be restarted before its partition gives up, as [max_restarts: integer, max_seconds: integer]. OTP's own intensity by default.

    A group multiplies :max_restarts by its worker count, so a broker dropping every worker registered with it at once counts as one round rather than as many. See docs/architecture.md for what that trades and how the two numbers relate to Pulsar.Backoff.

    The default value is [max_restarts: 3, max_seconds: 5].

  • :resource_restart_intensity (keyword/0) - How many times a partition may give up before the consumer or producer it belongs to does, and how many resources may do that before the client does. The failure then reaches whatever supervises the client.

    Much smaller than :worker_restart_intensity, and deliberately so: a partition only gives up when it genuinely cannot run, and tying this to the larger budget would make escalation depend on how quickly the failure comes back.

    The default value is [max_restarts: 3, max_seconds: 5].

  • :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 | :running | :restarting | :unavailable}

Stops and removes the entire pool for a broker URL, including all of its connections.

A later lookup may discover and start a new pool for the URL. If this removes the client's final pool, discovery has no broker to ask until start_broker/2 starts one explicitly or the client restarts its configured bootstrap pool.

Returns :ok after removing the pool, {:error, :not_found} when the client or pool is absent, {:error, :running} or {:error, :restarting} if a concurrent operation starts it during removal, or {:error, :unavailable} if the broker supervisor fails during the operation.