IoT Platform:Subscribe to device messages with a server-side subscription

Prepare the development environment

The sample AMQP client in this topic is developed in Java. We recommend using the Apache Qpid JMS client. For usage instructions, see Qpid JMS 0.57.0.

This sample uses the following development environment:

• Operating system: Windows 10 (64-bit)

• JDK: JDK 8

• Integrated development environment (IDE): IntelliJ IDEA Community Edition

Configure an AMQP server-side subscription

1. On the Overview tab in the IoT Platform console, click IoT_Test_Environment.

2. Create a consumer group that your server will use to retrieve forwarded messages.

   1. In the left-side navigation pane, choose Message Forwarding > Server-Side Subscription, and click the Consumer Groups tab.

   2. Click Create Consumer Group.

   3. In the Create Consumer Group dialog box, set the consumer group name to Thermostat Data Receiver and click OK.

      After the consumer group is created, you are redirected to the Consumer Group Details page. Find the consumer group ID and click Copy to save the consumer group ID. You will need this ID to connect your AMQP client.

3. Configure a server-side subscription for the Home Thermostat product. This lets your server subscribe to various types of messages from devices under this product.

   1. In the left-side navigation pane, choose Message Forwarding > Server-Side Subscription.

   2. On the Server-Side Subscription page, on the Subscriptions tab, click Create Subscription.

   3. In the Create Subscription dialog box, set the parameters and click OK.

      Parameter	Description
      Product	Select Home Thermostat.
      Subscription Type	Select AMQP.
      Consumer Group	Select the Thermostat Data Receiver consumer group that you created.
      Message Type	Select Device Upstream Notification and Submit historical TSL Data. For more information about message types, see Configure an AMQP server-side subscription.

   You can also use the data forwarding feature to forward data to a consumer group of an AMQP subscription and use an AMQP client to receive messages. For example, you can forward the processed TSL Data Submission content from the Thermostat Data Forwarding data forwarding rule to the Thermostat Data Reception consumer group of the AMQP subscription. For more information, see Forward data to a consumer group of a server-side AMQP subscription.

AMQP client connection sample

1. Download and decompress the demo package.

2. Open IntelliJ IDEA and import the amqp-demo sample project from the package.

   The pom.xml file already contains the Maven dependency required to download the Qpid JMS client.

3. In the AmqpClient.java file, located in the src/main/java/com.aliyun.iotx.demo directory, modify the parameters to connect the JMS client to IoT Platform. See the following table for details.

       private static String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
       private static String accessSecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
       private static String consumerGroupId = "1yRO*****************";
       // The ID of your instance. If you are using a public instance that was activated before July 30, 2021, leave this parameter empty.
       private static String iotInstanceId = "iot-cn-******";
       // The client ID. The value of this parameter is displayed in the Client ID column on the consumer group status page in the IoT Platform console.
       // We recommend that you use a unique identifier, such as a machine UUID, MAC address, or IP address, as the client ID to easily distinguish between different clients.
       private static String clientId = "test_IoT";
       // The endpoint for your instance. For more information, see the documentation on how to connect an AMQP client.
       private static String host = "iot-cn-******.amqp.iothub.aliyuncs.com";
       // The number of connections to be established by a single process.
       // The consumption rate of a single connection is limited. You can create up to 128 connections. For more information, see Limits.
       // The number of connections is related to the consumption rate and rebalancing. We recommend that you add one connection for every 500 QPS increase in consumption rate.
       private static int connectionCount = 4;

   Parameter	Description
   accessKey	Log in to the IoT Platform console. Hover over your profile picture and click AccessKey Management to obtain your AccessKey ID and AccessKey secret. Important To prevent the security risk of hardcoding your AccessKey pair into business code, this example uses environment variables. You must create the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables in your local operating system and set their values to your AccessKey ID and AccessKey secret, respectively.
   accessSecret
   consumerGroupId	The ID of the Thermostat Data Receiver consumer group that you created. See Step 2 in the Configure an AMQP server-side subscription section of this topic.
   iotInstanceId	The ID of your IoT Test Environment instance. You can find the instance ID on the Overview page in the IoT Platform console.
   clientId	The client ID. You must define this ID. The ID can be up to 64 characters in length. We recommend that you use a unique identifier, such as the UUID, MAC address, or IP address of the server where your AMQP client is located. After the AMQP client is connected and starts, log on to the IoT Platform console. On the Consumer Groups tab of the Message Forwarding > > > Server-side Subscription page for the instance, click View next to the consumer group. The Consumer Group Details page displays this parameter. This helps you identify different clients.
   host	The AMQP endpoint for your instance. For more information, see Create an enterprise edition instance. You can also find the endpoint in the IoT Platform console. On the Instance Details page of your IoT Test Environment instance, click View Development Configurations and then click the AMQP tab.
   connectionCount	The number of connections for the AMQP client. The maximum value is 128. This parameter is used to scale out real-time message pushes. On the Consumer Group Details page, connected clients are displayed in the ${clientId}+"-"+number format. The minimum value for number is 0.

