ApsaraMQ for MQTT:Export status events of ApsaraMQ for MQTT clients

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 code is deployed in a backend application. 
         */
        Properties properties = new Properties();
        /**
         * Specify the group ID of the ApsaraMQ for RocketMQ client. Note that the group ID that you specify for this parameter is different from the ID of the group on the ApsaraMQ for MQTT instance. You must specify group IDs based on the corresponding service instructions. 
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID_XXXX");
        /**
         * The AccessKey ID of the Alibaba Cloud account. You can obtain the AccessKey ID from the Resource Access Management (RAM) console. 
         */
        properties.put(PropertyKeyConst.AccessKey, "XXXX");
        /**
         * The AccessKey secret of the Alibaba Cloud account. You can obtain the AccessKey secret from the RAM console. The AccessKey secret is required only if the signature authentication mode is used. 
         */
        properties.put(PropertyKeyConst.SecretKey, "XXXX");
        /**
         * Specify the TCP endpoint. 
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "http://XXXX");
        /**
         * If you use an ApsaraMQ for RocketMQ consumer to process status notifications of the ApsaraMQ for MQTT client, the consumer must subscribe to the topic that is used to receive the status notifications. You must create the topic in advance by referring to the documentation in the ApsaraMQ for RocketMQ console. 
         */
        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 in 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_XXXXxXX@@@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 all channels have disconnection events, the 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;
    }

}