Connect to your cluster using the C# client

This document describes how to connect to a cluster through a C# client, and use the C# producer and consumer to produce and consume messages to and from a topic. The C# 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

Install Pulsar client 2.4 or higher versions.

Connect to your 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 Console, follow these steps.

  1. On the left navigation pane, in the Admin area, click Pulsar Clusters.

  2. Select the Details 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 Console, follow these steps.

  1. On the left navigation pane, click Service Accounts.

  2. 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

For a complete example of how to connect to a cluster through the C# client, see C# client examples.

Create a C# consumer to consume messages

You can create and configure a C# consumer to consume messages using the OAuth2 credential file as follows. For more information about the placeholders in the code sample, see parameters for OAuth2 authentication.

using System;
using System.IO;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
using Pulsar.Client.Api;
using Pulsar.Client.Common;

namespace CsharpExamples
{
    internal class Oauth2
    {
        internal static async Task RunOauth()
        {
            var fileUri = new Uri("{{ file://YOUR-KEY-FILE-PATH }}");
            var issuerUrl = new Uri("https://auth.streamnative.cloud/");
            var audience = "urn:sn:pulsar:${orgName}:${instanceName}";

            const string serviceUrl = "${brokerServiceURL}";
            var topicName = "persistent://${tenant}/${namespace}/${topic}";
            const string subscriptionName = "${subscription}";

            var client = await new PulsarClientBuilder()
                .ServiceUrl(serviceUrl)
                .Authentication(AuthenticationFactoryOAuth2.ClientCredentials(issuerUrl, audience, fileUri))
                .BuildAsync();

            var consumer = await client.NewConsumer()
                .Topic(topicName)
                .SubscriptionName(subscriptionName)
                .SubscribeAsync();

            for(int i=0; i< 10; i++){
                var message = await consumer.ReceiveAsync();
                Console.WriteLine($"Received: {Encoding.UTF8.GetString(message.Data)}");

                await consumer.AcknowledgeAsync(message.MessageId);
            }
        }
    }
}

Create a C# producer to produce messages

You can create and configure a C# producer to produce messages using the OAuth2 credential file as follows. For more information about the placeholders in the code sample, see parameters for OAuth2 authentication.

using System;
using System.IO;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
using Pulsar.Client.Api;
using Pulsar.Client.Common;

namespace CsharpExamples
{
    internal class Oauth2
    {
        internal static async Task RunOauth()
        {
            var fileUri = new Uri("{{ file://YOUR-KEY-FILE-PATH }}");
            var issuerUrl = new Uri("https://auth.streamnative.cloud/");
            var audience = "urn:sn:pulsar:${orgName}:${instanceName}";

            const string serviceUrl = "${brokerServiceURL}";
            var topicName = "persistent://${tenant}/${namespace}/${topic}";

            var client = await new PulsarClientBuilder()
                .ServiceUrl(serviceUrl)
                .Authentication(AuthenticationFactoryOAuth2.ClientCredentials(issuerUrl, audience, fileUri))
                .BuildAsync();

            var producer = await client.NewProducer()
                .Topic(topicName)
                .CreateAsync();

            for(int i=0; i< 10; i++){
                var messageId = await producer.SendAsync(Encoding.UTF8.GetBytes($"Sent from C# at '{DateTime.Now}'"));
                Console.WriteLine($"MessageId is: '{messageId}'");
            }
        }
    }
}

Parameters for OAuth2 authentication

  • fileUri: your downloaded OAuth2 credential. This parameter supports the following two pattern formats:
    • file:///path/to/file: the path to your downloaded OAuth2 credential file.
    • data:application/json;base64,<base64-encoded value>: the credential file content encoded into Base64 format.
  • audience: the Uniform Resource Name (URN), which is a combination of the urn:sn:pulsar, your organization name, and your Pulsar instance name.
  • ${brokerServiceURL}: the broker service URL of your Pulsar cluster.
  • ${tenant}/${namespace}/${topic}: the full name of the topic for message production & consumption. It is a combination of the tenant name, the namespace name and the topic name.
  • ${subscription}: the name of the subscription that will determine how messages are delivered.

Connect to your 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 Console, follow these steps.

  1. On the left navigation pane, in the Admin area, click Pulsar Clusters.

  2. Select the Details 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.
  • 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.

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

  1. On the left navigation pane, click Service Accounts.

  2. 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.

Step 3: Connect to your cluster

For a complete example of how to connect to a cluster through the C# client, see C# client examples.

Create a C# consumer to consume messages

You can create and configure a C# consumer to consume messages using Token authentication as follows. For more information about the placeholders in the code sample, see parameters for Token authentication.

using System;
using System.Text;
using System.Threading.Tasks;
using Pulsar.Client.Api;

namespace CsharpExamples
{
    internal class JWT
    {
        internal static async Task Consumer()
        {
            const string serviceUrl = "${brokerServiceURL}";
            const string subscriptionName = "${subscription}";
            var topicName = "persistent://${tenant}/${namespace}/${topic}";
            var token = "${token}";

            var client = await new PulsarClientBuilder()
                .ServiceUrl(serviceUrl)
                .Authentication(AuthenticationFactory.Token(token))
                .BuildAsync();

            var consumer = await client.NewConsumer()
                .Topic(topicName)
                .SubscriptionName(subscriptionName)
                .SubscribeAsync();

            for(int i=0; i< 10; i++){
                var message = await consumer.ReceiveAsync();
                Console.WriteLine($"Received: {Encoding.UTF8.GetString(message.Data)}");

                await consumer.AcknowledgeAsync(message.MessageId);
            }
        }
    }
}

Create a C# producer to produce messages

You can create and configure a C# producer to produce messages using Token authentication as follows. For more information about the placeholders in the code sample, see parameters for Token authentication.

using System;
using System.Text;
using System.Threading.Tasks;
using Pulsar.Client.Api;

namespace CsharpExamples
{
    internal class JWT
    {
        internal static async Task Producer()
        {
            const string serviceUrl = "${brokerServiceURL}";
            var topicName = "persistent://${tenant}/${namespace}/${topic}";
            var token = "${token}";

            var client = await new PulsarClientBuilder()
                .ServiceUrl(serviceUrl)
                .Authentication(AuthenticationFactory.Token(token))
                .BuildAsync();

            var producer = await client.NewProducer()
                .Topic(topicName)
                .CreateAsync();

            for(int i=0; i< 10; i++){
                var messageId = await producer.SendAsync(Encoding.UTF8.GetBytes($"Sent from C# at '{DateTime.Now}'"));
                Console.WriteLine($"MessageId is: '{messageId}'");
            }
        }
    }
}

Parameters for Token authentication

  • ${brokerServiceURL}: the broker service URL of your Pulsar cluster.
  • ${subscription}: the name of the subscription that will determine how messages are delivered.
  • ${tenant}/${namespace}/${topic}: the full name of the topic for message production & consumption. It is a combination of the tenant name, the namespace name and the topic name.
  • ${token}: the token of your service account.
Previous
Overview