4. Run the AmqpClient.java sample code. Output similar to the following indicates the AMQP client has connected to IoT Platform and is receiving messages.

   Note

   The demo code includes the Thread.sleep(60 * 1000); code snippet, which terminates the program one minute after it starts. In your application, set the run time based on your business requirements.

   [main] INFO com.aliyun.iotx.demo.AmqpClient - amqp demo is started successfully, and will exit after 60s
   [AmqpProvider :(2):[amqps://xxx.aliyuncs.com:5671]] DEBUG org.apache.qpid.jms.provider.amqp.AmqpConsumer - Dispatching received message: JmsInboundMessageDispatch { sequence
   [JmsSession [ID:d0cd8ff1-5c3xxx] xxx dispatcher] DEBUG org.apache.qpid.jms.provider.failover.FailoverProvider - Executing Failover Task: message acknowledge -> JmsInboundMessag...
   [AmqpProvider :(2):[amqps://xxx.aliyuncs.com:5671]] DEBUG org.apache.qpid.jms.provider.amqp.AmqpConsumer - Delivered Ack of message: JmsInboundMessageDispatch { sequence =
   [JmsSession [ID:d0cd8ff1-5c3xxx] xxx dispatcher] DEBUG org.apache.qpid.jms.provider.failover.FailoverProvider - Executing Failover Task: message acknowledge -> JmsInboundMessag...
   [AmqpProvider :(2):[amqps://xxx.aliyuncs.com:5671]] DEBUG org.apache.qpid.jms.provider.amqp.AmqpConsumer - Accepted Ack of message: JmsInboundMessageDispatch { sequence = 1
   [pool-1-thread-1] INFO com.aliyun.iotx.demo.AmqpClient - receive message,
    topic = /a1xxx/device1/thing/event/property/post,
    messageId = 1380044478987006465,
    content = {"deviceType":"CustomCategory","iotId":"nTT9rxxx","requestId":"1617863381153","checkFailedData":{},"productKey":"a10xxx","gmtCreate":1617863218430,"deviceName":"device1","items":{"LightSwitch":{"value":1,"time":1617863218430}}}}}}}}

   Response parameters:

   Parameter	Value	Description
   topic	/***********/******/thing/event/property/post	The topic used to report device properties.
   messageId	1324198300680719360	The ID of the message.
   content	{"temperature":23,"humidity":21,"time":1604548451951}	The payload of the message.

   After the code runs successfully, you can view the basic status of the consumer group.

   For more information, see Manage AMQP consumer groups.

   The consumer group status includes metrics such as the real-time message consumption rate, backlogged message consumption rate, and message backlog. The online client list below the metrics shows the real-time consumption rate per connection and backlogged consumption rate per connection for each client.

View logs

After you complete the configuration, log in to the IoT Platform console. In your enterprise edition instance, navigate to Maintenance > Device Log. Select the Home Thermostat product. You can view the complete logs on the Cloud Run Log tab.

The log page displays four records for Device1. The business types are Server-Side Subscription (with an AMQP operation), Cloud-to-Device Message, TSL, and Device-to-Cloud Message. The status code for all records is 200, indicating a normal message flow.