Struct Pulsar 

pub struct Pulsar<Exe: Executor> { /* private fields */ }

Pulsar client

This is the starting point of this API, used to create connections, producers and consumers

While methods are provided to create the client, producers and consumers directly, the builders should be used for more clarity:

use pulsar::{Pulsar, TokioExecutor};

let addr = "pulsar://127.0.0.1:6650";
// you can indicate which executor you use as the return type of client creation
let pulsar: Pulsar<_> = Pulsar::builder(addr, TokioExecutor)
    .with_auth(auth)
    .with_connection_retry_options(retry)
    .build()
    .await?;

let mut producer = pulsar
    .producer()
    .with_topic("non-persistent://public/default/test")
    .with_name("my producer")
    .build()
    .await?;

Implementations

impl<Exe: Executor> Pulsar<Exe>

pub fn builder<S: Into<String>>(url: S, executor: Exe) -> PulsarBuilder<Exe>

creates a new client builder

use pulsar::{Pulsar, TokioExecutor};

let addr = "pulsar://127.0.0.1:6650";
// you can indicate which executor you use as the return type of client creation
let pulsar: Pulsar<_> = Pulsar::builder(addr, TokioExecutor).build().await?;

pub fn consumer(&self) -> ConsumerBuilder<Exe>

creates a consumer builder

use pulsar::{SubType, Consumer};

let mut consumer: Consumer<TestData, _> = pulsar
    .consumer()
    .with_topic("non-persistent://public/default/test")
    .with_consumer_name("test_consumer")
    .with_subscription_type(SubType::Exclusive)
    .with_subscription("test_subscription")
    .build()
    .await?;

pub fn producer(&self) -> ProducerBuilder<Exe>

creates a producer builder

let mut producer = pulsar
    .producer()
    .with_topic("non-persistent://public/default/test")
    .with_name("my producer")
    .build()
    .await?;

pub fn reader(&self) -> ConsumerBuilder<Exe>

creates a reader builder

use pulsar::reader::Reader;

let mut reader: Reader<TestData, _> = pulsar
    .reader()
    .with_topic("non-persistent://public/default/test")
    .with_consumer_name("my_reader")
    .into_reader()
    .await?;

pub async fn lookup_topic<S: Into<String>>( &self, topic: S, ) -> Result<BrokerAddress, Error>

gets the address of a broker handling the topic

let broker_address = pulsar.lookup_topic("persistent://public/default/test").await?;

pub async fn lookup_partitioned_topic_number<S: Into<String>>( &self, topic: S, ) -> Result<u32, Error>

gets the number of partitions for a partitioned topic

let nb = pulsar.lookup_partitioned_topic_number("persistent://public/default/test").await?;

pub async fn lookup_partitioned_topic<S: Into<String>>( &self, topic: S, ) -> Result<Vec<(String, BrokerAddress)>, Error>

gets the address of brokers handling the topic’s partitions. If the topic is not a partitioned topic, result will be a single element containing the topic and address of the non-partitioned topic provided.

let broker_addresses = pulsar.lookup_partitioned_topic("persistent://public/default/test").await?;

pub async fn get_topics_of_namespace( &self, namespace: String, mode: Mode, ) -> Result<Vec<String>, Error>

gets the list of topics from a namespace

use pulsar::message::proto::command_get_topics_of_namespace::Mode;

let topics = pulsar.get_topics_of_namespace("public/default".to_string(), Mode::Persistent).await?;

pub async fn send<S: Into<String>, M: SerializeMessage + Sized>( &self, topic: S, message: M, ) -> Result<SendFuture, Error>

Sends a message on a topic.

This function will lazily initialize and re-use producers as needed. For better control over producers, creating and using a Producer is recommended.

use pulsar::message::proto::command_get_topics_of_namespace::Mode;

let topics = pulsar.send("persistent://public/default/test", "hello world!").await?;

Trait Implementations

impl<Exe: Clone + Executor> Clone for Pulsar<Exe>

fn clone(&self) -> Pulsar<Exe>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.