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 {
        /**
         * 初始化云消息隊列 RocketMQ 版接收客戶端，實際業務中一般部署在服務端應用中。
         */
        Properties properties = new Properties();
        /**
         * 設置云消息隊列 RocketMQ 版Group ID，在云消息隊列 RocketMQ 版控制臺創建。
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-XXXXX");
        
        /**
         * AccessKey ID，阿里云身份驗證，在阿里云RAM控制臺創建。
         * 阿里云賬號AccessKey擁有所有API的訪問權限，建議您使用RAM用戶進行API訪問或日常運維。
         * 強烈建議不要把AccessKey ID和AccessKey Secret保存到工程代碼里，否則可能導致AccessKey泄露，威脅您賬號下所有資源的安全。
         * 本示例以將AccessKey 和 AccessKeySecret 保存在環境變量為例說明。
         */
        properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
        /**
         * AccessKey Secret，阿里云身份驗證，在阿里云RAM控制臺創建。僅在簽名鑒權模式下需要設置。
         */
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
      
         /**
         * 設置TCP接入點，該接入點為云消息隊列 RocketMQ 版實例的接入點。云消息隊列 RocketMQ 版控制臺實例詳情頁面獲取。
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "http://xxxxx.XXXXX.mq-internet.aliyuncs.com");
        /**
         * 設置云消息隊列 RocketMQ 版的Topic，在云消息隊列 RocketMQ 版控制臺創建。
         * 云消息隊列 RocketMQ 版和微消息隊列MQTT配合使用時，RocketMQ客戶端僅操作一級Topic。
         */
        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();
        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
        /**
         * 初始化云消息隊列 MQTT 版發送客戶端，實際業務中云消息隊列 MQTT 版一般部署在移動終端環境。
         */

        /**
         * 您在控制臺創建的微消息隊列MQTT的實例ID。
         */
        String instanceId = "XXXXX";
         /**
         * 設置接入點，進入微消息隊列MQTT版控制臺實例詳情頁面獲取。
         */
        String endPoint = "XXXXXX.mqtt.aliyuncs.com";
        /**
         * AccessKey ID，阿里云身份驗證，在阿里云RAM控制臺創建。
         * 阿里云賬號AccessKey擁有所有API的訪問權限，建議您使用RAM用戶進行API訪問或日常運維。
         * 強烈建議不要把AccessKey ID和AccessKey Secret保存到工程代碼里，否則可能導致AccessKey泄露，威脅您賬號下所有資源的安全。
         * 本示例以將AccessKey 和 AccessKeySecret 保存在環境變量為例說明。
         */
        String accessKey = System.getenv("MQTT_AK_ENV");
        /**
         * AccessKey Secret，阿里云身份驗證，在阿里云RAM控制臺創建。僅在簽名鑒權模式下需要設置。
         */
        String secretKey = System.getenv("MQTT_SK_ENV");
        /**
         * MQTT客戶端ID，由業務系統分配，需要保證每個TCP連接都不一樣，保證全局唯一，如果不同的客戶端對象（TCP連接）使用了相同的clientId會導致連接異常斷開。
         * clientId由兩部分組成，格式為GroupID@@@DeviceID，其中GroupID在云消息隊列 MQTT 版控制臺創建，DeviceID由業務方自己設置，clientId總長度不得超過64個字符。
         */
        String clientId = "GID_XXXX@@@XXXXX";
       /**
         * 云消息隊列 MQTT 版支持子級Topic，用來做自定義的過濾，此處為示例，可以填寫任何字符串。
         * 需要注意的是，完整的Topic長度不得超過128個字符。
         */
        final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";
        /**
         * QoS參數代表傳輸質量，可選0，1，2。詳細信息，請參見名詞解釋。
         */
        final int qosLevel = 0;
        ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();
         /**
         * 客戶端協議和端口。客戶端使用的協議和端口必須匹配，如果是SSL加密則設置ssl://endpoint:8883。
         */
        final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        /**
         * 設置客戶端發送超時時間，防止無限阻塞。
         */
        mqttClient.setTimeToWait(5000);
        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                /**
                 * 客戶端連接成功后就需要盡快訂閱需要的Topic。
                 */
                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);
            /**
             *  發送普通消息時，Topic必須和接收方訂閱的Topic一致，或者符合通配符匹配規則。
             */
            mqttClient.publish(mq4IotTopic, message);
        }
        Thread.sleep(Long.MAX_VALUE);

    }

}