ApsaraMQ for MQTT:Client status notification rules

import com.aliyun.openservices.lmq.example.util.ConnectionOptionWrapper;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class MQ4IoTSendMessageToMQ4IoTUseSignatureMode {
    public static void main(String[] args) throws Exception {
        /**
         * The ID of the ApsaraMQ for MQTT instance. Obtain this from the console after you purchase the instance.
         */
        String instanceId = "XXXXX";
        /**
         * The endpoint. Obtain the endpoint after you purchase and configure an ApsaraMQ for MQTT instance. 
         * You must use the assigned domain name for the endpoint. Using a direct IP address can cause client exceptions.
         */
        String endPoint = "XXXXX.mqtt.aliyuncs.com";
        /**
         * The AccessKey ID of your Alibaba Cloud account. Obtain this from the console.
         * The AccessKey pair of an Alibaba Cloud account has permissions on all API operations. We recommend that you use a RAM user for API access or routine O&M to prevent security risks.
         * We strongly recommend that you do not hardcode your AccessKey ID and AccessKey secret in your project code. This can lead to security risks if the credentials are leaked.
         * This example shows how to store the AccessKey ID and AccessKey secret in environment variables. Before running this code, configure the MQTT_AK_ENV and MQTT_SK_ENV environment variables.
         * For example: export MQTT_AK_ENV=<access_key_id>
         *      export MQTT_SK_ENV=<access_key_secret>
         * Replace <access_key_id> with your AccessKey ID and <access_key_secret> with your AccessKey secret.
         */
        String accessKey = System.getenv("MQTT_AK_ENV");
        /**
         * The AccessKey secret of your Alibaba Cloud account. Obtain this from the console. This is required only for signature-based authentication.
         */
        String secretKey = System.getenv("MQTT_SK_ENV");
        /**
         * The client ID for ApsaraMQ for MQTT, which is assigned by the business system. Each TCP connection must have a unique client ID. 
         * If different client objects (TCP connections) use the same client ID, connection exceptions will occur.
         * The client ID consists of two parts in the GroupID@@@DeviceId format. The group ID is obtained from the ApsaraMQ for MQTT console, and the DeviceId is set by your business. The total length of the client ID cannot exceed 64 characters.
         */
        String clientId = "GID_XXXXX@@@XXXXX";
        ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();
        /**
         * The protocol and port used by the client must match. For more information, see https://www.alibabacloud.com/help/en/message-queue-for-mqtt/latest/limits
         * For SSL encryption, set the endpoint to ssl://endpoint:8883.
         */
        final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        /**
         * Set a timeout for the client to prevent it from blocking indefinitely.
         */
        mqttClient.setTimeToWait(5000);
        final ExecutorService executorService = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>());
        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println("connect success");
            }
            @Override
            public void connectionLost(Throwable throwable) {
                throwable.printStackTrace();
            }
            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                System.out.println("receive msg from topic " + s + " , body is " + new String(mqttMessage.getPayload()));
            }
            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                System.out.println("send msg succeed topic is : " + iMqttDeliveryToken.getTopics()[0]);
            }
        });
        mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());
        Thread.sleep(Long.MAX_VALUE);
    }
}

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 the ApsaraMQ for RocketMQ receiving client. In a real-world scenario, this is typically deployed in a backend application.
         */
        Properties properties = new Properties();
        /**
         * Set the ApsaraMQ for RocketMQ group ID. Create this in the ApsaraMQ for RocketMQ console.
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID_XXXX");
        /**
         * The AccessKey ID for Alibaba Cloud authentication. Create this in the Alibaba Cloud RAM console.
         * The AccessKey pair of an Alibaba Cloud account has permissions on all API operations. We recommend that you use a RAM user for API access or routine O&M to prevent security risks.
         * We strongly recommend that you do not hardcode your AccessKey ID and AccessKey secret in your project code. This can lead to security risks if the credentials are leaked.
         * This example shows how to store the AccessKey ID and AccessKey secret in environment variables.
         */
        properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
        /**
         * The AccessKey secret for Alibaba Cloud authentication. Create this in the Alibaba Cloud RAM console. This is required only for signature-based authentication.
         */
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
        /**
         * Set the TCP endpoint for the ApsaraMQ for RocketMQ instance. Obtain this from the instance details page in the ApsaraMQ for RocketMQ console.
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
        /**
         * When using an ApsaraMQ for RocketMQ consumer to process MQTT client status notifications, subscribe to the status notification topic.
         */
        final String parentTopic = "GID_XXXX_MQTT";
        /**
         * Client status data. In a production environment, we recommend using an external persistent store like a database or Redis to save this information.
         * This prevents data loss if the application restarts. This example uses a simple in-memory implementation.
         */
        MqttClientStatusStore mqttClientStatusStore = new MemoryHashMapStoreImpl();
        Consumer consumer = ONSFactory.createConsumer(properties);
        /**
         * This example only processes whether the client is online, so we only need to be concerned with '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 for processing connection and disconnection notifications.
     * In a real deployment, applications that consume status notifications may be deployed on multiple machines.
     * Therefore, client online status data should be maintained in a shared external store like a database or Redis.
     * Additionally, you should handle message idempotence to prevent state machine errors from duplicate messages.
     */
    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"));
                /**
                 * First, store the new event.
                 */
                mqttClientStatusStore.addEvent(clientId, channelId, eventType, event);
                /**
                 * Read the event list for the current channel.
                 */
                Set<ClientStatusEvent> events = mqttClientStatusStore.getEvent(clientId, channelId);
                if (events == null || events.isEmpty()) {
                    return Action.CommitMessage;
                }
                /**
                 * If both the online and offline events have been received in the event list, the current channel is disconnected.
                 * You can then clear the data for this channel.
                 */
                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;
        }
    }
    /**
     * Checks if a client ID has an active TCP connection based on the status table.
     * 1. If there is no channel map, the client is definitely offline.
     * 2. If the channel map is not empty, check if the channel data contains only an online event. If so, it means there is an active connection, and the client is online.
     * If all channels have disconnection events, the client is definitely 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;
    }
}