MQTT网关可自定义认证协议

jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/supports/DeviceGatewayProperties.java

@@ -2,13 +2,14 @@ package org.jetlinks.community.gateway.supports;
 
 import lombok.Getter;
 import lombok.Setter;
+import org.jetlinks.community.ValueObject;
 
 import java.util.HashMap;
 import java.util.Map;
 
 @Getter
 @Setter
-public class DeviceGatewayProperties {
+public class DeviceGatewayProperties  implements ValueObject {
 
     private String id;
 
@@ -18,5 +19,8 @@ public class DeviceGatewayProperties {
 
     private Map<String,Object> configuration=new HashMap<>();
 
-
+    @Override
+    public Map<String, Object> getAll() {
+        return configuration;
+    }
 }

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttServerDeviceGateway.java

@@ -7,6 +7,7 @@ import org.hswebframework.web.logger.ReactiveLogger;
 import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor;
 import org.jetlinks.community.gateway.monitor.GatewayMonitors;
 import org.jetlinks.community.gateway.monitor.MonitorSupportDeviceGateway;
+import org.jetlinks.core.ProtocolSupport;
 import org.jetlinks.core.device.AuthenticationResponse;
 import org.jetlinks.core.device.DeviceOperator;
 import org.jetlinks.core.device.DeviceRegistry;
@@ -63,19 +64,24 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
 
     private final AtomicBoolean started = new AtomicBoolean();
 
+    private final Mono<ProtocolSupport> supportMono;
+
     private Disposable disposable;
 
     public MqttServerDeviceGateway(String id,
                                    DeviceRegistry registry,
                                    DeviceSessionManager sessionManager,
                                    MqttServer mqttServer,
-                                   DecodedClientMessageHandler messageHandler) {
+                                   DecodedClientMessageHandler messageHandler,
+                                   Mono<ProtocolSupport> customProtocol
+                                   ) {
         this.gatewayMonitor = GatewayMonitors.getDeviceGatewayMonitor(id);
         this.id = id;
         this.registry = registry;
         this.sessionManager = sessionManager;
         this.mqttServer = mqttServer;
         this.messageHandler = messageHandler;
+        this.supportMono = customProtocol;
     }
 
     @Override
@@ -88,7 +94,7 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
         if (started.getAndSet(true) || disposable != null) {
             return;
         }
-        disposable = (Disposable) mqttServer
+        disposable = mqttServer
             .handleConnection()
             .filter(conn -> {
                 if (!started.get()) {
@@ -110,30 +116,29 @@ class MqttServerDeviceGateway implements DeviceGateway, MonitorSupportDeviceGate
 
     //处理连接，并进行认证
     private Mono<Tuple3<DeviceOperator, AuthenticationResponse, MqttConnection>> handleConnection(MqttConnection connection) {
-        return Mono.justOrEmpty(connection.getAuth())
-            //没有认证信息,则拒绝连接.
-            .switchIfEmpty(Mono.fromRunnable(() -> {
-                connection.reject(MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED);
-                gatewayMonitor.rejected();
-            }))
-            .flatMap(auth ->
-                registry.getDevice(connection.getClientId())
-                    .flatMap(device -> device
-                        .authenticate(new MqttAuthenticationRequest(connection.getClientId(), auth.getUsername(), auth.getPassword(), getTransport()))
-                        .switchIfEmpty(Mono.fromRunnable(() -> connection.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD)))
-                        .flatMap(resp -> {
-                            String deviceId = StringUtils.isEmpty(resp.getDeviceId()) ? device.getDeviceId() : resp.getDeviceId();
-                            //认证返回了新的设备ID,则使用新的设备
-                            if (!deviceId.equals(device.getDeviceId())) {
-                                return registry
-                                    .getDevice(deviceId)
-                                    .map(operator -> Tuples.of(operator, resp, connection))
-                                    .switchIfEmpty(Mono.fromRunnable(() -> connection.reject(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED)))
-                                    ;
-                            }
-                            return Mono.just(Tuples.of(device, resp, connection));
-                        })
-                    ))
+        return  Mono
+            .justOrEmpty(connection.getAuth())
+            .flatMap(auth -> {
+                MqttAuthenticationRequest request = new MqttAuthenticationRequest(connection.getClientId(), auth.getUsername(), auth.getPassword(), getTransport());
+                return supportMono
+                    //使用自定义协议来认证
+                    .map(support -> support.authenticate(request, registry))
+                    .defaultIfEmpty(Mono.defer(() -> registry
+                        .getDevice(connection.getClientId())
+                        .flatMap(device -> device.authenticate(request))))
+                    .flatMap(Function.identity())
+                    .switchIfEmpty(Mono.fromRunnable(() -> connection.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD)));
+            })
+            .flatMap(resp -> {
+                String deviceId = StringUtils.isEmpty(resp.getDeviceId()) ? connection.getClientId() : resp.getDeviceId();
+                //认证返回了新的设备ID,则使用新的设备
+                return registry
+                    .getDevice(deviceId)
+                    .map(operator -> Tuples.of(operator, resp, connection))
+                    .switchIfEmpty(Mono.fromRunnable(() -> connection.reject(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED)))
+                    ;
+            })
+            //设备注册信息不存在,拒绝连接
             .onErrorResume((err) -> Mono.fromRunnable(() -> {
                 gatewayMonitor.rejected();
                 connection.reject(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);

jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttServerDeviceGatewayProvider.java

@@ -1,5 +1,6 @@
 package org.jetlinks.community.network.mqtt.gateway.device;
 
+import org.jetlinks.core.ProtocolSupports;
 import org.jetlinks.core.device.DeviceRegistry;
 import org.jetlinks.core.server.session.DeviceSessionManager;
 import org.jetlinks.community.gateway.DeviceGateway;
@@ -24,14 +25,18 @@ public class MqttServerDeviceGatewayProvider implements DeviceGatewayProvider {
 
     private final DecodedClientMessageHandler messageHandler;
 
+    private final ProtocolSupports protocolSupports;
+
     public MqttServerDeviceGatewayProvider(NetworkManager networkManager,
                                            DeviceRegistry registry,
                                            DeviceSessionManager sessionManager,
-                                           DecodedClientMessageHandler messageHandler) {
+                                           DecodedClientMessageHandler messageHandler,
+                                           ProtocolSupports protocolSupports) {
         this.networkManager = networkManager;
         this.registry = registry;
         this.sessionManager = sessionManager;
         this.messageHandler = messageHandler;
+        this.protocolSupports=protocolSupports;
     }
 
     @Override
@@ -53,10 +58,15 @@ public class MqttServerDeviceGatewayProvider implements DeviceGatewayProvider {
     public Mono<DeviceGateway> createDeviceGateway(DeviceGatewayProperties properties) {
         return networkManager
             .<MqttServer>getNetwork(getNetworkType(), properties.getNetworkId())
-            .map(mqttServer -> {
-                MqttServerDeviceGateway gateway = new MqttServerDeviceGateway(properties.getId(), registry, sessionManager, mqttServer, messageHandler);
-
-                return gateway;
-            });
+            .map(mqttServer -> new MqttServerDeviceGateway(
+                properties.getId(),
+                registry,
+                sessionManager,
+                mqttServer,
+                messageHandler,
+                properties.getString("protocol")
+                    .map(id -> Mono.defer(() -> protocolSupports.getProtocol(id)))
+                    .orElse(Mono.empty())
+            ));
     }
 }