This topic describes how to use the .NET AMQP SDK to access Alibaba Cloud IoT Platform and receive messages subscribed.

Development environment

The following table describes the development environment.

Framework Supported version
.NET Framework 3.5, 4.0, 4.5, and later
.NET Micro Framework 4.2 and later
.NET nanoFramework 1.0 and later
.NET Compact Framework 3.9 and later
.Net Core on Windows 10 and Ubuntu 14.04 1.0 and later
Mono on Linux N/A

Download the SDK

We recommend that you use the AMQP.Net Lite library for the .NET AMQP SDK. You can visit AMQP.Net Lite to download the library and read its instructions.

Add the dependency

Add the following dependency to the packages.config file:

<Packages>
  <package id="AMQPNetLite" version="2.2.0" targetFramework="net47" />
</packages>

Sample code

For information about parameters in the following demo, see AMQP client access instructions.

using System;
using System.Text;
using Amqp;
using Amqp.Sasl;
using Amqp.Framing;
using System.Threading;
using System.Security.Cryptography.X509Certificates;
using System.Net.Security;
using System.Security.Cryptography;

namespace amqp
{
    class MainClass
    {
        static string Host = "${uid}.iot-amqp.${regionId}.aliyuncs.com";
        static int Port = 5671;
        static string AccessKey = "${YourAccessKey}";
        static string AccessSecret = "${YourAccessSecret}";
        static string consumerGroupId = "${YourConsumerGroupId}";
        //user-defined client id
        static string clientId = "${YourClientId}";

        static int Count = 0;
        static int IntervalTime = 10000;

        static Address address;

        public static void Main(string[] args)
        {
            long timestamp = GetCurrentMilliseconds();
            string param = "authId=" + AccessKey + "&timestamp=" + timestamp;
            string password = doSign(param, AccessSecret, "HmacMD5");
            string username = clientId + "|authMode=aksign,signMethod=hmacmd5,consumerGroupId=" + consumerGroupId
                + ",authId=" + AccessKey + ",timestamp=" + timestamp + "|";

            DoConnectAmqp(username, password);

            Thread.Sleep(1000000);
        }

        static void DoConnectAmqp(string username, string password)
        {
            address = new Address(Host, Port, username, password);
            //Create Connection
            ConnectionFactory cf = new ConnectionFactory();
            //use local tls if neccessary
            //cf.SSL.ClientCertificates.Add(GetCert());
            //cf.SSL.RemoteCertificateValidationCallback = ValidateServerCertificate;
            cf.SASL.Profile = SaslProfile.Anonymous;
            cf.AMQP.IdleTimeout = 120000;
            cf.AMQP.ContainerId = "client.1.2";
            cf.AMQP.HostName = "contoso.com";
            cf.AMQP.MaxFrameSize = 8 * 1024;
            var connection = cf.CreateAsync(address).Result;

            //Connection Exception Closed
            connection.AddClosedCallback(ConnClosed);

            //Receive Message
            DoReceive(connection);
        }

        static void DoReceive(Connection connection)
        {
            //Create Session
            var session = new Session(connection);

            //Create Link and Receive Message
            var receiver = new ReceiverLink(session, "queueName", null);
            while (true)
            {
                receiver.Start(20, (link, message) =>
                {
                    object messageId = message.ApplicationProperties["messageId"];
                    object topic = message.ApplicationProperties["topic"];
                    string body = Encoding.UTF8.GetString((Byte[])message.Body);
                    Console.WriteLine("receive message, topic=" + topic + ", messageId=" + messageId + ", body=" + body);

                    //Acknowledge Message
                    link.Accept(message);
                });
            }
        }

        static void ConnClosed(IAmqpObject _, Error e)
        {
            Console.WriteLine("ocurr error: " + e);
            if(Count < 3)
            {
                Count += 1;
                Thread.Sleep(IntervalTime * Count);
            }
            else
            {
                Thread.Sleep(120000);
            }

            //Reconnection
            DoConnectAmqp(address.User, address.Password);
        }

        static X509Certificate GetCert()
        {
            string certPath = Environment.CurrentDirectory + "/root.crt";
            X509Certificate crt = new X509Certificate(certPath);

            return crt;
        }

        static bool ValidateServerCertificate(object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
        {
            return true;
        }

        static long GetCurrentMilliseconds()
        {
            DateTime dt1970 = new DateTime(1970, 1, 1);
            DateTime current = DateTime.Now;
            return (long)(current - dt1970).TotalMilliseconds;
        }

        static string doSign(string param, string accessSecret, string signMethod)
        {
            //signMethod = HmacMD5
            byte[] key = Encoding.UTF8.GetBytes(accessSecret);
            byte[] signContent = Encoding.UTF8.GetBytes(param);
            var hmac = new HMACMD5(key);
            byte[] hashBytes = hmac.ComputeHash(signContent);
            return Convert.ToBase64String(hashBytes);
        }
    }
}