Push events from ApsaraMQ for MQTT to Function Compute

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.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * The following sample code provides an example on how an ApsaraMQ for MQTT client object sends messages to itself in signature authentication mode. Modify the parameters for client initialization based on your business requirements. 
 * In signature authentication mode, the AccessKey ID and AccessKey secret that are obtained from the Resource Access Management (RAM) console are used to calculate an independent signature for the authentication of the client object. 
 * In actual business scenarios, you can complete the signature process in a trusted environment because the AccessKey secret is private. 
 *
 * For information about the complete demo project, visit https://github.com/AliwareMQ/lmq-demo. 
 */
public class MQ4IoTProducerDemo {
    public static void main(String[] args) throws Exception {
        /**
         * The ID of the ApsaraMQ for MQTT instance. You can obtain the ID from the ApsaraMQ for MQTT console after you purchase the instance. 
         */
        String instanceId = "XXXXX";
        /**
         * The endpoint of the ApsaraMQ for MQTT instance. You can obtain the endpoint from the ApsaraMQ for MQTT console after you purchase and configure the instance. You must use the assigned domain name instead of the IP address to connect to the instance. Otherwise, an exception may occur on the client object. 
         */
        String endPoint = "XXXXX.mqtt.aliyuncs.com";
        /**
         * The AccessKey ID. You can obtain the AccessKey ID from the RAM console. 
         */
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        /**
         * The AccessKey secret. You can obtain the AccessKey secret from the RAM console. The secretKey parameter is required only in signature authentication mode. 
         */
        String secretKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        /**
         * The ID that the system assigns to the ApsaraMQ for MQTT client object. Each TCP connection must use a unique client ID. If different TCP connections use the same client ID, an exception occurs and the connections are closed. 
         * The value of the clientId parameter is in the GroupID@@@DeviceID format. GroupID is the ID of the consumer group that you created in the ApsaraMQ for MQTT console, and DeviceID is the custom ID of the device. The value of the clientId parameter can be up to 64 characters in length. 
         */
        String clientId = "GID_XXXXX@@@XXXXX";
        /**
         * The parent topic that you created in the ApsaraMQ for MQTT console. 
         * If you specify a topic that does not exist or a topic that the ApsaraMQ for MQTT client object is not authorized to access, the ApsaraMQ for MQTT broker closes the connection. 
         */
        final String parentTopic = "XXXXX";
        /**
         * ApsaraMQ for MQTT allows you to use a subtopic to filter messages. You can specify a string as a subtopic name. 
         */
        final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";
        /**
         * The quality of service (QoS) level in message transmission. Valid values: 0, 1, and 2. 
         */
        final int qosLevel = 0;
        ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();
        /**
         * The protocol and port that are used by the ApsaraMQ for MQTT client object must match. 
         * If Secure Sockets Layer (SSL) encryption is enabled, the protocol and port must be specified as ssl://endpoint:8883. 
         */
        final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        /**
         * The timeout period for the ApsaraMQ for MQTT client object to wait for a response. After the timeout period ends, the client object no longer waits for a response. 
         */
        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) {
                /**
                 * The topic to which the consumer must subscribe at the earliest opportunity after the client connection is established. 
                 */
                System.out.println("connect success");
            }

            @Override
            public void connectionLost(Throwable throwable) {
                throwable.printStackTrace();
            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                /**
                 * The callback that is invoked to consume messages. Make sure that the callback does not throw exceptions. If a response is returned for the callback, the messages are consumed. 
                 * Messages must be consumed in the specified period. If specific messages are not consumed within the timeout period specified by the ApsaraMQ for MQTT broker, the broker may attempt to resend the messages. Make sure that deduplication is performed to ensure idempotence for message consumption. For information about the specified timeout period, see Limits. 
                 */
                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());
        for (int i = 0; i < 10; i++) {
            MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
            message.setQos(qosLevel);
            /**
             * The topic to which messages are sent. This topic must be the one to which the consumer subscribes or can be matched by using wildcards. 
             */
            mqttClient.publish(mq4IotTopic, message);
            /**
             * ApsaraMQ for MQTT supports point-to-point (P2P) messaging. If the producer sends a message to an only consumer and obtains the value of the clientId parameter, a P2P message is sent. 
             * In P2P messaging, the consumer does not need to subscribe to the topic to which the producer sends messages. This way, the consumption logic is simplified. In P2P messaging, a topic must be specified in the {{parentTopic}}/p2p/{{targetClientId}} format. 
             */
            String receiverId = "xxx";
            final String p2pSendTopic = parentTopic + "/p2p/" + receiverId;
            message = new MqttMessage("hello mq4Iot p2p msg".getBytes());
            message.setQos(qosLevel);
            mqttClient.publish(p2pSendTopic, message);
        }
        Thread.sleep(Long.MAX_VALUE);
    }
}