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

The sample code initializes a RocketMQ consumer that subscribes to a topic, creates an MQTT client that publishes 10 test messages to a subtopic, and relies on the data outbound rule to forward these messages to the RocketMQ topic for consumption.

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 {
        // ============================================================
        // 1. Initialize the RocketMQ consumer (backend application)
        // ============================================================
        Properties properties = new Properties();

        // RocketMQ group ID
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-XXXXX");

        // Load credentials from environment variables
        properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));

        // RocketMQ TCP endpoint (from the Instance Details page)
        properties.put(PropertyKeyConst.NAMESRV_ADDR,
            "http://xxxxx.XXXXX.mq-internet.aliyuncs.com");

        // RocketMQ topic (must be a parent topic for MQTT-RocketMQ data exchange)
        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();

        // ============================================================
        // 2. Initialize the MQTT client (device or mobile app)
        // ============================================================

        // MQTT instance ID
        String instanceId = "XXXXX";

        // MQTT endpoint (from the Instance Details page)
        String endPoint = "XXXXXX.mqtt.aliyuncs.com";

        // Load credentials from environment variables
        String accessKey = System.getenv("MQTT_AK_ENV");
        String secretKey = System.getenv("MQTT_SK_ENV");

        // Client ID: GroupID@@@DeviceID (must be unique per TCP connection)
        String clientId = "GID_XXXX@@@XXXXX";

        // Subtopic: parentTopic + "/" + subtopic name (max 128 characters)
        final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";

        // QoS level: 0, 1, or 2
        final int qosLevel = 0;

        ConnectionOptionWrapper connectionOptionWrapper =
            new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();

        // Protocol and port: tcp://endpoint:1883 or ssl://endpoint:8883
        final MqttClient mqttClient =
            new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);

        // Client timeout in milliseconds
        mqttClient.setTimeToWait(5000);
        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 {
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                System.out.println("send msg succeed topic is : "
                    + iMqttDeliveryToken.getTopics()[0]);
            }
        });
        mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());

        // ============================================================
        // 3. Publish 10 test messages
        // ============================================================
        for (int i = 0; i < 10; i++) {
            MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
            message.setQos(qosLevel);
            mqttClient.publish(mq4IotTopic, message);
        }
        Thread.sleep(Long.MAX_VALUE);
    }
}