ApsaraMQ for MQTT:Send client status notification messages from ApsaraMQ for MQTT to ApsaraMQ for RocketMQ

Network access

Process

The following figure shows the process of sending and receiving status notification messages about an ApsaraMQ for MQTT client.

Prerequisites

• The integrated development environment (IDE) is installed. For more information, see IDE. You can use IntelliJ IDEA or Eclipse. In the example, IntelliJ IDEA is used.

• The Java Development Kit (JDK) is installed. For more information, see JDK.

• An ApsaraMQ for RocketMQ instance is created, and a topic and a group are created on the instance. For more information, see Step 2: Create resources.

Step 1: Create an ApsaraMQ for MQTT instance and obtain the endpoint of the instance

1. Log on to the ApsaraMQ for MQTT console. In the left-side navigation pane, click Instances.

2. In the top navigation bar, select the region where the instance that you want to manage resides. Then, in the upper-left corner, click Create Instance.

3. In the panel that appears, use the default value Subscription for the Billing Method parameter. Then, Click OK.

4. On the buy page that appears, select the instance specifications that you want to purchase based on your business requirements, select the terms of service, and then click Buy Now.

   For information about the instance editions provided by ApsaraMQ for MQTT and their feature differences, see Instance editions.

5. On the order payment page, follow the on-screen instructions to complete the payment.

6. After you complete the payment, click Console on the page that appears.

7. Go back to the ApsaraMQ for MQTT console. In the left-side navigation pane, click Instances. In the top navigation bar, select the region in which your instance is deployed.

8. On the Instances page, click the name of the instance or click Details in the Actions column of the instance to go to the Instance Details page.

9. On the Instance Details page, click the Endpoints tab. On the tab, you can view the endpoint information. In the example, the public endpoint is used.

   

Step 2: Create a parent topic

The MQTT protocol supports multi-level topics. You must create a parent topic in the ApsaraMQ for MQTT console. You can specify subtopics in the code without the need to create them in the console. A parent topic and its subtopics are named in the following format: <Name of a parent topic>/<Name of a level-2 topic>/<Name of a level-3 topic>. A parent topic and its subtopics are separated by forward slashes (/). Example: SendMessage/demo/producer. The total length of the names of a parent topic and its subtopics cannot exceed 64 characters. For more information about topics, see Terms.

1. Log on to the ApsaraMQ for MQTT console. In the left-side navigation pane, click Instances.

2. In the top navigation bar, select the region where the instance that you want to manage resides. On the Instances page, click the instance name to go to the Instance Details page.

3. In the left-side navigation pane, click Topics. In the upper-left corner of the Topics page, click Create Topic.
4. In the Create Topic panel, set the Name and Description parameters for the topic and click OK in the lower-left corner.

Step 3: Create a group ID

For more information about group IDs, see Terms.

1. Log on to the ApsaraMQ for MQTT console. In the left-side navigation pane, click Instances.

2. In the top navigation bar, select the region where the instance that you want to manage resides. On the Instances page, click the instance name to go to the Instance Details page.

3. In the left-side navigation pane, click Groups. In the upper-left corner of the Groups page, click Create Group.
4. In the Create Group panel, set the Group ID parameter and click OK.

Step 4: Create a rule for client status notifications

The parameter values that you set in the rule must be the same as those that correspond to the resources you create.

1. Log on to the ApsaraMQ for MQTT console. In the left-side navigation pane, click Instances.

2. In the top navigation bar, select the region where the instance that you want to manage resides. On the Instances page, click the instance name to go to the Instance Details page.

3. In the left-side navigation pane, click Rules. In the upper-left corner of the Rules page, click Create Rule.

4. On the Create Rule page, perform the following steps:

   1. In the Configure Basic Information step, configure the following parameters and click Next.

      Parameter	Example	Description
      Rule ID	111111	The ID of the rule. An ID is a globally unique identifier of a rule. A rule ID can contain only letters, digits, hyphens (-), and underscores (_) and must contain at least one letter or digit. A rule ID must be 3 to 64 characters in length. If the value contains more than 64 characters, it is automatically truncated. The ID of a rule cannot be updated after the rule is created.
      Description	migrate from rocketmq	The description of the rule.
      Status	Enable	Specifies whether to enable the current rule. Valid values: Enable Disable
      Rule Type	Client Status Notification	The type of the rule. Valid values: Data Outbound: The rule is used to export data from ApsaraMQ for MQTT to other Alibaba Cloud services. For more information, see Export data from ApsaraMQ for MQTT to other Alibaba Cloud services. Data Inbound: The rule is used to import data from other Alibaba Cloud services to ApsaraMQ for MQTT. For more information, see Import data from other cloud services to ApsaraMQ for MQTT. Client Status Notification: The rule is used to export the data for online and offline events of ApsaraMQ for MQTT clients to other Alibaba Cloud services. For more information, see Export online and offline events of Message Queue for MQTT clients.

   2. In the Configure Rule Source step, specify the data source and click Next.

      Parameter	Example	Description
      Group ID	GID_Client_Status	The group ID of the devices from which the event data is exported. For information about group IDs, see Terms.

   3. In the Configure Rule Destination step, specify the destination to which the data is forwarded and click Create.

      Parameter	Example	Description
      Destination Service Type	Message Queue for Apache RocketMQ	The cloud service to which you want to forward the event data of a ApsaraMQ for MQTT client. Note Only ApsaraMQ for RocketMQ is supported.
      Message Queue for Apache RocketMQ Instance	MQ_INST_13801563067*****_BbyOD2jQ	The ID of the ApsaraMQ for RocketMQ instance to which the data is forwarded. Note You can select only an instance that resides in the same region as the ApsaraMQ for MQTT instance.
      Topic	TopicB	The ApsaraMQ for RocketMQ topic to which the data is forwarded. In this example, the notifications about the online or offline events of a ApsaraMQ for MQTT client are forwarded to Topic B.

   You can view the client status notification rule that you create on Rules page.

Step 5: Send and receive messages by using the SDK for Java

1. Download a third-party open source SDK for Java. Download link: Eclipse Paho Java Client.

2. Download the demo of the ApsaraMQ for MQTT SDK for Java for reference during code development. Download link: mqtt-java-demo.

3. Decompress the demo project package to a specific folder.

4. In IntelliJ IDEA, import the extracted files to create a project and check whether the pom.xml file contains the following dependencies:

   <dependencies>
           <dependency>
               <groupId>commons-codec</groupId>
               <artifactId>commons-codec</artifactId>
               <version>1.10</version>
           </dependency>
           <dependency>
               <groupId>org.eclipse.paho</groupId>
               <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
               <version>1.2.2</version>
           </dependency>
           <dependency>
               <groupId>org.apache.httpcomponents</groupId>
               <artifactId>httpclient</artifactId>
               <version>4.5.2</version>
           </dependency>
           <dependency>
               <groupId>com.alibaba</groupId>
               <artifactId>fastjson</artifactId>
               <version>1.2.83</version>
           </dependency>
           <dependency>
               <groupId>com.aliyun.openservices</groupId>
               <artifactId>ons-client</artifactId>
               <version>1.8.5.Final</version>
           </dependency>
           <dependency>
               <groupId>com.aliyun</groupId>
               <artifactId>aliyun-java-sdk-onsmqtt</artifactId>
               <version>1.0.3</version>
           </dependency>
           <dependency>
               <groupId>com.aliyun</groupId>
               <artifactId>aliyun-java-sdk-core</artifactId>
               <version>4.5.0</version>
           </dependency>
   </dependencies>

   Note

   For information about the ons-client versions of the SDK for Java, see Release notes.

