Connect to your cluster using the Rust client

This document describes how to connect to a cluster through a Rust client, and use the Rust producer and consumer to produce and consume messages to and from a topic. The Rust client supports connecting to a Pulsar cluster through either OAuth2 or Token authentication.

Note

This document assumes that you have created a Pulsar cluster and a service account, and have granted the service account produce and consume permissions to the namespace for the target topic.

Prerequisites

See the minimum supported versions required for the underlying libraries for more details.

Connect to cluster through OAuth2 authentication

To connect a Pulsar cluster through OAuth2 authentication, follow these steps.

Step 1: Get the broker service URL of your cluster

To get the service URL of a Pulsar cluster through the StreamNative Cloud Console, follow these steps.

On the left navigation pane, in the Admin area, click Pulsar Clusters.
Select the Overview tab, and in the Access Points area, click Copy at the end of the row of the service URL.

Step 2: Get the OAuth2 credential file of your service account

To get an OAuth2 credential file of a service account through the StreamNative Cloud Console, follow these steps.

On the left navigation pane, click Service Accounts.
In the row of the service account you want to use, in the Key File column, click the Download icon to download the OAuth2 credential file to your local directory.
The OAuth2 credential file should be something like this:
```
{
  "type": "SN_SERVICE_ACCOUNT",
  "client_id": "CLIENT_ID",
  "client_secret": "CLIENT_SECRET",
  "client_email": "[email protected]",
  "issuer_url": "https://auth.streamnative.cloud"
}
```

Step 3: Connect to your cluster

   let addr = "{{ SERVICE_URL }}".to_string();
   let mut builder = Pulsar::builder(addr, TokioExecutor);
   builder = builder.with_auth_provider(OAuth2Authentication::client_credentials(OAuth2Params {
     issuer_url: "https://auth.streamnative.cloud/".to_string(),
     credentials_url: "{{ file://YOUR-KEY-FILE-PATH }}".to_string(), // Absolute path of your downloaded key file
     audience: Some("{{ YOUR-AUDIENCE }}".to_string()),
     scope: None,
   }));

SERVICE_URL: the broker service URL of your Pulsar cluster.
issuer_url: the URL of your OAuth2 authentication provider. You can get the value from your downloaded OAuth2 credential file.
credentials_url: the path to your downloaded OAuth2 credential file. The credentials_url parameter supports the following three pattern formats:
- file:///path/to/file
- file:/path/to/file
- data:application/json;base64,<base64-encoded value>
audience: the audience parameter is the Uniform Resource Name (URN), which is a combination of the urn:sn:pulsar, the organization name, and the Pulsar instance name, in this format urn:sn:pulsar:<org_name>:<instance_name>.

Create a Rust consumer to consume messages

 use futures::TryStreamExt;
 use pulsar::{Consumer, ConsumerOptions, Pulsar, SubType, TokioExecutor};
 use pulsar::authentication::oauth2::{OAuth2Authentication, OAuth2Params};
 use pulsar::consumer::InitialPosition;

 #[tokio::main]
 async fn main() -> Result<(), pulsar::Error> {
     env_logger::init();

     let addr = "{{ SERVICE_URL }}".to_string();
     let mut builder = Pulsar::builder(addr, TokioExecutor);
     builder = builder.with_auth_provider(OAuth2Authentication::client_credentials(OAuth2Params {
         issuer_url: "https://auth.streamnative.cloud/".to_string(),
         credentials_url: "{{ file://YOUR-KEY-FILE-PATH }}".to_string(), // Absolute path of your downloaded key file
         audience: Some("{{ YOUR-AUDIENCE }}".to_string()),
         scope: None,
     }));

     let pulsar: Pulsar<_> = builder.build().await?;
     let mut consumer: Consumer<String, _> = pulsar
         .consumer()
         .with_topic("persistent://public/default/test-topic")
         .with_subscription_type(SubType::Exclusive)
         .with_subscription("test-sub")
         .with_options(ConsumerOptions::default()
             .with_initial_position(InitialPosition::Earliest))
         .build()
         .await?;

     let mut counter = 0usize;
     while let Some(msg) = consumer.try_next().await? {
         consumer.ack(&msg).await?;
         let payload = match msg.deserialize() {
             Ok(payload) => payload,
             Err(e) => {
                 println!("could not deserialize message: {:?}", e);
                 break;
             }
         };

         counter += 1;
         println!("Received message '{:?}' id='{:?}'", payload, msg.message_id());

         if counter > 10 {
             consumer.close().await.expect("Unable to close consumer");
             break;
         }
     }

     Ok(())
 }

Create a Rust producer to produce messages

 use pulsar::{Pulsar, TokioExecutor};
 use pulsar::authentication::oauth2::{OAuth2Authentication, OAuth2Params};

 #[tokio::main]
 async fn main() -> Result<(), pulsar::Error> {
     env_logger::init();

     let addr = "{{ SERVICE_URL }}".to_string();
     let mut builder = Pulsar::builder(addr, TokioExecutor);
     builder = builder.with_auth_provider(OAuth2Authentication::client_credentials(OAuth2Params {
         issuer_url: "https://auth.streamnative.cloud/".to_string(),
         credentials_url: "{{ file://YOUR-KEY-FILE-PATH }}".to_string(), // Absolute path of your downloaded key file
         audience: Some("{{ YOUR-AUDIENCE }}".to_string()),
         scope: None,
     }));

     let pulsar: Pulsar<_> = builder.build().await?;
     let mut producer = pulsar
         .producer()
         .with_topic("persistent://public/default/test-topic")
         .build()
         .await?;

     let mut counter = 0usize;
     loop {
         producer
             .send(format!("Hello-{}", counter))
             .await?
             .await
             .unwrap();

         counter += 1;
         println!("{counter} messages");

         if counter > 10 {
             producer.close().await.expect("Unable to close connection");
             break;
         }
     }

     Ok(())
 }

Connect to cluster through Token authentication

To connect a Pulsar cluster through Token authentication, follow these steps.

Step 1: Get the broker service URL of your cluster

To get the service URL of a Pulsar cluster through the StreamNative Cloud Console, follow these steps.

On the left navigation pane, in the Admin area, click Pulsar Clusters.
Select the Overview tab, and in the Access Points area, click Copy at the end of the row of the service URL.

Step 2: Get the Token of your service account

Note

Before getting the token of a service account, verify that the service account is authorized as a superuser or an admin of the tenants and namespaces.

To get a token using the StreamNative Cloud Console, follow these steps.

On the left navigation pane, click Service Accounts.
In the row of the service account you want to use, in the Token column, click Generate new token, then click the Copy icon to copy the token to your clipboard.

Note

A token has a system-defined Time-To-Live (TTL) of 7 days. Before a token expires, ensure that you generate a new token for your service account.

Step 3: Connect to your cluster

 let addr = "{{ SERVICE_URL }}".to_string();
 let mut builder = Pulsar::builder(addr, TokioExecutor);
 let token = "{{ AUTH_PARAMS_TOKE }}".to_string();
 builder = builder.with_auth(Authentication {
     name: "token".to_string(),
     data: token.into_bytes(),
 });

SERVICE_URL: the broker service URL of your Pulsar cluster.
AUTH_PARAMS_TOKE: the token of your service account.

For a complete example about how to connect to a cluster through the Rust client, see Rust client examples.