ApsaraMQ for MQTT:Export data from ApsaraMQ for MQTT to ApsaraMQ for RocketMQ

import com.aliyun.openservices.lmq.example.util.ConnectionOptionWrapper;
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.Properties;
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 MQ4IoTSendMessageToRocketMQ {
    public static void main(String[] args) throws Exception {
        /**
         * Initialize the ApsaraMQ for RocketMQ client as a receiver. In most business scenarios, the receiver is deployed on a backend application. 
         */
        Properties properties = new Properties();
        /**
         * The ID of the group that you created in the ApsaraMQ for RocketMQ console. 
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-XXXXX");
        
        /**
         * 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 prevent 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 in your account may be exposed to potential security risks. 
         * In this example, the AccessKey ID and AccessKey secret are stored in 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. This parameter is required only if you use the signature authentication mode. 
         */
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
      
         /**
         * The TCP endpoint that you want to use 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, "http://xxxxx.XXXXX.mq-internet.aliyuncs.com");
        /**
         * The topic that you created in the ApsaraMQ for RocketMQ console. 
         * When you exchange data between ApsaraMQ for RocketMQ and ApsaraMQ for MQTT, only parent topics can be used by the ApsaraMQ for RocketMQ client. 
         */
        final String parentTopic = "XXXXX";
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(parentTopic, "*", new MessageListener() {
            public Action consume(Message message, ConsumeContext consumeContext) {
                System.out.println("recv msg:" + message);
                return Action.CommitMessage;
            }
        });
        consumer.start();
        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
        /**
         * Initialize the ApsaraMQ for MQTT client as a sender. In most business scenarios, the sender is deployed on a mobile terminal. 
         */

        /**
         * The ID of the ApsaraMQ for MQTT instance that you created in the console. 
         */
        String instanceId = "XXXXX";
         /**
         * The endpoint that you want to use to access the ApsaraMQ for MQTT instance. You can obtain the endpoint on the Instance Details page in the ApsaraMQ for MQTT console. 
         */
        String endPoint = "XXXXXX.mqtt.aliyuncs.com";
        /**
         * The AccessKey ID that you created in the Alibaba Cloud RAM console for identity authentication. 
         * The AccessKey pair of an Alibaba Cloud account has permissions on all API operations. To prevent 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 in your account may be exposed to potential security risks. 
         * In this example, the AccessKey ID and AccessKey secret are stored in environment variables. 
         */
        String accessKey = System.getenv("MQTT_AK_ENV");
        /**
         * The AccessKey secret that you created in the Alibaba Cloud RAM console for identity authentication. This parameter is required only if you use the signature authentication mode. 
         */
        String secretKey = System.getenv("MQTT_SK_ENV");
        /**
         * The globally unique ID that the system assigns to the ApsaraMQ for MQTT client. The client ID must vary based on the TCP connection. If multiple TCP connections use the same client ID, exceptions occur and the connections are unexpectedly closed. 
         * The client ID consists of a group ID and a device ID and is in the GroupID@@@DeviceID format. The group ID is the ID of the group that you created in the ApsaraMQ for MQTT console. The device ID is a custom ID that you specify. The client ID cannot exceed 64 characters in length. 
         */
        String clientId = "GID_XXXX@@@XXXXX";
       /**
         * ApsaraMQ for MQTT allows you to use a subtopic to filter messages. You can specify a string as the name of the subtopic. 
         * The value of the mq4IotTopic parameter can be up to 128 characters in length. 
         */
        final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";
        /**
         * The quality of service (QoS) level for message transmission. Valid values: 0, 1, and 2. For more information, see Terms. 
         */
        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. The protocol and port that are used by the ApsaraMQ for MQTT client must match. If Secure Sockets Layer (SSL) encryption is used, specify ssl://endpoint:8883 as the protocol and port. 
         */
        final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        /**
         * The timeout period during which the ApsaraMQ for MQTT client waits for a response. The timeout period prevents the ApsaraMQ for MQTT client from waiting for a response for an indefinite period of time. 
         */
        mqttClient.setTimeToWait(5000);
        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 {
            }

            @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. If the messages that are sent are normal messages, the topic must be the topic to which the consumer subscribes or the topic that can be matched by using wildcards. 
             */
            mqttClient.publish(mq4IotTopic, message);
        }
        Thread.sleep(Long.MAX_VALUE);

    }

}