5. In the MQTTClientStatusNoticeProcessDemo.java class, specify parameter values based on instructions provided in the comments. Most parameter values correspond to the ApsaraMQ for MQTT resources that you created from Step 1 to Step 3 and the resources that you created in ApsaraMQ for RocketMQ. Then, execute the main() function to send and receive messages.

   Sample code:

   Note Before you use the sample code to send and receive messages, you must configure environment variables to obtain the credentials that are used to access ApsaraMQ for MQTT. For information about how to configure environment variables, see Configure an access credential.

   The environment variable name of the AccessKey ID that is used to access ApsaraMQ for MQTT is MQTT_AK_ENV, and the environment variable name of the AccessKey secret that is used to access ApsaraMQ for MQTT is MQTT_SK_ENV.

   package com.aliyun.openservices.lmq.example.demo;
   
   import com.alibaba.fastjson.JSON;
   import com.alibaba.fastjson.JSONObject;
   import com.aliyun.openservices.ons.api.Action;
   import com.aliyun.openservices.ons.api.ConsumeContext;
   import com.aliyun.openservices.ons.api.Consumer;
   import com.aliyun.openservices.ons.api.Message;
   import com.aliyun.openservices.ons.api.MessageListener;
   import com.aliyun.openservices.ons.api.ONSFactory;
   import com.aliyun.openservices.ons.api.PropertyKeyConst;
   import java.util.Map;
   import java.util.Properties;
   import java.util.Set;
   
   public class MQTTClientStatusNoticeProcessDemo {
       public static void main(String[] args) {
           /**
            * Initialize an ApsaraMQ for RocketMQ client as a receiver. In most business scenarios, the receiver is deployed in a backend application. 
            */
           Properties properties = new Properties();
           /**
            * The group ID that you created in the ApsaraMQ for RocketMQ console. 
            */
           properties.setProperty(PropertyKeyConst.GROUP_ID, "GID_XXXX");
           /**
            * The AccessKey ID that you created in the Alibaba Cloud Resource Access Management (RAM) console for identity authentication. 
            * The AccessKey pair of an Alibaba Cloud account has permissions on all API operations. To avoid security risks, we recommend that you use a RAM user to call API operations or perform routine O&M. 
            * We strongly recommend that you do not save an AccessKey pair in the project code. Otherwise, the AccessKey pair may be leaked and all resources that are contained in your account may be exposed to potential security risks. 
            * In this example, the AccessKey pair is saved in the environment variables. 
            */
           properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
           /**
            * The AccessKey secret that you created in the Alibaba Cloud RAM console for identity authentication. The AccessKey secret is required only when the signature authentication mode is used. 
            */
           properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
           /**
            * The TCP endpoint that is used to access the ApsaraMQ for RocketMQ instance. You can obtain the TCP endpoint on the Instance Details page in the ApsaraMQ for RocketMQ console. 
            */
           properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
           /**
            * Specify a topic that you want to use to receive online or offline events when you use the ApsaraMQ for RocketMQ client to receive client status notification messages about an ApsaraMQ for MQTT client. 
            */
           final String parentTopic = "GID_XXXX_MQTT";
           /**
            * The client status data. In a production environment, we recommend that you use an external persistent storage system, such as a database or Redis system, to store status data to prevent the loss of status data upon application restart. In this example, the status data is stored on an on-premises machine. 
            */
           MqttClientStatusStore mqttClientStatusStore = new MemoryHashMapStoreImpl();
           Consumer consumer = ONSFactory.createConsumer(properties);
           /**
            * The following code determines only whether the client is online. Therefore, you need to only pay attention to the connect and tcpclean events. 
            */
           consumer.subscribe(parentTopic, "connect||tcpclean", new MqttClientStatusNoticeListener(mqttClientStatusStore));
           consumer.start();
           String clientId = "GID_XXXXX@@@XXXXX";
           while (true) {
               System.out.println("ClientStatus :" + checkClientOnline(clientId, mqttClientStatusStore));
               try {
                   Thread.sleep(1000);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
       }
   
       /**
        * The logic that is used to process client status notifications. 
        * During an actual deployment process, applications that consume status notification messages may be deployed on multiple servers. Therefore, the status data of clients can be maintained in external shared storage such as a database or Redis system. 
        * If a state machine repeatedly receives messages, perform consumption idempotence processing on messages to prevent errors that may occur. 
        */
       static class MqttClientStatusNoticeListener implements MessageListener {
           private MqttClientStatusStore mqttClientStatusStore;
   
           public MqttClientStatusNoticeListener(
               MqttClientStatusStore mqttClientStatusStore) {
               this.mqttClientStatusStore = mqttClientStatusStore;
           }
   
           @Override
           public Action consume(Message message, ConsumeContext context) {
               try {
                   JSONObject msgBody = JSON.parseObject(new String(message.getBody()));
                   System.out.println(msgBody);
                   String eventType = msgBody.getString("eventType");
                   String clientId = msgBody.getString("clientId");
                   String channelId = msgBody.getString("channelId");
                   ClientStatusEvent event = new ClientStatusEvent();
                   event.setChannelId(channelId);
                   event.setClientIp(msgBody.getString("clientIp"));
                   event.setEventType(eventType);
                   event.setTime(msgBody.getLong("time"));
                   /**
                    * Store new events first. 
                    */
                   mqttClientStatusStore.addEvent(clientId, channelId, eventType, event);
                   /**
                    * Read the event list of the current connection. 
                    */
                   Set<ClientStatusEvent> events = mqttClientStatusStore.getEvent(clientId, channelId);
                   if (events == null || events.isEmpty()) {
                       return Action.CommitMessage;
                   }
                   /**
                    * If all online and offline events in the list are received and the current connection is closed, the data of the connection can be cleared. 
                    */
                   boolean findOnlineEvent = false;
                   boolean findOfflineEvent = false;
                   for (ClientStatusEvent clientStatusEvent : events) {
                       if (clientStatusEvent.isOnlineEvent()) {
                           findOnlineEvent = true;
                       } else {
                           findOfflineEvent = true;
                       }
                   }
                   if (findOnlineEvent && findOfflineEvent) {
                       mqttClientStatusStore.deleteEvent(clientId, channelId);
                   }
                   return Action.CommitMessage;
               } catch (Throwable e) {
                   e.printStackTrace();
               }
               return Action.ReconsumeLater;
           }
       }
   
       /**
        * Check whether an active TCP connection is present for an ApsaraMQ for MQTT client based on the channel table. 
        * 1. If the channel table is empty, the ApsaraMQ for MQTT client is offline. 
        * 2. If the channel table is not empty, check whether only online events are received in a connection. If yes, an active connection is present and the ApsaraMQ for MQTT client is online. 
        * If offline events are received in all connections, the ApsaraMQ for MQTT client is offline. 
        *
        * @param clientId
        * @param mqttClientStatusStore
        * @return
        */
       public static boolean checkClientOnline(String clientId,
           MqttClientStatusStore mqttClientStatusStore) {
           Map<String, Set<ClientStatusEvent>> channelMap = mqttClientStatusStore.getEventsByClientId(clientId);
           if (channelMap == null) {
               return false;
           }
           for (Set<ClientStatusEvent> events : channelMap.values()) {
               boolean findOnlineEvent = false;
               boolean findOfflineEvent = false;
               for (ClientStatusEvent event : events) {
                   if (event.isOnlineEvent()) {
                       findOnlineEvent = true;
                   } else {
                       findOfflineEvent = true;
                   }
               }
               if (findOnlineEvent & !findOfflineEvent) {
                   return true;
               }
           }
           return false;
       }
   
   }

Verify the results

After the producer sends a message and the consumer subscribes to the message, you can query the trace of the message in the ApsaraMQ for MQTT console to verify whether the message is sent and received. For more information, see Query message traces.

References

• Use ApsaraMQ for MQTT SDK for Java to implement messaging between ApsaraMQ for MQTT clients

• Import data from other Alibaba Cloud services to ApsaraMQ for MQTT

• Export data from ApsaraMQ for MQTT to other Alibaba Cloud services