🥮 mooncakes.io
    Sign in

    moonpulsar

    Apache Pulsar binary protocol client for MoonBit: connection management, producer (sync/async send, batching) and consumer (subscription modes, ack/nack, flow control)

    pulsar
    messaging
    producer
    consumer
    Download zip
    Author
    Version
    0.1.1
    License
    Apache-2.0
    Repository
    Last updated
    24 days ago
    Downloads
    18

    #moonpulsar

    English | 简体中文

    Apache Pulsar binary protocol client for MoonBit, targeting the native backend. Supports producers and consumers, batching, partitioned topics, readers, transactions, schemas, authentication and message encryption. See the capability matrix for supported behavior and verification limits.

    Source repository | Documentation

    #Requirements

    #Build environment

    • Use the native backend; JS and Wasm are not supported by this client. CI runs on Linux (Ubuntu 24.04). macOS has been tested locally; Windows is not verified.
    • Install the MoonBit toolchain and a C compiler with the system SDK/libc development files. The locally verified toolchain is moon 0.1.20260920 / moonc 0.10.14+7d59c7ec9; this is a tested version, not a declared minimum. CI installs the current toolchain.
    • Building the library does not require OpenSSL or zlib development headers. Run moon update to fetch MoonBit dependencies.

    #Runtime dependencies

    Requirements depend on the features your application uses:

    FeatureLinuxmacOS
    Plain TCP (pulsar://), compression, RSA message encryptionNo OpenSSL or system zlib runtime neededSame
    Ordinary TLS (pulsar+ssl://, including a custom CA)moonbitlang/async@0.22.1 loads libssl.so.3, with fallbacks to libssl.so.1.1 and libssl.soUses the system TLS library through moonbitlang/async@0.22.1
    TLS with a client certificate or explicit protocol/cipher settingsOpenSSL 3: libssl.so.3 and libcrypto.so.3Homebrew openssl@3
    ECIES message encryptionOpenSSL 3: libcrypto.so.3Homebrew openssl@3

    Linux libraries must be discoverable by the system dynamic loader. The macOS OpenSSL 3 adapters look under /opt/homebrew/opt/openssl@3/lib or /usr/local/opt/openssl@3/lib. These shared libraries are needed on the machine running the application, not just the build machine.

    #Tests and examples

    • moon test --target native includes TLS and ECIES tests, so it needs the TLS libraries above, including OpenSSL 3. It uses in-process mock brokers: no external broker or Docker is required.
    • Examples need a reachable Pulsar broker. The quickstart below uses Docker to start Pulsar 4.2.4 locally. Broker CI covers 4.2.4 and 3.3.9; these are verified versions, not a complete compatibility range.
    • Additional ABI and live interoperability checks have separate dependencies; see Development and verification.

    #Installation

    These instructions target 0.1.1. Check registry availability before adding the dependency:

    moon view pangbit/moonpulsar --versions --json

    If 0.1.1 is unavailable, or to run repository examples and tests, use the source workspace:

    git clone https://github.com/pangbit/moonpulsar.git
cd moonpulsar
moon update
moon check --target native
moon test --target native

    To run a roundtrip, use a disposable local Pulsar 4.2.4 broker. In a separate terminal:

    docker run --rm --name moonpulsar-demo -p 6650:6650 -p 8080:8080 \
  apachepulsar/pulsar:4.2.4 bin/pulsar standalone -a 127.0.0.1

    After the broker is ready, run from the repository root:

    moon run examples/roundtrip --target native

    A successful run prints roundtrip OK. Stop the disposable container after use with docker stop moonpulsar-demo.

    Once registry availability is confirmed, a separate executable can declare "pangbit/moonpulsar@0.1.1" and "moonbitlang/async@0.22.1" in moon.mod, set preferred_target = "native", and use this moon.pkg:

    import {
  "pangbit/moonpulsar" @pulsar,
  "moonbitlang/async",
}
supported_targets = "+native"
pkgtype(kind: "executable")

    #Usage

    ///|
    async fn main {
      @async.with_task_group(async fn(group) {
        let client = @pulsar.Client::connect(group, "pulsar://127.0.0.1:6650")
        let topic = "persistent://public/default/my-topic"
        let consumer = client.create_consumer(topic, "my-subscription")
        let producer = client.create_producer(topic)
        let receipt = producer.send(@pulsar.ProducerMessage::new(b"hello"))
        println("sent entry=\{receipt.message_id.entry_id}")
        let message = consumer.receive()
        message.ack()
        consumer.close()
        producer.close()
        client.close()
      })
    }

    Background tasks (connection reader, keepalive) are tied to the task group passed to Client::connect, following moonbitlang/async structured concurrency. The consumer is created before sending so a new subscription can receive the message.

    The @pulsar alias in this snippet is declared in the moon.pkg above. See examples/roundtrip for a larger executable.

    #Documentation

    #License

    Apache-2.0. The vendored Pulsar protocol definition and generated bindings carry Apache Pulsar attribution in NOTICE.

    Authentication

    source
    pub(open) trait Authentication {
      fn method_name(Self) -> String
      fn auth_data(Self) -> Bytes
    }

    Pluggable authentication for the CONNECT handshake.

    PulsarError

    source
    pub(all) suberror PulsarError {
      ConnectionClosed
      ConnectFailed(String)
      ServerError(code~ : Int, message~ : String)
      UnexpectedResponse(String)
      LookupFailed(String)
      TooManyRedirects
      ProducerClosed
      ProducerQueueFull
      ClientMemoryFull
      ConsumerClosed
      ChecksumMismatch
      InvalidFrame(String)
      SchemaIncompatible(String)
      RequestTimeout
      AuthenticationFailed(String)
    } derive(
    Debug    )

    Errors raised by the Pulsar client.
    impl Show for PulsarError

    PulsarError::output

    source
    fn PulsarError::output(self : PulsarError, logger : &Logger) -> Unit

    PulsarError::to_repr

    source

    PulsarError::to_string

    source
    fn PulsarError::to_string(self : PulsarError) -> String

    AckDispatcher

    source
    type AckDispatcher

    AckGroupingOptions

    source
    pub struct AckGroupingOptions {
      // private fields
    }

    AckGroupingOptions::new

    source
    fn AckGroupingOptions::new(max_size? : Int, max_time_ms? : Int) -> AckGroupingOptions raise PulsarError

    Admin

    source
    pub struct Admin {
      // private fields
    }

    The Pulsar admin REST API (usually http://host:8080).

    Admin::create_partitioned_topic

    source
    async fn Admin::create_partitioned_topic(self : Admin, topic : String, partitions : Int) -> Unit

    Create a partitioned topic with the given number of partitions.

    Admin::create_subscription

    source
    async fn Admin::create_subscription(self : Admin, topic : String, subscription : String, initial_position? : SubscriptionInitialPosition) -> Unit

    Create a durable subscription at the earliest or latest position.

    Admin::create_topic

    source
    async fn Admin::create_topic(self : Admin, topic : String) -> Unit

    Create a non-partitioned topic.

    Admin::delete_partitioned_topic

    source
    async fn Admin::delete_partitioned_topic(self : Admin, topic : String, force? : Bool) -> Unit

    Delete a partitioned topic.

    Admin::delete_subscription

    source
    async fn Admin::delete_subscription(self : Admin, topic : String, subscription : String, force? : Bool) -> Unit

    Delete a topic subscription.

    Admin::delete_topic

    source
    async fn Admin::delete_topic(self : Admin, topic : String, force? : Bool) -> Unit

    Delete a non-partitioned topic.

    Admin::expire_messages

    source
    async fn Admin::expire_messages(self : Admin, topic : String, subscription : String, older_than_seconds : Int) -> Unit

    Expire messages older than the given number of seconds.

    Admin::list_partitioned_topics

    source
    async fn Admin::list_partitioned_topics(self : Admin, tenant : String, ns : String, domain? : String) -> Array[String]

    List partitioned topics of a namespace.

    Admin::list_subscriptions

    source
    async fn Admin::list_subscriptions(self : Admin, topic : String) -> Array[String]

    List subscription names for a topic.

    Admin::list_topics

    source
    async fn Admin::list_topics(self : Admin, tenant : String, ns : String, domain? : String) -> Array[String]

    List topics of a namespace.

    Admin::new

    source
    fn Admin::new(base_url : String, token? : String) -> Admin

    Create an admin client, e.g. Admin::new("http://localhost:8080").

    Admin::partition_count

    source
    async fn Admin::partition_count(self : Admin, topic : String) -> Int

    Return the partition count, or zero for a non-partitioned topic.

    Admin::partitioned_topic_stats

    source
    async fn Admin::partitioned_topic_stats(self : Admin, topic : String, per_partition? : Bool) -> Json

    Partitioned topic statistics as raw JSON. Include per-partition detail when per_partition is true.

    Admin::remove_topic_property

    source
    async fn Admin::remove_topic_property(self : Admin, topic : String, key : String) -> Unit

    Remove one user-defined topic property.

    Admin::skip_all_messages

    source
    async fn Admin::skip_all_messages(self : Admin, topic : String, subscription : String) -> Unit

    Clear all pending messages for a subscription.

    Admin::skip_messages

    source
    async fn Admin::skip_messages(self : Admin, topic : String, subscription : String, count : Int) -> Unit

    Advance a subscription by count messages.

    Admin::topic_properties

    source
    async fn Admin::topic_properties(self : Admin, topic : String) -> Map[String, String]

    Read user-defined properties of a topic.

    Admin::topic_stats

    source
    async fn Admin::topic_stats(self : Admin, topic : String) -> Json

    Topic statistics as raw JSON (see Pulsar's TopicStats schema).

    Admin::update_partitioned_topic

    source
    async fn Admin::update_partitioned_topic(self : Admin, topic : String, partitions : Int) -> Unit

    Increase the number of partitions of an existing topic.

    Admin::update_topic_properties

    source
    async fn Admin::update_topic_properties(self : Admin, topic : String, properties : Map[String, String]) -> Unit

    Replace user-defined properties of a topic.

    AthenzZtsOptions

    source
    pub(all) struct AthenzZtsOptions {
      zts_url : String
      provider_domain : String
      tenant_domain : String
      tenant_service : String
      private_key_file : String
      certificate_file : String?
      ca_pem_file : String?
      key_id : String
      role : String
      principal_header : String
      min_expiry_secs : Int
      max_expiry_secs : Int
      refresh_before_ms : Int
      max_retries : Int
      retry_delay_ms : Int
    }

    Settings for exchanging a service identity for an Athenz role token. zts_url is the ZTS server base URL, without /zts/v1.

    AthenzZtsOptions::new

    source
    fn AthenzZtsOptions::new(zts_url : String, provider_domain : String, tenant_domain : String, tenant_service : String, private_key_file : String, certificate_file? : String, ca_pem_file? : String, key_id? : String, role? : String, principal_header? : String, min_expiry_secs? : Int, max_expiry_secs? : Int, refresh_before_ms? : Int, max_retries? : Int, retry_delay_ms? : Int) -> AthenzZtsOptions raise PulsarError

    AuthData

    source
    pub(all) struct AuthData {
      method_name : String
      data : Bytes
    } derive(
    Debug    )

    Authentication data attached to the CONNECT command.

    AuthData::to_repr

    source

    BasicAuth

    source
    pub(all) struct BasicAuth {
      username : String
      password : String
    }

    Username/password authentication for a broker's basic auth provider.

    BasicAuth::auth_data

    source
    fn BasicAuth::auth_data(self : BasicAuth) -> Bytes

    BasicAuth::method_name

    source
    fn BasicAuth::method_name(_ : BasicAuth) -> String

    BasicAuth::new

    source
    fn BasicAuth::new(username : String, password : String) -> BasicAuth raise PulsarError

    BatchAckTracker

    source
    type BatchAckTracker

    BatchingConfig

    source
    pub(all) struct BatchingConfig {
      max_messages : Int
      max_bytes : Int
      max_delay_ms : Int
    } derive(
    Debug    )

    Batching configuration for a producer.

    BatchingConfig::default

    source

    BatchingConfig::to_repr

    source

    ChunkAssembler

    source
    type ChunkAssembler

    ChunkAssemblyPolicy

    source
    pub struct ChunkAssemblyPolicy {
      // private fields
    }

    Limits for incomplete chunked messages buffered by a reader or consumer.

    ChunkAssemblyPolicy::new

    source
    fn ChunkAssemblyPolicy::new(expiry_ms? : Int, max_pending_messages? : Int, auto_ack_incomplete? : Bool) -> ChunkAssemblyPolicy raise PulsarError

    Client

    source
    pub struct Client {
      // private fields
    }

    A Pulsar client. Background tasks of every connection it opens run in the task group passed to Client::connect.

    Client::close

    source
    fn Client::close(self : Client) -> Unit

    Close all connections opened by this client.

    Client::connect

    source
    async fn Client::connect(group : 
    TaskGroup    [Unit], service_url : String, auth? : &Authentication) -> Client

    Connect to a Pulsar cluster, e.g. Client::connect(group, "pulsar://localhost:6650"). Connectivity and the protocol handshake are verified eagerly.

    Client::connect_tls_insecure

    source
    async fn Client::connect_tls_insecure(group : 
    TaskGroup    [Unit], service_url : String, auth? : &Authentication) -> Client

    Connect with TLS certificate verification disabled (for testing against self-signed broker certificates).

    Client::connect_tls_with_ca

    source
    async fn Client::connect_tls_with_ca(group : 
    TaskGroup    [Unit], service_url : String, ca_pem_file : String, auth? : &Authentication) -> Client

    Connect to a TLS broker using a PEM file as the trusted root. The same trust root is used for topic lookup redirects and reconnects.

    Client::connect_with_options

    source
    async fn Client::connect_with_options(group : 
    TaskGroup    [Unit], options : ClientOptions) -> Client

    Connect using settings shared by all broker connections, including redirected lookups and reconnects.

    Client::create_consumer

    source
    async fn Client::create_consumer(self : Client, topic : String, subscription : String, sub_type? : SubscriptionType, key_shared_policy? : KeySharedPolicy, consumer_name? : String, initial_position? : SubscriptionInitialPosition, priority_level? : Int, durable? : Bool, read_compacted? : Bool, properties? : Map[String, String], subscription_properties? : Map[String, String], receiver_queue_size? : Int, ack_timeout_ms? : Int, nack_backoff? : NackBackoff, dead_letter_policy? : DeadLetterPolicy, schema? : SchemaInfo, message_crypto? : MessageCrypto, decryption_failure_action? : DecryptionFailureAction, encryption_key_names? : Array[String], partition_refresh_ms? : Int, chunk_policy? : ChunkAssemblyPolicy, retry_topic? : String, enable_batch_index_ack? : Bool, ack_grouping? : AckGroupingOptions, auto_scaled_receiver_queue? : Bool, interceptors? : Array[ConsumerInterceptor]) -> Consumer

    Create a consumer subscribed to topic.

    Client::create_multi_topic_consumer

    source
    async fn Client::create_multi_topic_consumer(self : Client, topics : Array[String], subscription : String, sub_type? : SubscriptionType, key_shared_policy? : KeySharedPolicy, consumer_name? : String, initial_position? : SubscriptionInitialPosition, priority_level? : Int, durable? : Bool, read_compacted? : Bool, properties? : Map[String, String], subscription_properties? : Map[String, String], receiver_queue_size? : Int, ack_timeout_ms? : Int, nack_backoff? : NackBackoff, dead_letter_policy? : DeadLetterPolicy, schema? : SchemaInfo, message_crypto? : MessageCrypto, decryption_failure_action? : DecryptionFailureAction, encryption_key_names? : Array[String], partition_refresh_ms? : Int, chunk_policy? : ChunkAssemblyPolicy, retry_topic? : String, enable_batch_index_ack? : Bool, ack_grouping? : AckGroupingOptions, auto_scaled_receiver_queue? : Bool, interceptors? : Array[ConsumerInterceptor]) -> Consumer

    Subscribe to an explicit list of topics with one logical consumer. Received messages retain their source topic in Message::topic.

    Client::create_pattern_consumer

    source
    async fn Client::create_pattern_consumer(self : Client, ns : String, pattern : String, subscription : String, refresh_interval_ms? : Int, sub_type? : SubscriptionType, key_shared_policy? : KeySharedPolicy, consumer_name? : String, initial_position? : SubscriptionInitialPosition, priority_level? : Int, durable? : Bool, read_compacted? : Bool, properties? : Map[String, String], subscription_properties? : Map[String, String], receiver_queue_size? : Int, ack_timeout_ms? : Int, nack_backoff? : NackBackoff, dead_letter_policy? : DeadLetterPolicy, schema? : SchemaInfo, message_crypto? : MessageCrypto, decryption_failure_action? : DecryptionFailureAction, encryption_key_names? : Array[String], partition_refresh_ms? : Int, chunk_policy? : ChunkAssemblyPolicy, retry_topic? : String, enable_batch_index_ack? : Bool, ack_grouping? : AckGroupingOptions, auto_scaled_receiver_queue? : Bool, interceptors? : Array[ConsumerInterceptor]) -> Consumer

    Subscribe to topics matching a regular expression in tenant/namespace. New topics are discovered at the configured polling interval.

    Client::create_producer

    source
    async fn Client::create_producer(self : Client, topic : String, name? : String, properties? : Map[String, String], access_mode? : ProducerAccessMode, batching? : BatchingConfig, compression? : CompressionCodec, message_crypto? : MessageCrypto, encryption_key_names? : Array[String], chunk_size? : Int, automatic_chunking? : Bool, schema? : SchemaInfo, partition_refresh_ms? : Int, routing? : ProducerRoutingMode, max_pending_messages? : Int, block_if_queue_full? : Bool, send_timeout_ms? : Int, interceptors? : Array[ProducerInterceptor]) -> Producer

    Create a producer for topic. The topic is looked up and the producer is attached to the owning broker.

    Client::create_reader

    source
    async fn Client::create_reader(self : Client, topic : String, start_message_id? : MessageId, receiver_queue_size? : Int, read_compacted? : Bool, partition_refresh_ms? : Int, include_start_message_id? : Bool, name? : String, properties? : Map[String, String], subscription_name? : String, subscription_role_prefix? : String, chunk_policy? : ChunkAssemblyPolicy, schema? : SchemaInfo, message_crypto? : MessageCrypto, decryption_failure_action? : DecryptionFailureAction) -> Reader

    Create a reader over topic starting after start_message_id (default: latest). Readers are non-durable and do not ack. Partitioned topics are aggregated over one reader per partition.

    Client::create_retry_consumer

    source
    async fn Client::create_retry_consumer(self : Client, topic : String, subscription : String, retry_topic : String, dead_letter_policy : DeadLetterPolicy, sub_type? : SubscriptionType, key_shared_policy? : KeySharedPolicy, consumer_name? : String, initial_position? : SubscriptionInitialPosition, priority_level? : Int, durable? : Bool, read_compacted? : Bool, properties? : Map[String, String], subscription_properties? : Map[String, String], receiver_queue_size? : Int, ack_timeout_ms? : Int, nack_backoff? : NackBackoff, schema? : SchemaInfo, message_crypto? : MessageCrypto, decryption_failure_action? : DecryptionFailureAction, encryption_key_names? : Array[String], partition_refresh_ms? : Int, chunk_policy? : ChunkAssemblyPolicy, enable_batch_index_ack? : Bool, ack_grouping? : AckGroupingOptions, auto_scaled_receiver_queue? : Bool, interceptors? : Array[ConsumerInterceptor]) -> Consumer

    Consume a source and its retry-letter topic as one stream. Call Consumer::reconsume_later to publish a failed message with a delay; retries beyond the dead-letter limit are sent to the dead-letter topic.

    Client::create_table_view

    source
    async fn Client::create_table_view(self : Client, topic : String, partition_refresh_ms? : Int, schema? : SchemaInfo, message_crypto? : MessageCrypto) -> TableView

    Build a live raw-bytes view of a compacted topic from its earliest message. Messages without keys are ignored; an empty payload removes the key. Initial last-message queries must succeed before the view is returned.

    Client::create_typed_table_view

    source
    async fn[T] Client::create_typed_table_view(self : Client, topic : String, codec : SchemaCodec[T], partition_refresh_ms? : Int, message_crypto? : MessageCrypto) -> TypedTableView[T]

    Create a typed view after the broker accepts the declared schema and every entry in the initial snapshot decodes successfully.

    Client::decode_avro

    source
    async fn[T] Client::decode_avro(self : Client, message : Message, reader : SchemaCodec[T]) -> T

    Decode a message using its broker-published writer schema version and a typed reader codec. The parsed writer schema is cached per topic/version.

    Client::get_schema

    source
    async fn Client::get_schema(self : Client, topic : String, version? : Bytes) -> VersionedSchema

    Retrieve the topic's latest schema, or a specific version. Broker errors such as an unknown version or incompatible schema are returned explicitly.

    Client::partitions_of

    source
    async fn Client::partitions_of(self : Client, topic : String) -> Int

    Number of partitions of a topic (0 or 1 for non-partitioned topics).

    Client::topics_of_namespace

    source
    async fn Client::topics_of_namespace(self : Client, ns : String, pattern? : String, mode? : NamespaceTopicMode) -> Array[String]

    List topics in tenant/namespace, optionally filtered by a MoonBit regular expression.

    Client::transaction_coordinator

    source
    async fn Client::transaction_coordinator(self : Client, tc_id : UInt64) -> TransactionCoordinator

    Connect to the broker owning transaction coordinator tc_id. The coordinator uses a connection managed by this client.

    ClientEvent

    source
    pub(all) enum ClientEvent {
      SendCompleted(String, ProducerMessage, Result[SendReceipt, Error])
      MessageReceived(Message)
      ReceiveFailed(String, Error)
      Acknowledged(String, MessageId, Result[Unit, Error])
      NegativelyAcknowledged(String, MessageId, Result[Unit, Error])
      ProducerReconnecting(String)
      ProducerReconnected(String, Result[Unit, Error])
      ConsumerReconnecting(String)
      ConsumerReconnected(String, Result[Unit, Error])
    }

    ClientMemoryLease

    source
    type ClientMemoryLease

    ClientMemoryLimiter

    source
    type ClientMemoryLimiter

    ClientMetrics

    source
    pub struct ClientMetrics {
      // private fields
    }

    ClientMetrics::new

    source

    ClientMetrics::prometheus_text

    source
    fn ClientMetrics::prometheus_text(self : ClientMetrics) -> String

    Export bounded-label counters and operation-duration histograms.

    ClientMetrics::record

    source
    fn ClientMetrics::record(self : ClientMetrics, event : ClientEvent) -> Unit

    ClientMetrics::record_span

    source
    fn ClientMetrics::record_span(self : ClientMetrics, event : ClientSpanEvent) -> Unit

    Record a span's completion in fixed millisecond buckets.

    ClientMetrics::snapshot

    source

    ClientMetricsSnapshot

    source
    pub(all) struct ClientMetricsSnapshot {
      sends : Int64
      send_failures : Int64
      received : Int64
      receive_failures : Int64
      acknowledgments : Int64
      ack_failures : Int64
      negative_acknowledgments : Int64
      negative_ack_failures : Int64
      producer_reconnects : Int64
      consumer_reconnects : Int64
      reconnect_failures : Int64
    } derive(Eq, 
    Debug    )

    ClientMetricsSnapshot::equal

    source

    ClientMetricsSnapshot::not_equal

    source

    ClientMetricsSnapshot::to_repr

    source

    ClientOperation

    source
    pub(all) enum ClientOperation {
      Send
      Receive
      Ack
      Reconnect
    }

    Operations with optional span and latency reporting.

    ClientOptions

    source
    pub struct ClientOptions {
      // private fields
    }

    Options for a Pulsar client. Durations are milliseconds.

    ClientOptions::new

    source
    fn ClientOptions::new(service_url : String, auth? : &Authentication, auth_provider? : async () -> AuthData, connection_timeout_ms? : Int, operation_timeout_ms? : Int, keepalive_interval_ms? : Int, tls_insecure? : Bool, ca_pem_file? : String, tls_identity? : TlsIdentity, tls_identity_provider? : async () -> TlsIdentity, tls_min_version? : TlsVersion, tls_max_version? : TlsVersion, tls_cipher_list? : String, tls13_ciphersuites? : String, listener_name? : String, lookup_properties? : Map[String, String], reconnect_policy? : RetryPolicy, max_connections_per_broker? : Int, connection_idle_timeout_ms? : Int, max_memory_bytes? : Int, on_event? : (ClientEvent) -> Unit, on_span? : (ClientSpanEvent) -> Unit) -> ClientOptions raise PulsarError

    ClientSpan

    source
    pub(all) struct ClientSpan {
      id : Int64
      operation : ClientOperation
      topic : String
      message_id : MessageId?
      started_at_ms : Int64
    }

    A span's stable correlation ID and start time. message_id is present when the operation already knows the broker message position.

    ClientSpanEvent

    source
    pub(all) enum ClientSpanEvent {
      Started(ClientSpan)
      Finished(ClientSpan, Int64, Bool, String?)
    }

    CompressionCodec

    source
    pub(all) enum CompressionCodec {
      None
      Lz4
      Zlib
      Zstd
      Snappy
    } derive(Eq, 
    Debug    )

    Compression applied to message payloads.

    CompressionCodec::equal

    source

    CompressionCodec::not_equal

    source
    fn CompressionCodec::not_equal(x : CompressionCodec, y : CompressionCodec) -> Bool

    CompressionCodec::to_repr

    source

    Connection

    source
    pub struct Connection {
      // private fields
    }

    A single connection to a Pulsar broker speaking the binary protocol.

    Connection::close

    source
    fn Connection::close(self : Connection) -> Unit

    Close the connection and fail all pending requests.

    Connection::connect

    source
    async fn Connection::connect(group : 
    TaskGroup    [Unit], addr : String, auth? : AuthData, auth_provider? : () -> AuthData, tls? : Bool, tls_insecure? : Bool, tls_trust? : 
    TrustedRoot    , tls_identity? : TlsIdentity, tls_identity_provider? : async () -> TlsIdentity, tls_min_version? : TlsVersion, tls_max_version? : TlsVersion, tls_cipher_list? : String, tls13_ciphersuites? : String, keepalive_interval_ms? : Int, timeout_ms? : Int, connection_timeout_ms? : Int, async_auth_provider? : async () -> AuthData) -> Connection

    Connect to addr ("host:port"), perform the protocol handshake and start the background reader/keepalive tasks in group.

    Connection::is_closed

    source
    fn Connection::is_closed(self : Connection) -> Bool

    Whether the connection is still usable.

    Connection::max_message_size

    source
    fn Connection::max_message_size(self : Connection) -> Int?

    Maximum frame payload announced by the broker, when available.

    Connection::next_request_id

    source
    fn Connection::next_request_id(self : Connection) -> UInt64

    Allocate a fresh request id for this connection.

    Connection::on_close

    source
    fn Connection::on_close(self : Connection, handler : (Error?) -> Unit) -> Unit

    Register a handler invoked once when the connection breaks.

    Connection::remote_protocol_version

    source
    fn Connection::remote_protocol_version(self : Connection) -> Int

    Protocol version reported by the broker during the handshake.

    Connection::request

    source

    Send a request command and wait for the correlated response. CommandError responses are raised as ServerError.

    Connection::send_command

    source
    async fn Connection::send_command(self : Connection, cmd : 
    BaseCommand    ) -> Unit

    Send a fire-and-forget command frame (ACK, FLOW, PONG, ...).

    Connection::send_payload_frame

    source
    async fn Connection::send_payload_frame(self : Connection, cmd : 
    BaseCommand    , metadata : 
    MessageMetadata    , payload : Bytes) -> Unit

    Send a frame with attached message metadata and payload (SEND).

    Connection::server_version

    source
    fn Connection::server_version(self : Connection) -> String

    Server version reported by the broker during the handshake.

    ConnectionManager

    source
    pub struct ConnectionManager {
      // private fields
    }

    Caches up to the configured number of connections per broker address.

    ConnectionManager::close

    source
    fn ConnectionManager::close(self : ConnectionManager) -> Unit

    Close all cached connections; further get calls fail.

    ConnectionManager::get

    source
    async fn ConnectionManager::get(self : ConnectionManager, addr : String) -> Connection

    Get an open connection to addr, connecting on first use or after the cached connection broke.

    ConnectionManager::is_closed

    source
    fn ConnectionManager::is_closed(self : ConnectionManager) -> Bool

    Whether the manager has been closed.

    ConnectionManager::new

    source
    fn ConnectionManager::new(group : 
    TaskGroup    [Unit], auth? : AuthData, auth_provider? : () -> AuthData, tls? : Bool, tls_insecure? : Bool, tls_trust? : 
    TrustedRoot    , tls_identity? : TlsIdentity, tls_identity_provider? : async () -> TlsIdentity, tls_min_version? : TlsVersion, tls_max_version? : TlsVersion, tls_cipher_list? : String, tls13_ciphersuites? : String, connection_timeout_ms? : Int, operation_timeout_ms? : Int, keepalive_interval_ms? : Int, listener_name? : String, lookup_properties? : Map[String, String], async_auth_provider? : async () -> AuthData, max_connections_per_broker? : Int, connection_idle_timeout_ms? : Int) -> ConnectionManager raise PulsarError

    ConnectionManager::uses_tls

    source
    fn ConnectionManager::uses_tls(self : ConnectionManager) -> Bool

    Whether connections opened by this manager use TLS.

    Consumer

    source
    pub struct Consumer {
      // private fields
    }

    A consumer for a topic, partitioned or not. Partitioned consumers merge the message streams of every partition.

    Consumer::ack

    source
    async fn Consumer::ack(self : Consumer, id : MessageId, await_response? : Bool) -> Unit

    Acknowledge a message individually.

    Consumer::ack_cumulative

    source
    async fn Consumer::ack_cumulative(self : Consumer, id : MessageId, await_response? : Bool) -> Unit

    Acknowledge cumulatively: this message and all before it.

    Consumer::ack_ids

    source
    async fn Consumer::ack_ids(self : Consumer, ids : Array[MessageId], await_response? : Bool) -> Unit

    Acknowledge multiple IDs belonging to a single-topic consumer.

    Consumer::ack_txn

    source
    async fn Consumer::ack_txn(self : Consumer, txn : Transaction, id : MessageId, await_response? : Bool) -> Unit

    Acknowledge a message as part of a transaction.

    Consumer::close

    source
    async fn Consumer::close(self : Consumer) -> Unit

    Close the consumer (all partitions).

    Consumer::is_closed

    source
    fn Consumer::is_closed(self : Consumer) -> Bool

    Consumer::last_message_ids

    source
    async fn Consumer::last_message_ids(self : Consumer) -> Map[String, MessageId]

    Return the broker's last message ID for each subscribed topic.

    Consumer::nack

    source
    async fn Consumer::nack(self : Consumer, id : MessageId) -> Unit

    Negatively acknowledge a message, triggering redelivery.

    Consumer::partitioned

    source
    fn Consumer::partitioned(group : 
    TaskGroup    [Unit], consumers : Array[Consumer]) -> Consumer

    Create a consumer aggregating all partitions of a partitioned topic.

    Consumer::receive

    source
    async fn Consumer::receive(self : Consumer) -> Message

    Receive the next message, blocking until one is available.

    Consumer::reconsume_later

    source
    async fn Consumer::reconsume_later(self : Consumer, message : Message, delay_ms? : Int, custom_properties? : Map[String, String]) -> Unit

    Publish a message for delayed reconsumption. Its source is acknowledged only after the retry or dead-letter producer receives a broker receipt. Once the retry count exceeds the dead-letter delivery limit, the message goes directly to the dead-letter topic.

    Consumer::seek

    source
    async fn Consumer::seek(self : Consumer, position : SeekPosition) -> Unit

    Reset a subscription cursor. Timestamp, earliest, and latest positions apply to every source. A regular message id must select exactly one source.

    Consumer::subscription

    source
    fn Consumer::subscription(self : Consumer) -> String

    Consumer::topic

    source
    fn Consumer::topic(self : Consumer) -> String

    Consumer::topics

    source
    fn Consumer::topics(self : Consumer) -> Array[String]

    Current source topics of this consumer. Pattern subscriptions gain topics as discovery refreshes succeed.

    Consumer::try_receive

    source
    fn Consumer::try_receive(self : Consumer) -> Message? raise

    Non-blocking receive: returns None if no message is buffered.

    Consumer::unsubscribe

    source
    async fn Consumer::unsubscribe(self : Consumer, force? : Bool) -> Unit

    Remove the subscription from every source topic and close the consumer.

    ConsumerInterceptor

    source
    pub struct ConsumerInterceptor {
      // private fields
    }

    ConsumerInterceptor::new

    source
    fn ConsumerInterceptor::new(before_consume? : (Message) -> Unit, on_receive_error? : (Error) -> Unit, on_ack? : (MessageId, Result[Unit, Error]) -> Unit, on_nack? : (MessageId, Result[Unit, Error]) -> Unit) -> ConsumerInterceptor

    DeadLetterPolicy

    source
    pub struct DeadLetterPolicy {
      // private fields
    }

    Route a message to a dead-letter topic after max_deliveries application deliveries. For example, one delivery permits the initial attempt and routes the next redelivery.

    DeadLetterPolicy::max_deliveries

    source
    fn DeadLetterPolicy::max_deliveries(self : DeadLetterPolicy) -> UInt

    DeadLetterPolicy::new

    source
    fn DeadLetterPolicy::new(max_deliveries : UInt, topic : String) -> DeadLetterPolicy raise PulsarError

    DeadLetterPolicy::topic

    source
    fn DeadLetterPolicy::topic(self : DeadLetterPolicy) -> String

    DecryptionFailureAction

    source
    pub(all) enum DecryptionFailureAction {
      Fail
      Consume
      Discard
    }

    How a consumer or reader handles an encrypted message that cannot be decrypted. Consume delivers ciphertext only for non-batched messages.

    EncryptionPrivateKey

    source
    pub struct EncryptionPrivateKey {
      // private fields
    }

    EncryptionPrivateKey::from_pem

    source
    fn EncryptionPrivateKey::from_pem(pem : Bytes) -> EncryptionPrivateKey raise PulsarError

    Parse an unencrypted RSA or EC private key in PKCS#1, SEC1 or PKCS#8 PEM format.

    EncryptionPublicKey

    source
    pub struct EncryptionPublicKey {
      // private fields
    }

    EncryptionPublicKey::from_pem

    source
    fn EncryptionPublicKey::from_pem(pem : Bytes) -> EncryptionPublicKey raise PulsarError

    Parse an RSA or EC public key in SubjectPublicKeyInfo or PKCS#1 PEM format.

    IncomingFrame

    source
    pub(all) struct IncomingFrame {
      cmd : 
    BaseCommand
      metadata : 
    MessageMetadata    ?
      payload : Bytes?
      wire_size : Int
    } derive(
    Debug    )

    A frame received from the broker: the command plus, for MESSAGE frames, the message metadata and payload.

    IncomingFrame::to_repr

    source

    KeySharedPolicy

    source
    pub struct KeySharedPolicy {
      // private fields
    }

    Broker-side dispatch policy for a KeyShared subscription.

    KeySharedPolicy::auto_split

    source
    fn KeySharedPolicy::auto_split(allow_out_of_order_delivery? : Bool) -> KeySharedPolicy

    KeySharedPolicy::sticky

    source
    fn KeySharedPolicy::sticky(ranges : Array[(Int, Int)], allow_out_of_order_delivery? : Bool) -> KeySharedPolicy raise PulsarError

    Reserve non-overlapping inclusive hash ranges within 0..65535.

    Message

    source
    pub struct Message {
      publish_time : UInt64
      // private fields
    }

    A message received by a consumer.

    Message::ack

    source
    async fn Message::ack(self : Message, await_response? : Bool) -> Unit

    Acknowledge this message individually.

    Message::ack_txn

    source
    async fn Message::ack_txn(self : Message, txn : Transaction, await_response? : Bool) -> Unit

    Acknowledge this message as part of a transaction.

    Message::decode

    source
    fn[T] Message::decode(self : Message, codec : SchemaCodec[T]) -> T raise PulsarError

    Decode this message's payload with a schema codec. The original message remains available for metadata and acknowledgement.

    Message::decryption_failed

    source
    fn Message::decryption_failed(self : Message) -> Bool

    Whether the payload is encrypted ciphertext delivered by the Consume decryption failure action.

    Message::event_time

    source
    fn Message::event_time(self : Message) -> UInt64?

    Message::id

    source
    fn Message::id(self : Message) -> MessageId

    Message::key

    source
    fn Message::key(self : Message) -> String?

    Message::nack

    source
    async fn Message::nack(self : Message) -> Unit

    Negatively acknowledge this message, triggering redelivery.

    Message::ordering_key

    source
    fn Message::ordering_key(self : Message) -> Bytes?

    Message::payload

    source
    fn Message::payload(self : Message) -> Bytes

    Message::producer_name

    source
    fn Message::producer_name(self : Message) -> String

    Message::properties

    source
    fn Message::properties(self : Message) -> Map[String, String]

    Message::publish_time

    source
    fn Message::publish_time(self : Message) -> UInt64

    Message::redelivery_count

    source
    fn Message::redelivery_count(self : Message) -> UInt

    How many times the broker redelivered this message.

    Message::schema_version

    source
    fn Message::schema_version(self : Message) -> Bytes?

    Version of the schema used to publish this message, when supplied by the broker.

    Message::sequence_id

    source
    fn Message::sequence_id(self : Message) -> UInt64?

    Message::topic

    source
    fn Message::topic(self : Message) -> String

    MessageCrypto

    source
    pub struct MessageCrypto {
      // private fields
    }

    MessageCrypto::add_private_key

    source
    fn MessageCrypto::add_private_key(self : MessageCrypto, name : String, pem : Bytes) -> Unit raise PulsarError

    Replace a consumer key. Already received ciphertext retains its key name.

    MessageCrypto::add_public_key

    source
    fn MessageCrypto::add_public_key(self : MessageCrypto, name : String, pem : Bytes) -> Unit raise PulsarError

    Replace a producer key. Subsequent sends use the replacement immediately.

    MessageCrypto::decrypt

    source
    fn MessageCrypto::decrypt(self : MessageCrypto, metadata : 
    MessageMetadata    , payload : Bytes) -> Bytes raise PulsarError

    Decrypt an encrypted Pulsar payload. Missing keys and failed authentication return an error; callers must not deliver unauthenticated plaintext.

    MessageCrypto::encrypt

    source
    async fn MessageCrypto::encrypt(self : MessageCrypto, metadata : 
    MessageMetadata    , payload : Bytes, key_names : Array[String]) -> Bytes raise PulsarError

    Encrypt a wire payload and fill Pulsar's encryption metadata. The caller must compress first and send the returned bytes without further encoding.

    MessageCrypto::load_private_key_file

    source
    async fn MessageCrypto::load_private_key_file(self : MessageCrypto, name : String, path : String) -> Unit raise PulsarError

    Load a PEM private key from a file; call again after rotating the file.

    MessageCrypto::load_public_key_file

    source
    async fn MessageCrypto::load_public_key_file(self : MessageCrypto, name : String, path : String) -> Unit raise PulsarError

    Load a PEM public key from a file; call again after rotating the file.

    MessageCrypto::new

    source
    fn MessageCrypto::new(public_key_reader? : (String) -> Bytes?, private_key_reader? : (String) -> Bytes?) -> MessageCrypto

    Optional readers are called for every send or encrypted delivery. A reader takes precedence over keys added with add_public_key / add_private_key. Return None when the named key is unavailable; callbacks must be synchronous and must not raise.

    MessageId

    source
    pub(all) struct MessageId {
      ledger_id : UInt64
      entry_id : UInt64
      batch_index : Int
      partition : Int
    } derive(Compare, Eq, Hash, 
    Debug    )

    Identifies a persisted message in a topic.

    MessageId::compare

    source
    fn MessageId::compare(MessageId, MessageId) -> Int

    MessageId::earliest

    source
    fn MessageId::earliest() -> MessageId

    The earliest available position.

    MessageId::equal

    source
    fn MessageId::equal(MessageId, MessageId) -> Bool

    MessageId::hash

    source
    fn MessageId::hash(self : MessageId) -> Int

    MessageId::hash_combine

    source
    fn MessageId::hash_combine(MessageId, Hasher) -> Unit

    MessageId::latest

    source
    fn MessageId::latest() -> MessageId

    The latest available position.

    MessageId::not_equal

    source
    fn MessageId::not_equal(x : MessageId, y : MessageId) -> Bool

    MessageId::op_ge

    source
    fn MessageId::op_ge(x : MessageId, y : MessageId) -> Bool

    MessageId::op_gt

    source
    fn MessageId::op_gt(x : MessageId, y : MessageId) -> Bool

    MessageId::op_le

    source
    fn MessageId::op_le(x : MessageId, y : MessageId) -> Bool

    MessageId::op_lt

    source
    fn MessageId::op_lt(x : MessageId, y : MessageId) -> Bool

    MessageId::to_repr

    source

    NackBackoff

    source
    pub struct NackBackoff {
      // private fields
    }

    Exponential delay before redelivering negatively acknowledged messages.

    NackBackoff::default

    source
    fn NackBackoff::default() -> NackBackoff

    Default to one second initially, capped at ten minutes.

    NackBackoff::delay_ms

    source
    fn NackBackoff::delay_ms(self : NackBackoff, redelivery_count : UInt) -> Int

    NackBackoff::new

    source
    fn NackBackoff::new(initial_ms : Int, max_ms : Int) -> NackBackoff raise PulsarError

    NamespaceTopicMode

    source
    pub(all) enum NamespaceTopicMode {
      Persistent
      NonPersistent
      AllTopics
    } derive(Eq, 
    Debug    )

    Topic kinds included in namespace discovery.

    NamespaceTopicMode::equal

    source

    NamespaceTopicMode::not_equal

    source

    NamespaceTopicMode::to_repr

    source

    PendingRequest

    source
    type PendingRequest

    PoolConnection

    source
    type PoolConnection

    Producer

    source
    pub struct Producer {
      // private fields
    }

    A producer for a topic, partitioned or not. Partitioned producers route keyed messages by xxhash32(key) % partitions and keyless messages round-robin.

    Producer::close

    source
    async fn Producer::close(self : Producer) -> Unit

    Close the producer (all partitions), flushing pending batches first.

    Producer::flush

    source
    async fn Producer::flush(self : Producer) -> Unit

    Flush buffered batches on every partition.

    Producer::is_closed

    source
    fn Producer::is_closed(self : Producer) -> Bool

    Producer::last_sequence_id

    source
    fn Producer::last_sequence_id(self : Producer) -> UInt64?

    Highest confirmed sequence id across this producer's partitions.

    Producer::name

    source
    fn Producer::name(self : Producer) -> String

    The broker-assigned producer name (of the first partition).

    Producer::partition_count

    source
    fn Producer::partition_count(self : Producer) -> Int

    Number of producer partitions currently available for routing.

    Producer::partitioned

    source
    fn Producer::partitioned(partitions : Array[Producer], routing? : ProducerRoutingMode) -> Producer

    Create a producer on a partitioned topic from one producer per partition.

    Producer::send

    source
    async fn Producer::send(self : Producer, msg : ProducerMessage) -> SendReceipt

    Send a message and wait for the broker's receipt.

    Producer::send_async

    source
    async fn Producer::send_async(self : Producer, msg : ProducerMessage) -> ReceiptFuture

    Send a message without waiting for the receipt.

    Producer::send_txn

    source
    async fn Producer::send_txn(self : Producer, txn : Transaction, msg : ProducerMessage) -> SendReceipt

    Send a message as part of a transaction.

    Producer::topic

    source
    fn Producer::topic(self : Producer) -> String

    The topic this producer publishes to.

    ProducerAccessMode

    source
    pub(all) enum ProducerAccessMode {
      Shared
      Exclusive
      WaitForExclusive
      ExclusiveWithFencing
    } derive(Eq, 
    Debug    )

    How a producer acquires permission to publish to a topic.

    ProducerAccessMode::equal

    source

    ProducerAccessMode::not_equal

    source

    ProducerAccessMode::to_repr

    source

    ProducerInterceptor

    source
    pub struct ProducerInterceptor {
      // private fields
    }

    ProducerInterceptor::new

    source
    fn ProducerInterceptor::new(before_send? : (ProducerMessage) -> ProducerMessage, on_send_result? : (ProducerMessage, Result[SendReceipt, Error]) -> Unit) -> ProducerInterceptor

    ProducerMessage

    source
    pub(all) struct ProducerMessage {
      payload : Bytes
      key : String?
      properties : Map[String, String]
      event_time : UInt64?
      sequence_id : UInt64?
      ordering_key : Bytes?
      deliver_at : UInt64?
      deliver_after : UInt64?
    }

    A message to be produced.

    ProducerMessage::new

    source
    fn ProducerMessage::new(payload : Bytes, key? : String, properties? : Map[String, String], event_time? : UInt64, sequence_id? : UInt64, ordering_key? : Bytes, deliver_at? : UInt64, deliver_after? : UInt64) -> ProducerMessage

    Create a message to produce. Only payload is required.

    ProducerRoutingMode

    source
    pub(all) enum ProducerRoutingMode {
      KeyHash
      RoundRobin
      SinglePartition
      Custom((ProducerMessage, Int) -> Int)
    }

    Choose a partition for a partitioned producer. The default hashes keyed messages and distributes keyless messages in round-robin order.

    Reader

    source
    pub struct Reader {
      // private fields
    }

    A reader over a topic, partitioned or not.

    Reader::close

    source
    async fn Reader::close(self : Reader) -> Unit

    Close the reader (all partitions).

    Reader::has_message_available

    source
    async fn Reader::has_message_available(self : Reader) -> Bool

    True when there are buffered messages or the broker reports unseen messages beyond the last one read (any partition). Query failures are raised.

    Reader::is_closed

    source
    fn Reader::is_closed(self : Reader) -> Bool

    Reader::last_message_id

    source
    async fn Reader::last_message_id(self : Reader) -> MessageId

    Return the last ID for a single-topic reader. A partitioned reader has multiple independent last IDs and must be queried per partition.

    Reader::last_message_ids

    source
    async fn Reader::last_message_ids(self : Reader) -> Map[String, MessageId]

    Return the broker's last message ID for each topic this reader follows. Partitioned readers report each partition topic separately.

    Reader::next

    source
    async fn Reader::next(self : Reader) -> Message?

    Read the next message, or None once the end of the topic is reached.

    Reader::partitioned

    source
    fn Reader::partitioned(group : 
    TaskGroup    [Unit], readers : Array[SingleReader]) -> Reader

    Aggregate one reader per partition. The merged stream ends (next returns None) once every partition reached its end.

    Reader::seek

    source
    async fn Reader::seek(self : Reader, position : SeekPosition) -> Unit

    Reposition all partition readers.

    ReceiptFuture

    source
    pub struct ReceiptFuture {
      // private fields
    }

    A future receipt for a message sent with SingleProducer::send_async.

    ReceiptFuture::wait

    source
    async fn ReceiptFuture::wait(self : ReceiptFuture) -> SendReceipt

    Wait for the broker's send receipt.

    RetryPolicy

    source
    pub(all) struct RetryPolicy {
      max_retries : Int
      initial_backoff_ms : Int
      max_backoff_ms : Int
    } derive(
    Debug    )

    Retry policy for broker operations such as reconnect and lookup.

    RetryPolicy::default

    source
    fn RetryPolicy::default() -> RetryPolicy

    Default policy: unlimited retries, 100ms initial backoff doubling up to 30 seconds.

    RetryPolicy::to_repr

    source

    SchemaCodec

    source
    pub struct SchemaCodec[T] {
      // private fields
    }

    Converts values to and from Pulsar message payloads. Pass info() to producer/consumer creation, then use encode and decode for messages.

    SchemaCodec::avro

    source
    fn[T] SchemaCodec::avro(definition : Bytes, to_datum : (T) -> 
    Datum     raise PulsarError, from_datum : (
    Datum    ) -> T raise PulsarError) -> SchemaCodec[T] raise PulsarError

    Avro binary payloads with typed mappings to and from Avro datums. The definition must be a valid Avro schema. Record fields follow schema order, and malformed payloads are reported as InvalidFrame.

    SchemaCodec::avro_datum

    source
    fn SchemaCodec::avro_datum(definition : Bytes) -> SchemaCodec[
    Datum    ] raise PulsarError

    Avro binary payloads represented directly as schema-checked datums.

    SchemaCodec::bytes

    source
    fn SchemaCodec::bytes() -> SchemaCodec[Bytes]

    SchemaCodec::decode

    source
    fn[T] SchemaCodec::decode(self : SchemaCodec[T], payload : Bytes) -> T raise PulsarError

    Decode a payload into a typed value.

    SchemaCodec::double

    source
    fn SchemaCodec::double() -> SchemaCodec[Double]

    SchemaCodec::encode

    source
    fn[T] SchemaCodec::encode(self : SchemaCodec[T], value : T) -> Bytes raise PulsarError

    Encode a value into a message payload.

    SchemaCodec::float

    source
    fn SchemaCodec::float() -> SchemaCodec[Float]

    SchemaCodec::info

    source
    fn[T] SchemaCodec::info(self : SchemaCodec[T]) -> SchemaInfo

    Schema declaration for Client::create_producer or Client::create_consumer.

    SchemaCodec::int16

    source
    fn SchemaCodec::int16(little_endian? : Bool) -> SchemaCodec[Int]

    The official Go client uses little-endian integers; pass little_endian=true to exchange INT16 values with it.

    SchemaCodec::int32

    source
    fn SchemaCodec::int32(little_endian? : Bool) -> SchemaCodec[Int]

    The official Go client uses little-endian integers; pass little_endian=true to exchange INT32 values with it.

    SchemaCodec::int64

    source
    fn SchemaCodec::int64(little_endian? : Bool) -> SchemaCodec[Int64]

    The official Go client uses little-endian integers; pass little_endian=true to exchange INT64 values with it.

    SchemaCodec::int8

    source
    fn SchemaCodec::int8() -> SchemaCodec[Int]

    INT8 values use Int because MoonBit has no signed eight-bit scalar.

    SchemaCodec::json

    source
    fn[T : ToJson + 
    FromJson    ] SchemaCodec::json(definition : Bytes) -> SchemaCodec[T]

    JSON payloads for a MoonBit type with JSON conversion support. Supply the broker's JSON schema definition separately from the value encoding.

    SchemaCodec::new

    source
    fn[T] SchemaCodec::new(info : SchemaInfo, encode : (T) -> Bytes raise PulsarError, decode : (Bytes) -> T raise PulsarError) -> SchemaCodec[T]

    Use a custom payload codec with its broker schema declaration.

    SchemaCodec::protobuf

    source

    Use a generated MoonBit protobuf message with the legacy PROTOBUF schema. definition is the Avro-style JSON schema definition used by the Go and Java clients for this schema type. Payloads are ordinary protobuf bytes.

    SchemaCodec::protobuf_native

    source
    fn[M : 
    Read     + 
    Write    ] SchemaCodec::protobuf_native(descriptor_set : Bytes, root_file : String, root_message : String) -> SchemaCodec[M] raise PulsarError

    Use a generated MoonBit protobuf message with a Protobuf Native schema. The descriptor set must describe M. Decoder errors are reported as InvalidFrame.

    SchemaCodec::string

    source
    fn SchemaCodec::string() -> SchemaCodec[String]

    UTF-8 string payloads with Pulsar's STRING schema.

    SchemaInfo

    source
    pub(all) struct SchemaInfo {
      name : String
      type_ : 
    Schema_Type
      schema_data : Bytes
      properties : Map[String, String]
    } derive(Eq, 
    Debug    )

    A topic schema as declared to the broker.

    SchemaInfo::avro

    source
    fn SchemaInfo::avro(definition : Bytes) -> SchemaInfo

    Avro schema with the given JSON definition.

    SchemaInfo::bytes

    source
    fn SchemaInfo::bytes() -> SchemaInfo

    Raw bytes schema (no conversion).

    SchemaInfo::double

    source
    fn SchemaInfo::double() -> SchemaInfo

    SchemaInfo::equal

    source
    fn SchemaInfo::equal(SchemaInfo, SchemaInfo) -> Bool

    SchemaInfo::float

    source
    fn SchemaInfo::float() -> SchemaInfo

    SchemaInfo::int16

    source
    fn SchemaInfo::int16() -> SchemaInfo

    SchemaInfo::int32

    source
    fn SchemaInfo::int32() -> SchemaInfo

    SchemaInfo::int64

    source
    fn SchemaInfo::int64() -> SchemaInfo

    SchemaInfo::int8

    source
    fn SchemaInfo::int8() -> SchemaInfo

    SchemaInfo::json

    source
    fn SchemaInfo::json(definition : Bytes) -> SchemaInfo

    JSON schema with the given JSON-Schema definition.

    SchemaInfo::not_equal

    source
    fn SchemaInfo::not_equal(x : SchemaInfo, y : SchemaInfo) -> Bool

    SchemaInfo::protobuf

    source
    fn SchemaInfo::protobuf(definition : Bytes) -> SchemaInfo

    Legacy Protobuf schema with its Avro-style JSON definition.

    SchemaInfo::protobuf_native

    source
    fn SchemaInfo::protobuf_native(descriptor_set : Bytes, root_file : String, root_message : String) -> SchemaInfo raise PulsarError

    Protobuf Native schema metadata accepted by the Java and Go clients. descriptor_set is the binary google.protobuf.FileDescriptorSet, including the root file and its dependencies.

    SchemaInfo::string

    source
    fn SchemaInfo::string() -> SchemaInfo

    Built-in UTF-8 string schema.

    SchemaInfo::to_repr

    source

    SeekPosition

    source
    pub(all) enum SeekPosition {
      PositionId(MessageId)
      PositionTimestamp(UInt64)
    }

    A position a reader can seek to.

    SendReceipt

    source
    pub(all) struct SendReceipt {
      message_id : MessageId
      sequence_id : UInt64
      producer_id : UInt64
      producer_name : String
    } derive(
    Debug    )

    Confirmation that the broker persisted a message.

    SendReceipt::to_repr

    source

    SingleConsumer

    source
    pub struct SingleConsumer {
      // private fields
    }

    A consumer bound to a topic subscription.

    SingleConsumer::ack

    source
    async fn SingleConsumer::ack(self : SingleConsumer, id : MessageId, await_response? : Bool) -> Unit

    Acknowledge a message individually.

    SingleConsumer::ack_cumulative

    source
    async fn SingleConsumer::ack_cumulative(self : SingleConsumer, id : MessageId, await_response? : Bool) -> Unit

    Acknowledge cumulatively: this message and all before it.

    SingleConsumer::ack_ids

    source
    async fn SingleConsumer::ack_ids(self : SingleConsumer, ids : Array[MessageId], await_response? : Bool) -> Unit

    Acknowledge several IDs. Ordinary IDs share one ACK frame; batch indexes retain their per-entry bitmap handling. An empty list has no effect.

    SingleConsumer::ack_txn

    source
    async fn SingleConsumer::ack_txn(self : SingleConsumer, txn : Transaction, id : MessageId, await_response? : Bool) -> Unit

    Acknowledge a message as part of a transaction.

    SingleConsumer::close

    source
    async fn SingleConsumer::close(self : SingleConsumer) -> Unit

    Close the consumer and its subscription cursor is kept (durable).

    SingleConsumer::create

    source
    async fn SingleConsumer::create(group : 
    TaskGroup    [Unit], conn : Connection, topic : String, subscription : String, sub_type? : SubscriptionType, key_shared_policy? : KeySharedPolicy, consumer_name? : String, initial_position? : SubscriptionInitialPosition, priority_level? : Int, durable? : Bool, read_compacted? : Bool, properties? : Map[String, String], subscription_properties? : Map[String, String], receiver_queue_size? : Int, ack_timeout_ms? : Int, nack_backoff? : NackBackoff, reconnect_with? : async () -> Connection, schema? : SchemaInfo, message_crypto? : MessageCrypto, decryption_failure_action? : DecryptionFailureAction, reconnect_policy? : RetryPolicy, chunk_policy? : ChunkAssemblyPolicy, enable_batch_index_ack? : Bool, ack_grouping? : AckGroupingOptions, auto_scaled_receiver_queue? : Bool, on_event? : (ClientEvent) -> Unit) -> SingleConsumer

    Create a consumer on topic through an already-resolved connection.

    SingleConsumer::is_closed

    source
    fn SingleConsumer::is_closed(self : SingleConsumer) -> Bool

    SingleConsumer::last_message_id

    source
    async fn SingleConsumer::last_message_id(self : SingleConsumer) -> MessageId

    Return the broker's last message ID for this consumer's topic.

    SingleConsumer::nack

    source
    async fn SingleConsumer::nack(self : SingleConsumer, id : MessageId) -> Unit

    Negatively acknowledge a message, triggering redelivery.

    SingleConsumer::receive

    source
    async fn SingleConsumer::receive(self : SingleConsumer) -> Message

    SingleConsumer::seek

    source
    async fn SingleConsumer::seek(self : SingleConsumer, position : SeekPosition) -> Unit

    Reset this subscription to a message id or publish timestamp. Already buffered messages are discarded after the broker accepts the seek.

    SingleConsumer::subscription

    source
    fn SingleConsumer::subscription(self : SingleConsumer) -> String

    SingleConsumer::topic

    source
    fn SingleConsumer::topic(self : SingleConsumer) -> String

    SingleConsumer::try_receive

    source
    fn SingleConsumer::try_receive(self : SingleConsumer) -> Message? raise

    SingleConsumer::unsubscribe

    source
    async fn SingleConsumer::unsubscribe(self : SingleConsumer, force? : Bool) -> Unit

    Remove this subscription from the broker and close the consumer.

    SingleProducer

    source
    pub struct SingleProducer {
      // private fields
    }

    A producer bound to a single topic.

    SingleProducer::close

    source
    async fn SingleProducer::close(self : SingleProducer) -> Unit

    Close the producer, flushing any buffered batch first.

    SingleProducer::create

    source
    async fn SingleProducer::create(group : 
    TaskGroup    [Unit], conn : Connection, topic : String, name? : String, properties? : Map[String, String], access_mode? : ProducerAccessMode, batching? : BatchingConfig, reconnect_with? : async () -> Connection, compression? : CompressionCodec, message_crypto? : MessageCrypto, encryption_key_names? : Array[String], chunk_size? : Int, automatic_chunking? : Bool, schema? : SchemaInfo, reconnect_policy? : RetryPolicy, max_pending_messages? : Int, block_if_queue_full? : Bool, send_timeout_ms? : Int) -> SingleProducer

    Create a producer on topic through an already-resolved connection. reconnect_with should yield a fresh connection to the owning broker; it enables automatic re-registration and pending message replay.

    SingleProducer::flush

    source
    async fn SingleProducer::flush(self : SingleProducer) -> Unit

    Flush any buffered batch immediately.

    SingleProducer::last_sequence_id

    source
    fn SingleProducer::last_sequence_id(self : SingleProducer) -> UInt64?

    Highest sequence id confirmed by this producer or reported by the broker when it registered. None means that no sequence has been confirmed.

    SingleProducer::name

    source
    fn SingleProducer::name(self : SingleProducer) -> String

    The broker-assigned producer name.

    SingleProducer::send

    source
    async fn SingleProducer::send(self : SingleProducer, msg : ProducerMessage) -> SendReceipt

    Send a message and wait for the broker's receipt.

    SingleProducer::send_async

    source
    async fn SingleProducer::send_async(self : SingleProducer, msg : ProducerMessage) -> ReceiptFuture

    Send a message without waiting for the receipt; the returned future resolves once the broker confirms.

    SingleProducer::send_txn

    source
    async fn SingleProducer::send_txn(self : SingleProducer, txn : Transaction, msg : ProducerMessage) -> SendReceipt

    Send a message as part of a transaction.

    SingleProducer::topic

    source
    fn SingleProducer::topic(self : SingleProducer) -> String

    The topic this producer publishes to.

    SingleReader

    source
    pub struct SingleReader {
      // private fields
    }

    A reader over a topic starting at a given position.

    SingleReader::close

    source
    async fn SingleReader::close(self : SingleReader) -> Unit

    Close the reader.

    SingleReader::create

    source
    async fn SingleReader::create(group : 
    TaskGroup    [Unit], conn : Connection, topic : String, start_message_id? : MessageId, receiver_queue_size? : Int, read_compacted? : Bool, include_start_message_id? : Bool, name? : String, properties? : Map[String, String], subscription_name? : String, subscription_role_prefix? : String, chunk_policy? : ChunkAssemblyPolicy, schema? : SchemaInfo, message_crypto? : MessageCrypto, decryption_failure_action? : DecryptionFailureAction) -> SingleReader

    Create a reader on topic through an already-resolved connection, starting after start_message_id (default: the latest message). The subscription is non-durable and exclusive.

    SingleReader::has_message_available

    source
    async fn SingleReader::has_message_available(self : SingleReader) -> Bool

    True when there are buffered messages or the broker reports unseen messages beyond the last one read. Query failures are raised.

    SingleReader::last_message_id

    source
    async fn SingleReader::last_message_id(self : SingleReader) -> MessageId

    Return the broker's last message ID for this single-topic reader.

    SingleReader::next

    source
    async fn SingleReader::next(self : SingleReader) -> Message?

    Read the next message, or None once the current end of the topic is reached. Messages published later can still be read by calling next again.

    SingleReader::seek

    source
    async fn SingleReader::seek(self : SingleReader, position : SeekPosition) -> Unit

    Reposition the reader. Buffered messages are dropped.

    PositionId seeks re-subscribe starting at the given message id. Timestamp seeks issue a broker-side cursor reset; the broker disconnects the consumer during the reset, so the reader re-subscribes to the (reset) cursor afterwards.

    SubscriptionInitialPosition

    source
    pub(all) enum SubscriptionInitialPosition {
      Latest
      Earliest
    } derive(Eq, 
    Debug    )

    Where a subscription starts consuming.

    SubscriptionInitialPosition::equal

    source

    SubscriptionInitialPosition::not_equal

    source

    SubscriptionInitialPosition::to_repr

    source

    SubscriptionType

    source
    pub(all) enum SubscriptionType {
      Exclusive
      Shared
      Failover
      KeyShared
    } derive(Eq, 
    Debug    )

    Subscription mode, mirroring CommandSubscribe.SubType.

    SubscriptionType::equal

    source

    SubscriptionType::not_equal

    source
    fn SubscriptionType::not_equal(x : SubscriptionType, y : SubscriptionType) -> Bool

    SubscriptionType::to_repr

    source

    TableView

    source
    pub struct TableView {
      // private fields
    }

    TableView::close

    source
    async fn TableView::close(self : TableView) -> Unit

    TableView::contains_key

    source
    fn TableView::contains_key(self : TableView, key : String) -> Bool

    TableView::entries

    source
    fn TableView::entries(self : TableView) -> Map[String, Bytes]

    Return a separate map so callers cannot mutate the view itself.

    TableView::for_each

    source
    fn TableView::for_each(self : TableView, action : (String, Bytes) -> Unit) -> Unit

    TableView::for_each_and_listen

    source
    fn TableView::for_each_and_listen(self : TableView, action : (String, Bytes?) -> Unit) -> Unit

    Replay the current entries, then call action for future updates. Deletions are delivered as None.

    TableView::get

    source
    fn TableView::get(self : TableView, key : String) -> Bytes?

    TableView::is_empty

    source
    fn TableView::is_empty(self : TableView) -> Bool

    TableView::keys

    source
    fn TableView::keys(self : TableView) -> Array[String]

    TableView::size

    source
    fn TableView::size(self : TableView) -> Int

    TlsIdentity

    source
    pub(all) struct TlsIdentity {
      certificate_file : String
      private_key_file : String
    }

    PEM files used as a TLS client identity on a connection. Files are read again at each new TLS handshake, including lookup redirects and reconnects.

    TlsIdentity::new

    source
    fn TlsIdentity::new(certificate_file : String, private_key_file : String) -> TlsIdentity raise PulsarError

    TlsVersion

    source
    pub(all) enum TlsVersion {
      TLS12
      TLS13
    }

    Protocol versions available in the native TLS adapter.

    TokenAuth

    source
    pub(all) struct TokenAuth {
      token : String
    }

    Token authentication (AuthenticationProviderToken on the broker).

    TokenAuth::auth_data

    source
    fn TokenAuth::auth_data(self : TokenAuth) -> Bytes

    TokenAuth::method_name

    source
    fn TokenAuth::method_name(_ : TokenAuth) -> String

    TokenAuth::new

    source
    fn TokenAuth::new(token : String) -> TokenAuth

    TopicName

    source
    pub(all) struct TopicName {
      domain : String
      tenant : String
      namespace_ : String
      topic : String
    } derive(Eq, 
    Debug    )

    Components of a fully-qualified topic name.

    TopicName::equal

    source
    fn TopicName::equal(TopicName, TopicName) -> Bool

    TopicName::not_equal

    source
    fn TopicName::not_equal(x : TopicName, y : TopicName) -> Bool

    TopicName::parse

    source
    fn TopicName::parse(full : String) -> TopicName raise PulsarError

    Parse "persistent://tenant/namespace/topic" or its non-persistent form.

    TopicName::to_repr

    source

    Transaction

    source
    pub struct Transaction {
      // private fields
    }

    A transaction handle; pass to Producer::send_txn / Consumer::ack_txn and finish with commit or abort.

    Transaction::abort

    source
    async fn Transaction::abort(self : Transaction) -> Unit

    Abort the transaction, rolling back all its messages and acks.

    Transaction::add_partition

    source
    async fn Transaction::add_partition(self : Transaction, topic : String) -> Unit

    Register a topic as a producer participant of this transaction.

    Transaction::add_subscription

    source
    async fn Transaction::add_subscription(self : Transaction, topic : String, subscription : String) -> Unit

    Register a subscription as a consumer participant of this transaction.

    Transaction::commit

    source
    async fn Transaction::commit(self : Transaction) -> Unit

    Commit the transaction.

    Transaction::id

    source
    fn Transaction::id(self : Transaction) -> (UInt64, UInt64)

    The (most, least) bits of the transaction id.

    TransactionCoordinator

    source
    pub struct TransactionCoordinator {
      // private fields
    }

    A handle to a transaction coordinator channel on a broker.

    TransactionCoordinator::close

    source

    Release this coordinator. A dedicated connection is closed; a pooled connection becomes eligible for idle eviction when its last user releases it.

    TransactionCoordinator::connect

    source
    async fn TransactionCoordinator::connect(group : 
    TaskGroup    [Unit], addr : String, tc_id : UInt64, auth? : AuthData, tls? : Bool) -> TransactionCoordinator

    Open a transaction coordinator channel on a fresh connection to addr.

    TransactionCoordinator::new_transaction

    source
    async fn TransactionCoordinator::new_transaction(self : TransactionCoordinator, ttl_ms? : UInt64) -> Transaction

    Start a new transaction with the given time-to-live.

    TypedTableView

    source
    pub struct TypedTableView[T] {
      // private fields
    }

    A TableView that decodes values with a declared Pulsar schema. Tombstones remain None; malformed values raise on access and are passed to listeners as errors so a live update cannot silently replace a value.

    TypedTableView::close

    source
    async fn[T] TypedTableView::close(self : TypedTableView[T]) -> Unit

    TypedTableView::contains_key

    source
    fn[T] TypedTableView::contains_key(self : TypedTableView[T], key : String) -> Bool

    TypedTableView::entries

    source
    fn[T] TypedTableView::entries(self : TypedTableView[T]) -> Map[String, T] raise PulsarError

    TypedTableView::for_each

    source
    fn[T] TypedTableView::for_each(self : TypedTableView[T], action : (String, T) -> Unit) -> Unit raise PulsarError

    TypedTableView::for_each_and_listen

    source
    fn[T] TypedTableView::for_each_and_listen(self : TypedTableView[T], action : (String, Result[T?, PulsarError]) -> Unit) -> Unit

    Replay current entries, then report future values and tombstones. Decode failures are delivered as Err; the entry remains stored and get raises the same error until a valid update or tombstone arrives.

    TypedTableView::get

    source
    fn[T] TypedTableView::get(self : TypedTableView[T], key : String) -> T? raise PulsarError

    TypedTableView::is_empty

    source
    fn[T] TypedTableView::is_empty(self : TypedTableView[T]) -> Bool

    TypedTableView::keys

    source
    fn[T] TypedTableView::keys(self : TypedTableView[T]) -> Array[String]

    TypedTableView::size

    source
    fn[T] TypedTableView::size(self : TypedTableView[T]) -> Int

    VersionedSchema

    source
    pub(all) struct VersionedSchema {
      info : SchemaInfo
      version : Bytes?
    } derive(Eq, 
    Debug    )

    A schema returned by the broker, with its registry version when present.

    VersionedSchema::equal

    source

    VersionedSchema::not_equal

    source
    fn VersionedSchema::not_equal(x : VersionedSchema, y : VersionedSchema) -> Bool

    VersionedSchema::to_repr

    source

    athenz_role_token_file_provider

    source
    fn athenz_role_token_file_provider(path : String) -> (async () -> AuthData)

    Read an Athenz role token file afresh for each connection or challenge.

    athenz_role_token_provider

    source
    fn athenz_role_token_provider(supplier : async () -> String) -> (async () -> AuthData)

    Use an Athenz role token supplied by a sidecar or another ZTS client. The supplier is invoked on each connection and authentication challenge.

    athenz_zts_provider

    source
    fn athenz_zts_provider(options : AthenzZtsOptions) -> (async () -> AuthData)

    Sign a service NToken or present a client certificate to ZTS, then cache the role token until shortly before expiry. Failed refreshes retry and may use a still valid cached token. Pass the result as ClientOptions::new(auth_provider=...).

    broker_addr_of

    source
    fn broker_addr_of(url : String) -> (String, Bool) raise PulsarError

    Strip the scheme of a pulsar service URL, returning ("host:port", tls?).

    compress

    source
    fn compress(codec : CompressionCodec, payload : Bytes) -> Bytes raise PulsarError

    Compress payload with the codec; None returns the input unchanged.

    crc32c

    source
    fn crc32c(data : BytesView) -> UInt

    Compute the CRC32-C (Castagnoli) checksum of data.

    decode_frame_body

    source
    fn decode_frame_body(body : Bytes) -> IncomingFrame raise PulsarError

    Decode a frame body (without the leading total size field).

    decompress

    source
    fn decompress(codec : CompressionCodec, payload : Bytes, uncompressed_size? : Int) -> Bytes raise PulsarError

    Decompress a payload produced by compress.

    encode_command_frame

    source
    fn encode_command_frame(cmd : 
    BaseCommand    ) -> Bytes raise

    Encode a plain command frame (no payload attached).

    encode_send_frame

    source
    fn encode_send_frame(cmd : 
    BaseCommand    , metadata : 
    MessageMetadata    , payload : Bytes) -> Bytes raise

    Encode a message frame (SEND command with metadata and payload). Client-side frames never carry broker entry metadata.

    get_partition_count

    source
    async fn get_partition_count(manager : ConnectionManager, addr : String, topic : String) -> Int

    Number of partitions of a topic; 0 means non-partitioned.

    get_topics_of_namespace

    source
    async fn get_topics_of_namespace(manager : ConnectionManager, addr : String, ns : String, pattern? : String, mode? : NamespaceTopicMode) -> Array[String]

    Ask a broker for topics in tenant/namespace and apply an optional MoonBit regular expression locally. This also works with brokers that do not implement server-side pattern filtering.

    lookup_topic

    source
    async fn lookup_topic(manager : ConnectionManager, addr : String, topic : String, authoritative? : Bool) -> Connection

    Look up topic starting from the broker at addr and return a connection to the owning broker. Redirects are followed transparently; if the broker answers with proxy_through_service_url, the connection used for the lookup is returned.

    lz4_compress

    source
    fn lz4_compress(data : Bytes) -> Bytes

    Compress data as an LZ4 frame.

    lz4_compress_block

    source
    fn lz4_compress_block(data : BytesView) -> Bytes

    Compress raw LZ4 block format (the wire format Pulsar brokers exchange).

    lz4_decompress

    source
    fn lz4_decompress(data : Bytes) -> Bytes raise PulsarError

    Decompress an LZ4 frame.

    lz4_decompress_block_to

    source
    fn lz4_decompress_block_to(data : BytesView, size : Int) -> Bytes raise PulsarError

    Decompress a raw LZ4 block to exactly size bytes.

    oauth2_client_credentials_provider

    source
    fn oauth2_client_credentials_provider(token_endpoint : String, client_id : String, client_secret : String, audience? : String, scope? : String) -> (async () -> AuthData) raise PulsarError

    OAuth2 client-credentials authentication using client_secret_post. The provider requests a fresh access token on each connection or broker challenge; pass it as ClientOptions::new(auth_provider=...).

    read_frame

    source
    async fn[R : 
    Reader    ] read_frame(reader : R) -> IncomingFrame

    Read one frame from an async byte stream: 4-byte big-endian total size, then the frame body.

    snappy_compress

    source
    fn snappy_compress(data : Bytes) -> Bytes

    Compress data into Google's Snappy framed stream format.

    snappy_decompress

    source
    fn snappy_decompress(data : Bytes) -> Bytes raise PulsarError

    Decompress Google's Snappy framed stream format.

    split_host_port

    source
    fn split_host_port(addr : String) -> (String, Int) raise PulsarError

    Split "host:port" into (host, port); a missing port defaults to 6650.

    token_file_provider

    source
    fn token_file_provider(path : String) -> (async () -> AuthData)

    Read a token each time a connection or authentication challenge needs it. This permits token rotation without reconstructing the client.

    with_retry

    source
    async fn[T] with_retry(policy : RetryPolicy, f : async () -> T) -> T

    Run f with the retry policy, retrying any failure.

    xxhash32

    source
    fn xxhash32(data : BytesView) -> UInt

    Compute xxHash32 of data with seed 0.

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io