pulsar-rs

The pulsar-rs client is a pure Rust implementation for Apache Pulsar with the following capabilities:

• Connection Types: Supports URL-based connections using pulsar:// and pulsar+ssl:// with DNS lookup.
• Consumer Types: Supports multi-topic consumers using either a regex or a specific list of topics.
• Security: Supports TLS connections.
• Async Runtime: Configurable executor supporting Tokio or async-std.
• Resilience: Automatic reconnection with exponential backoff.
• Performance: Supports message batching and various compression algorithms (LZ4, zlib, zstd, or Snappy).
• Observability: Telemetry integration via the tracing crate.

Note: Compression algorithms and telemetry can be enabled or disabled via Cargo features.

To use pulsar-rs in your Rust project, add the following dependencies to your Cargo.toml. Note that the client is compatible with both tokio and async-std executors.

futures = "0.3"
pulsar = "6.3.1"
tokio = "1.0"

To run the pulsar-rs examples, you must have a running Apache Pulsar instance. You can spin up a standalone instance using Docker with the following command. This command maps the necessary ports (6650 for Pulsar protocol and 8080 for HTTP) and uses persistent volumes for data and configuration.

docker run -it \
  -p 6650:6650 \
  -p 8080:8080 \
  --mount source=pulsardata,target=/pulsar/data \
  --mount source=pulsarconf,target=/pulsar/conf \
  apachepulsar/pulsar:2.10.0 \
  bin/pulsar standalone

The Executor trait provides a unified interface for task spawning and time-based operations, allowing pulsar-rs to be compatible with both Tokio and async-std runtimes. By using this abstraction, the library can perform asynchronous tasks, spawn blocking operations, and handle intervals or delays without being hard-coded to a specific runtime.

To use a specific runtime, you must enable the corresponding Cargo features:

• For Tokio: tokio-runtime (or specific rustls variants like tokio-rustls-runtime-aws-lc-rs or tokio-rustls-runtime-ring).
• For async-std: async-std-runtime (or specific rustls variants like async-std-rustls-runtime-aws-lc-rs or async-std-rustls-runtime-ring).

If no runtime feature is enabled, attempting to use the executor will result in an unimplemented! panic.

You can optimize throughput by enabling message batching in ProducerOptions. The following options are available:

• batch_size: The maximum number of messages to include in a single batch.
• batch_byte_size: The maximum size in bytes for a single batch.
• batch_timeout: The maximum time to wait before sending a partial batch.

// Example of configuring batching in ProducerOptions
let options = producer::ProducerOptions {
    batch_size: Some(100),
    batch_byte_size: Some(1024 * 1024),
    batch_timeout: Some(std::time::Duration::from_millis(10)),
    ..Default::default()
};

The ConsumerBuilder supports two modes of topic discovery:

1. Explicit Topics: When using with_topic or with_topics, the builder performs a lookup for each topic. If multiple topics are provided, it creates a MultiTopicConsumer which manages several TopicConsumer instances internally.
2. Regex Matching: When using with_topic_regex, the builder creates a MultiTopicConsumer that periodically refreshes its list of topics based on the provided with_topic_refresh interval (defaulting to 30 seconds). It uses the with_lookup_namespace setting to scope the regex search.

Constraints:

• Consumer ID: You cannot specify a consumer_id if you are connecting to multiple topics or partitioned topics.
• Readers: You cannot create a Reader using into_reader() if more than one topic partition is identified. Readers are limited to a single topic partition.

Routing policies determine which partition a message is sent to:

• RoundRobin: Distributes messages across partitions. If a partition_key is provided, messages with the same key are routed to the same partition to maintain order.
• Single: Sends all messages to the same partition.
• Custom: Allows implementing the CustomRoutingPolicy trait to define your own logic based on the message and the number of available producers.

The proto module contains the generated Rust code from Pulsar's protobuf definitions. It exposes BaseCommand and Metadata (aliased from MessageMetadata).

• BaseCommand: The core protobuf message representing any Pulsar command (e.g., Connect, Subscribe, Send, Ack, Producer).
• Metadata: The metadata associated with a message payload.
• client_version(): A helper function that returns the current client version string in the format pulsar-rs-v<VERSION>.

pub mod proto {
    // ... includes pulsar.proto.rs
    pub fn client_version() -> String {
        format!("{}-v{}", "pulsar-rs", env!("CARGO_PKG_VERSION"))
    }
}

The client uses two helper traits to handle data conversion between application types and Pulsar payloads:

SerializeMessage

Used by producers to convert input into a producer::Message. Supported types include:

• producer::Message (identity)
• () (creates an empty message)
• &[u8], Vec<u8>, &str, String, &String (converts to byte payload)

DeserializeMessage

Used by consumers to convert a Payload into an output type. Supported types include:

• Vec<u8>: Returns the raw bytes.
• String: Returns Result<String, FromUtf8Error>.