ApsaraMQ for MQTT:Sample code for the .NET cloud SDK

Prerequisites

Before you begin, make sure that you have:

• .NET SDK installed. For installation instructions, see Microsoft .NET documentation

• The AliyunOnsmqttServerSdk NuGet package added to your project

• An ApsaraMQ for MQTT instance created, with the instance ID and endpoint available

• A parent topic and a group ID created in the ApsaraMQ for MQTT console

• An AccessKey pair (AccessKey ID and AccessKey secret) created in the Resource Access Management (RAM) console. For more information, see Obtain an AccessKey pair

Configure the connection

Create a ChannelConfig object with your instance endpoint, instance ID, credentials, and port.

using System.Text;
using AliyunOnsmqttServerSdk;
using AliyunOnsmqttServerSdk.Common;
using AliyunOnsmqttServerSdk.Config;
using AliyunOnsmqttServerSdk.Model;

// Instance endpoint.
// Public endpoint format:  <instance-id>-server-internet.mqtt.aliyuncs.com
// VPC endpoint format:     <instance-id>-server-internal.mqtt.aliyuncs.com
string domain = "<instance-id>-server-internet.mqtt.aliyuncs.com";

// Instance ID. Find this in the Basic Information section of the Instance Details
// page in the ApsaraMQ for MQTT console.
string instanceId = "<instance-id>";

// Load credentials from environment variables to avoid hard coding.
string ak = Environment.GetEnvironmentVariable("ALIBABA_CLOUD_ACCESS_KEY_ID");
string sk = Environment.GetEnvironmentVariable("ALIBABA_CLOUD_ACCESS_KEY_SECRET");

// Parent topic created in the ApsaraMQ for MQTT console.
string firstTopic = "<your-parent-topic>";

// Group ID created in the ApsaraMQ for MQTT console.
string gid = "<your-group-id>";

// Port for the cloud SDK. Must be 5672.
int port = 5672;

ChannelConfig channelConfig = new ChannelConfig(domain, instanceId, ak, sk, port);

Replace the following placeholders with your actual values:

Subscribe to a topic

Create a ServerConsumer to subscribe to a level-1 topic and receive messages.

ServerConsumer serverConsumer = new ServerConsumer(channelConfig);
serverConsumer.Start();

// Subscribe to the parent topic. The callback fires for each received message.
serverConsumer.SubscribeTopic(firstTopic,
    (string msgId, MessageProperties messageProperties, byte[] bArr) =>
    {
        Console.WriteLine($"Received message: {msgId}");
    });

Subscribe to client online and offline events

Call SubscribeStatus on the same ServerConsumer to monitor when clients connect to or disconnect from the broker.

// Subscribe to status events for the specified group ID.
serverConsumer.SubscribeStatus(gid,
    (statusNotice) =>
    {
        Console.WriteLine($"Client status event: {statusNotice.ClientId}, {statusNotice.EventType}");
    });

Publish messages

Create a ServerProducer to publish messages to a subtopic.

ProducerConfig producerConfig = new ProducerConfig();
ServerProducer serverProducer = new ServerProducer(channelConfig, producerConfig);
serverProducer.Start();

// Encode the message payload as UTF-8.
byte[] payload = Encoding.UTF8.GetBytes("test");

try
{
    // Publish a message every second to the subtopic "firstTopic/t2".
    while (true)
    {
        Thread.Sleep(1000);
        SendResult sendResult = serverProducer.SendMessage(firstTopic + "/t2", payload);
        Console.WriteLine($"Sent message: {sendResult.MsgId}");
    }
}
catch (Exception e)
{
    Console.WriteLine(e.StackTrace);
}

Complete sample

The following program combines all previous steps into a single runnable application.

using System.Text;
using AliyunOnsmqttServerSdk;
using AliyunOnsmqttServerSdk.Common;
using AliyunOnsmqttServerSdk.Config;
using AliyunOnsmqttServerSdk.Model;

class Program
{
    static void Main(string[] args)
    {
        // Instance endpoint.
        // Public endpoint:  <instance-id>-server-internet.mqtt.aliyuncs.com
        // VPC endpoint:     <instance-id>-server-internal.mqtt.aliyuncs.com
        string domain = "<instance-id>-server-internet.mqtt.aliyuncs.com";

        // Instance ID from the Basic Information section of the Instance Details page.
        string instanceId = "<instance-id>";

        // Load credentials from environment variables to avoid hard coding.
        string ak = Environment.GetEnvironmentVariable("ALIBABA_CLOUD_ACCESS_KEY_ID");
        string sk = Environment.GetEnvironmentVariable("ALIBABA_CLOUD_ACCESS_KEY_SECRET");

        // Parent topic created in the ApsaraMQ for MQTT console.
        string firstTopic = "<your-parent-topic>";

        // Group ID created in the ApsaraMQ for MQTT console.
        string gid = "<your-group-id>";

        // Port for the cloud SDK. Must be 5672.
        int port = 5672;

        // --- Connection ---
        ChannelConfig channelConfig = new ChannelConfig(domain, instanceId, ak, sk, port);

        // --- Consumer: subscribe to a topic ---
        ServerConsumer serverConsumer = new ServerConsumer(channelConfig);
        serverConsumer.Start();

        serverConsumer.SubscribeTopic(firstTopic,
            (string msgId, MessageProperties messageProperties, byte[] bArr) =>
            {
                Console.WriteLine($"Received message: {msgId}");
            });

        // --- Consumer: subscribe to client status events ---
        serverConsumer.SubscribeStatus(gid,
            (statusNotice) =>
            {
                Console.WriteLine($"Client status event: {statusNotice.ClientId}, {statusNotice.EventType}");
            });

        // --- Producer: publish messages ---
        ProducerConfig producerConfig = new ProducerConfig();
        ServerProducer serverProducer = new ServerProducer(channelConfig, producerConfig);
        serverProducer.Start();

        byte[] payload = Encoding.UTF8.GetBytes("test");

        try
        {
            while (true)
            {
                Thread.Sleep(1000);
                SendResult sendResult = serverProducer.SendMessage(firstTopic + "/t2", payload);
                Console.WriteLine($"Sent message: {sendResult.MsgId}");
            }
        }
        catch (Exception e)
        {
            Console.WriteLine(e.StackTrace);
        }
    }
}

Parameters