ApsaraMQ for MQTT:Export client status events

The following example uses an ApsaraMQ for RocketMQ consumer to receive and process client status notifications. The consumer subscribes to connect and tcpclean tags on the designated RocketMQ topic.

For the complete source code, see MQTTClientStatusNoticeProcessDemo.java.

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 consumer to receive status notifications.
        Properties properties = new Properties();

        // The RocketMQ consumer group ID (different from the ApsaraMQ for MQTT group ID).
        properties.setProperty(PropertyKeyConst.GROUP_ID, "<your-rocketmq-group-id>");

        // AccessKey pair. Get these from the RAM console.
        properties.put(PropertyKeyConst.AccessKey, "<your-access-key>");
        properties.put(PropertyKeyConst.SecretKey, "<your-secret-key>");

        // TCP endpoint of the ApsaraMQ for RocketMQ instance.
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-rocketmq-endpoint>");

        // The RocketMQ topic that receives MQTT client status notifications.
        // Create this topic in the ApsaraMQ for RocketMQ console beforehand.
        final String parentTopic = "GID_XXXX_MQTT";

        // In production, use an external store (database or Redis) to persist status data
        // across application restarts.
        MqttClientStatusStore mqttClientStatusStore = new MemoryHashMapStoreImpl();

        Consumer consumer = ONSFactory.createConsumer(properties);

        // Subscribe to "connect" and "tcpclean" events only.
        consumer.subscribe(parentTopic, "connect||tcpclean",
            new MqttClientStatusNoticeListener(mqttClientStatusStore));
        consumer.start();

        // Poll the status of a specific client.
        String clientId = "GID_XXXXxXX@@@XXXXX";
        while (true) {
            System.out.println("Client online: " + checkClientOnline(clientId, mqttClientStatusStore));
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * Processes incoming status notifications and maintains the channel-event map.
     *
     * In a multi-server deployment, store client status in a shared external store
     * (database or Redis). Implement idempotent consumption to handle 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"));

                // Store the new event.
                mqttClientStatusStore.addEvent(clientId, channelId, eventType, event);

                // Read the event list for this connection.
                Set<ClientStatusEvent> events = mqttClientStatusStore.getEvent(clientId, channelId);
                if (events == null || events.isEmpty()) {
                    return Action.CommitMessage;
                }

                // If both connect and tcpclean events exist for this channelId,
                // the connection is closed. Clean up the entry.
                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 whether the client has an active TCP connection.
     *
     * Logic:
     * 1. If the channel map is empty, the client is offline.
     * 2. If any channelId has only a connect event (no tcpclean), the client is online.
     * 3. If all channelIds have both connect and tcpclean events, the client is offline.
     */
    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;
    }

}

Replace the following placeholders with actual values: