> ## Documentation Index
> Fetch the complete documentation index at: https://docs.streamnative.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Connect to your cluster using the Pulsar C# client

This document describes how to connect to a cluster using 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 StreamNative cluster using either [OAuth2](#use-oauth2) or [API Keys](#use-apikeys) authentication.

<Note title="Note">
  This document assumes that you have created a StreamNative cluster and a service account, and have granted the service account `produce` and `consume` permissions to the namespace for the target topic.

  This guide uses the open source Pulsar .NET client from the `fsprojects/pulsar-client-dotnet` repository: [fsprojects/pulsar-client-dotnet](https://github.com/fsprojects/pulsar-client-dotnet).
</Note>

## Prerequisites

Install the Pulsar .NET client package from NuGet. For package details and versions, see [Pulsar.Client on NuGet](https://www.nuget.org/packages/Pulsar.Client).

<Note>
  Choose the `Pulsar.Client` major version based on your target framework:

  * If you target **.NET 8.0 or later**, install **`Pulsar.Client` 3.x**.
  * If you don't target .NET 8.0+ and your project supports **.NET Standard 2.0**, install **`Pulsar.Client` 2.x**.
</Note>

```bash theme={null}
dotnet add package Pulsar.Client
```

## Connect to your cluster using API keys

<span id="use-apikeys" />

To connect a StreamNative cluster using [API keys](/cloud/security/authentication/service-accounts/use-api-keys/api-keys-overview), follow these steps.

### Step 1: Get the broker service URL of your cluster

To get the service URL(s) of a StreamNative cluster, follow these steps.

<Tabs>
  <Tab title="StreamNative Console">
    1. Navigate to the **Cluster Dashboard** page by [switching to the cluster workspace](/cloud/get-started/cloud-console#switch-a-cluster).

    2. On the **Cluster Dashboard** page, click **Details** tab.

    3. You will see the available service URLs in the **Access Points** area.

    4. You can click **Copy** at the end of the row of the service URL that you want to use.
  </Tab>
</Tabs>

### Step 2: Create an API key of your service account

<Note title="Note">
  Before using an API key, verify that the service account is authorized to access the resources, such as tenants, namespaces, and topics.
</Note>

You can follow the instructions to [create an API key](/cloud/security/authentication/service-accounts/use-api-keys/api-keys-overview#using-api-keys-to-connect-to-your-cluster) for the service account you choose to use.

### Step 3: Connect to your cluster

For a complete example of how to connect to a cluster using the C# client, see [C# client examples](https://github.com/streamnative/cloud-manager/tree/master/ui/src/data/code/clients/csharp).

#### 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](/cloud/build/pulsar-clients/cloud-connect-dotnet#parameters-for-token-authentication).

```csharp theme={null}
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 = "${apikey}";

            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](/cloud/build/pulsar-clients/cloud-connect-dotnet#parameters-for-token-authentication).

```csharp theme={null}
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 = "${apikey}";

            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 StreamNative 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.
* `${apikey}`: an API key of your service account.

## Connect to your cluster using OAuth2 authentication

<span id="use-oauth2" />

To connect a StreamNative cluster using OAuth2 authentication, follow these steps.

### Step 1: Get the broker service URL of your cluster

To get the service URL(s) of a StreamNative cluster, follow these steps.

<Tabs>
  <Tab title="StreamNative Console">
    1. Navigate to the **Cluster Dashboard** page by [switching to the cluster workspace](/cloud/get-started/cloud-console#switch-a-cluster).

    2. On the **Cluster Dashboard** page, click **Details** tab.

    3. You will see the available service URLs in the **Access Points** area.

    4. You can click **Copy** at the end of the row of the service URL that you want to use.
  </Tab>
</Tabs>

### 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:

   ```json theme={null}
   {
     "type": "SN_SERVICE_ACCOUNT",
     "client_id": "CLIENT_ID",
     "client_secret": "CLIENT_SECRET",
     "client_email": "test@auth.streamnative.cloud",
     "issuer_url": "https://auth.streamnative.cloud"
   }
   ```

### Step 3: Connect to your cluster

For a complete example of how to connect to a cluster using the C# client, see [C# client examples](https://github.com/streamnative/cloud-manager/tree/master/ui/src/data/code/clients/csharp).

#### 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](/cloud/build/pulsar-clients/cloud-connect-dotnet#parameters-for-oauth2-authentication).

```csharp theme={null}
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](/cloud/build/pulsar-clients/cloud-connect-dotnet#parameters-for-oauth2-authentication).

```csharp theme={null}
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)](/cloud/references/glossary#urn), which is a combination of the `urn:sn:pulsar`, your organization name, and your Pulsar instance name.
  * `${orgName}`: the name of your [organization](/cloud/references/glossary#organization).
  * `${instanceName}`: the name of your [instance](/cloud/references/glossary#instance).
* `${brokerServiceURL}`: the broker service URL of your StreamNative 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.