View Source PulsarEx (PulsarEx v0.15.0)

Summary

Types

namespace()

subscription()

tenant()

topic()

topic_name()

Functions

active_consumers(cluster_name, tenant, namespace, subscription)

Count active (started) consumers for cluster/tenant/namespace/subscription, cheaply - a single Registry scan, no per-consumer calls

clusters(tenant, namespace, subscription)

List the distinct clusters with started consumers for tenant/namespace/subscription, so callers don't have to know a worker's configured cluster ahead of time. Uses ConsumerIDRegistry rather than ConsumerReadyRegistry so a cluster is still discovered while its consumers are down/reconnecting, not just while healthy.

consumer_gauges(cluster_name, tenant, namespace, subscription)

Desired, active, and ready consumer counts for cluster/tenant/namespace/subscription - desired comes from the consumer manager, active/ready from local registries

consumers()

List all the consumers on default cluster

consumers(cluster_name)

List all the consumers for cluster

consumers(cluster_name, topic_name)

List all the consumers for cluster on a specific topic

consumers(cluster_name, topic_name, subscription)

List all the consumers for cluster on a specific topic with specific subscription

produce(topic_name, payload, message_opts \\ [], producer_opts \\ [])

Publish a message to pulsar, create a pool of producers if not yet created

ready_consumers(cluster_name, tenant, namespace, subscription)

Count consumers that have reached the :READY state for cluster/tenant/namespace/subscription

start_consumer(topic_name, subscription, module, opts)

Start consumer from a fully quantified topic

start_consumer(tenant, namespace, topic, subscription, module, opts)

Start consumers with topic pattern

stop_consumer(topic, subscription)

Stop consumers with fully quantified topic

stop_consumer(tenant, namespace, topic, subscription)

Stop consumers with topic pattern

Types

namespace()

@type namespace() :: String.t() | Regex.t()

subscription()

@type subscription() :: String.t()

tenant()

@type tenant() :: String.t() | Regex.t()

topic()

@type topic() :: String.t() | Regex.t()

topic_name()

@type topic_name() :: String.t()

Functions

active_consumers(cluster_name, tenant, namespace, subscription)

Count active (started) consumers for cluster/tenant/namespace/subscription, cheaply - a single Registry scan, no per-consumer calls

clusters(tenant, namespace, subscription)

List the distinct clusters with started consumers for tenant/namespace/subscription, so callers don't have to know a worker's configured cluster ahead of time. Uses ConsumerIDRegistry rather than ConsumerReadyRegistry so a cluster is still discovered while its consumers are down/reconnecting, not just while healthy.

consumer_gauges(cluster_name, tenant, namespace, subscription)

Desired, active, and ready consumer counts for cluster/tenant/namespace/subscription - desired comes from the consumer manager, active/ready from local registries

consumers()

List all the consumers on default cluster

consumers(cluster_name)

List all the consumers for cluster

consumers(cluster_name, topic_name)

List all the consumers for cluster on a specific topic

consumers(cluster_name, topic_name, subscription)

List all the consumers for cluster on a specific topic with specific subscription

produce(topic_name, payload, message_opts \\ [], producer_opts \\ [])

@spec produce(String.t(), binary(), keyword(), keyword()) ::
  {:ok,
   %PulsarEx.Proto.MessageIdData{
     __unknown_fields__: term(),
     ack_set: term(),
     batch_index: term(),
     batch_size: term(),
     entryId: term(),
     ledgerId: term(),
     partition: term()
   }}
  | {:error, any()}

Publish a message to pulsar, create a pool of producers if not yet created

ready_consumers(cluster_name, tenant, namespace, subscription)

Count consumers that have reached the :READY state for cluster/tenant/namespace/subscription

start_consumer(topic_name, subscription, module, opts)

@spec start_consumer(topic_name(), subscription(), module(), keyword()) ::
  :ok | {:error, any()}

Start consumer from a fully quantified topic

start_consumer(tenant, namespace, topic, subscription, module, opts)

@spec start_consumer(
  tenant(),
  namespace(),
  topic(),
  subscription(),
  module(),
  keyword()
) ::
  :ok | {:error, any()}

Start consumers with topic pattern

stop_consumer(topic, subscription)

@spec stop_consumer(topic(), subscription()) :: :ok

Stop consumers with fully quantified topic

stop_consumer(tenant, namespace, topic, subscription)

@spec stop_consumer(tenant(), namespace(), topic(), subscription()) :: :ok

Stop consumers with topic pattern