✨ mica-mqtt-spring-boot-starter MqttClientSubscribe 注解支持自定义 MqttClientTemplate Bean

mica-mqtt-broker/src/main/java/net/dreamlu/iot/mqtt/broker/auth/MqttPublishPermission.java

+/*
+ * Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package net.dreamlu.iot.mqtt.broker.auth;
+
+import lombok.extern.slf4j.Slf4j;
+import net.dreamlu.iot.mqtt.codec.MqttQoS;
+import net.dreamlu.iot.mqtt.core.server.auth.IMqttServerPublishPermission;
+import org.springframework.context.annotation.Configuration;
+import org.tio.core.ChannelContext;
+
+/**
+ * mqtt 服务端校验客户端是否有发布权限
+ *
+ * @author L.cm
+ */
+@Slf4j
+@Configuration(proxyBeanMethods = false)
+public class MqttPublishPermission implements IMqttServerPublishPermission {
+
+	/**
+	 * 否有发布权限
+	 *
+	 * @param context  ChannelContext
+	 * @param clientId 客户端 id
+	 * @param topic    topic
+	 * @param qoS      MqttQoS
+	 * @param isRetain 是否保留消息
+	 * @return 否有发布权限
+	 */
+	public boolean hasPermission(ChannelContext context, String clientId, String topic, MqttQoS qoS, boolean isRetain) {
+		log.info("Mqtt client publish permission check clientId:{} topic:{}.", clientId, topic);
+		// 可自定义业务，判断客户端是否有发布的权限。
+		return true;
+	}
+
+}

mica-mqtt-spring-boot-starter/src/main/java/net/dreamlu/iot/mqtt/spring/client/MqttClientConfiguration.java

@@ -23,6 +23,7 @@ import net.dreamlu.iot.mqtt.core.client.MqttClientCreator;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.util.StringUtils;
@@ -91,14 +92,14 @@ public class MqttClientConfiguration {
 		return clientCreator;
 	}
 
-	@Bean
+	@Bean(MqttClientTemplate.DEFAULT_CLIENT_TEMPLATE_BEAN)
 	public MqttClientTemplate mqttClientTemplate(MqttClientCreator mqttClientCreator) {
 		return new MqttClientTemplate(mqttClientCreator);
 	}
 
 	@Bean
-	public MqttClientSubscribeDetector mqttClientSubscribeDetector(MqttClientTemplate clientTemplate) {
-		return new MqttClientSubscribeDetector(clientTemplate);
+	public MqttClientSubscribeDetector mqttClientSubscribeDetector(ApplicationContext applicationContext) {
+		return new MqttClientSubscribeDetector(applicationContext);
 	}
 
 }

mica-mqtt-spring-boot-starter/src/main/java/net/dreamlu/iot/mqtt/spring/client/MqttClientSubscribe.java

@@ -44,4 +44,11 @@ public @interface MqttClientSubscribe {
 	 */
 	MqttQoS qos() default MqttQoS.AT_MOST_ONCE;
 
+	/**
+	 * 客户端 bean 名称
+	 *
+	 * @return bean name
+	 */
+	String clientTemplateBean() default MqttClientTemplate.DEFAULT_CLIENT_TEMPLATE_BEAN;
+
 }

mica-mqtt-spring-boot-starter/src/main/java/net/dreamlu/iot/mqtt/spring/client/MqttClientSubscribeDetector.java

@@ -19,9 +19,9 @@ package net.dreamlu.iot.mqtt.spring.client;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import net.dreamlu.iot.mqtt.core.client.IMqttClientMessageListener;
-import net.dreamlu.iot.mqtt.core.client.MqttClient;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.context.ApplicationContext;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.lang.NonNull;
 import org.springframework.util.ClassUtils;
@@ -29,7 +29,6 @@ import org.springframework.util.ReflectionUtils;
 
 import java.lang.reflect.Modifier;
 import java.nio.ByteBuffer;
-import java.util.Objects;
 
 /**
  * MqttClient 订阅监听器
@@ -39,21 +38,20 @@ import java.util.Objects;
 @Slf4j
 @RequiredArgsConstructor
 public class MqttClientSubscribeDetector implements BeanPostProcessor {
-	private final MqttClientTemplate clientTemplate;
+	private final ApplicationContext applicationContext;
 
 	@Override
 	public Object postProcessAfterInitialization(@NonNull Object bean, String beanName) throws BeansException {
-		MqttClient mqttClient = clientTemplate.getMqttClient();
-		Objects.requireNonNull(mqttClient, "MqttClient is null.");
 		Class<?> userClass = ClassUtils.getUserClass(bean);
-		// 1. 查找类
+		// 1. 查找类上的 MqttClientSubscribe 注解
 		if (bean instanceof IMqttClientMessageListener) {
 			MqttClientSubscribe subscribe = AnnotationUtils.findAnnotation(userClass, MqttClientSubscribe.class);
 			if (subscribe != null) {
+				MqttClientTemplate clientTemplate = getMqttClientTemplate(applicationContext, subscribe.clientTemplateBean());
 				clientTemplate.subscribe(subscribe.value(), subscribe.qos(), (IMqttClientMessageListener) bean);
 			}
 		} else {
-			// 2. 查找方法
+			// 2. 查找方法上的 MqttClientSubscribe 注解
 			ReflectionUtils.doWithMethods(userClass, method -> {
 				MqttClientSubscribe subscribe = AnnotationUtils.findAnnotation(method, MqttClientSubscribe.class);
 				if (subscribe != null) {
@@ -78,7 +76,8 @@ public class MqttClientSubscribeDetector implements BeanPostProcessor {
 						throw new IllegalArgumentException("@MqttClientSubscribe on method " + method + " parameter type must String and ByteBuffer.");
 					}
 					// 4. 订阅
-					mqttClient.subscribe(subscribe.value(), subscribe.qos(), (topic, payload) ->
+					MqttClientTemplate clientTemplate = getMqttClientTemplate(applicationContext, subscribe.clientTemplateBean());
+					clientTemplate.subscribe(subscribe.value(), subscribe.qos(), (topic, payload) ->
 						ReflectionUtils.invokeMethod(method, bean, topic, payload)
 					);
 				}
@@ -87,4 +86,8 @@ public class MqttClientSubscribeDetector implements BeanPostProcessor {
 		return bean;
 	}
 
+	private static MqttClientTemplate getMqttClientTemplate(ApplicationContext applicationContext, String beanName) {
+		return applicationContext.getBean(beanName, MqttClientTemplate.class);
+	}
+
 }

mica-mqtt-spring-boot-starter/src/main/java/net/dreamlu/iot/mqtt/spring/client/MqttClientTemplate.java

@@ -39,6 +39,7 @@ import java.util.List;
 @Slf4j
 @RequiredArgsConstructor
 public class MqttClientTemplate implements InitializingBean, DisposableBean, Ordered {
+	public static final String DEFAULT_CLIENT_TEMPLATE_BEAN = "mqttClientTemplate";
 	private final MqttClientCreator mqttClientCreator;
 	private MqttClient client;