✨ 测试阿里云 iot。

mica-mqtt-codec/src/main/java/net/dreamlu/iot/mqtt/codec/MqttSubscriptionOption.java

@@ -13,6 +13,7 @@
  * License for the specific language governing permissions and limitations
  * under the License.
  */
+
 package net.dreamlu.iot.mqtt.codec;
 
 /**
@@ -20,9 +21,21 @@ package net.dreamlu.iot.mqtt.codec;
  */
 public final class MqttSubscriptionOption {
 
+	/**
+	 * 保留处理政策
+	 */
 	public enum RetainedHandlingPolicy {
+		/**
+		 * 订阅发送
+		 */
 		SEND_AT_SUBSCRIBE(0),
+		/**
+		 * 如果还没有订阅，请发送
+		 */
 		SEND_AT_SUBSCRIBE_IF_NOT_YET_EXISTS(1),
+		/**
+		 * 请勿发送订阅
+		 */
 		DONT_SEND_AT_SUBSCRIBE(2);
 
 		private final int value;

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/DefaultMqttClientProcessor.java

@@ -31,6 +31,7 @@ import java.nio.ByteBuffer;
  */
 public class DefaultMqttClientProcessor implements MqttClientProcessor {
 	private static final Logger logger = LoggerFactory.getLogger(DefaultMqttClientProcessor.class);
+	protected static final String MQTT_CONNECTED_KEY = "__MQTT_CONNECTED_KEY__";
 	private final MqttClientSubManage subManage;
 
 	public DefaultMqttClientProcessor(MqttClientSubManage subManage) {
@@ -42,6 +43,8 @@ public class DefaultMqttClientProcessor implements MqttClientProcessor {
 		MqttConnectReturnCode returnCode = message.variableHeader().connectReturnCode();
 		switch (message.variableHeader().connectReturnCode()) {
 			case CONNECTION_ACCEPTED:
+				// 标记为链接成功，只有链接成功之后才能 sub 和 pub
+				context.set(MQTT_CONNECTED_KEY, Boolean.TRUE);
 				logger.info("MQTT 连接成功！");
 				break;
 			case CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD:

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClient.java

@@ -187,4 +187,14 @@ public final class MqttClient {
 		return context;
 	}
 
+	/**
+	 * 判断 mqtt 是否链接成功，仅仅在链接成功之后才能 sub 和 pub
+	 *
+	 * @return 是否成功
+	 */
+	public boolean isConnected() {
+		Boolean connected = (Boolean) context.get(DefaultMqttClientProcessor.MQTT_CONNECTED_KEY);
+		return Boolean.TRUE.equals(connected);
+	}
+
 }

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClientAioListener.java

@@ -50,18 +50,19 @@ public class MqttClientAioListener extends DefaultClientAioListener {
 				.cleanSession(clientConfig.isCleanSession())
 				.protocolVersion(clientConfig.getProtocolVersion())
 				.willFlag(willMessage != null);
-			// 密码
+			// 2. 密码
 			String password = clientConfig.getPassword();
 			if (StrUtil.isNotBlank(password)) {
 				builder.password(password.getBytes(StandardCharsets.UTF_8));
 			}
-			// 遗嘱消息
+			// 3. 遗嘱消息
 			if (willMessage != null) {
 				builder.willTopic(willMessage.getTopic())
 					.willMessage(willMessage.getMessage())
 					.willRetain(willMessage.isRetain())
 					.willQoS(willMessage.getQos());
 			}
+			// 4. 发送链接请求
 			Tio.send(context, builder.build());
 		}
 	}

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClientCreator.java

@@ -217,6 +217,9 @@ public final class MqttClientCreator {
 		return willMessage(builder.build());
 	}
 
+
+
+
 	public MqttClient connect() throws Exception {
 		// 1. 生成 默认的 clientId
 		String clientId = getClientId();

mica-mqtt-example/src/main/java/net/dreamlu/iot/mqtt/client/MqttClientTest.java

 package net.dreamlu.iot.mqtt.client;
 
-import net.dreamlu.iot.mqtt.codec.MqttVersion;
 import net.dreamlu.iot.mqtt.core.client.MqttClient;
+import org.tio.client.ClientChannelContext;
 
 import java.nio.ByteBuffer;
 import java.util.Timer;
@@ -15,21 +15,44 @@ import java.util.TimerTask;
 public class MqttClientTest {
 
 	public static void main(String[] args) throws Exception {
+		String productKey = "g27jB42P9hm";
+		String deviceName = "3dbc1cb4";
+		String deviceSecret = "87b337020a99ddb9eab5bb68f8b2f891";
+
+		// 计算MQTT连接参数。
+		MqttSign sign = new MqttSign();
+		sign.calculate(productKey, deviceName, deviceSecret);
+
+		String username = sign.getUsername();
+		String password = sign.getPassword();
+		String clientId = sign.getClientid();
+		System.out.println("username: " + username);
+		System.out.println("password: " + password);
+		System.out.println("clientid: " + clientId);
+
 		// 初始化 mqtt 客户端
 		MqttClient client = MqttClient.create()
-			.ip("127.0.0.1")
-			.username("admin")
-			.password("123456")
-			.protocolVersion(MqttVersion.MQTT_5)
+			.ip(productKey + ".iot-as-mqtt.cn-shanghai.aliyuncs.com")
+			.port(443)
+			.username(username)
+			.password(password)
+			.clientId(clientId)
 			.connect();
 
-		Timer timer = new Timer();
-		timer.schedule(new TimerTask() {
-			@Override
-			public void run() {
-				client.publish("testtopicxx", ByteBuffer.wrap("mica最牛皮".getBytes()));
-			}
-		}, 1000, 2000);
+		Thread.sleep(1000L);
+//		Timer timer = new Timer();
+//		timer.schedule(new TimerTask() {
+//			@Override
+//			public void run() {
+				String content = "{\"id\":\"1\",\"version\":\"1.0\",\"params\":{\"LightSwitch\":1}}";
+				client.publish("/sys/g27jB42P9hm/" + deviceName + "/thing/event/property/post", ByteBuffer.wrap(content.getBytes()));
+//			}
+//		}, 1000, 2000);
+
+		Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+			System.out.println("disconnect..................");
+			client.disconnect();
+		}));
 
 	}
 }

mica-mqtt-example/src/main/java/net/dreamlu/iot/mqtt/client/MqttSign.java

+package net.dreamlu.iot.mqtt.client;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.math.BigInteger;
+
+class CryptoUtil {
+    private static String hmac(String plainText, String key, String algorithm, String format) throws Exception {
+        if (plainText == null || key == null) {
+            return null;
+        }
+
+        byte[] hmacResult = null;
+
+        Mac mac = Mac.getInstance(algorithm);
+        SecretKeySpec secretKeySpec = new SecretKeySpec(key.getBytes(), algorithm);
+        mac.init(secretKeySpec);
+        hmacResult = mac.doFinal(plainText.getBytes());
+        return String.format(format, new BigInteger(1, hmacResult));
+    }
+
+//    public static String hmacMd5(String plainText, String key) throws Exception {
+//        return hmac(plainText,key,"HmacMD5","%032x");
+//    }
+//
+//    public static String hmacSha1(String plainText, String key) throws Exception {
+//        return hmac(plainText,key,"HmacSHA1","%040x");
+//    }
+
+    public static String hmacSha256(String plainText, String key) throws Exception {
+        return hmac(plainText,key,"HmacSHA256","%064x");
+    }
+}
+
+public class MqttSign {
+    private String username = "";
+
+    private String password = "";
+
+    private String clientid = "";
+
+    public String getUsername() { return this.username;}
+
+    public String getPassword() { return this.password;}
+
+    public String getClientid() { return this.clientid;}
+
+    public void calculate(String productKey, String deviceName, String deviceSecret) {
+        if (productKey == null || deviceName == null || deviceSecret == null) {
+            return;
+        }
+
+        try {
+            //MQTT用户名
+            this.username = deviceName + "&" + productKey;
+
+            //MQTT密码
+            String timestamp = Long.toString(System.currentTimeMillis());
+            String plainPasswd = "clientId" + productKey + "." + deviceName + "deviceName" +
+                    deviceName + "productKey" + productKey + "timestamp" + timestamp;
+            this.password = CryptoUtil.hmacSha256(plainPasswd, deviceSecret);
+
+            //MQTT ClientId
+            this.clientid = productKey + "." + deviceName + "|" + "timestamp=" + timestamp +
+                    ",_v=paho-java-1.0.0,securemode=2,signmethod=hmacsha256|";
+        }catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}