fixed 网桥设备接口

jetlinks-manager/bridge-manager/src/main/java/org/jetlinks/community/bridge/core/DefaultBridgeConfigManager.java

@@ -63,7 +63,7 @@ public class DefaultBridgeConfigManager implements BridgeConfigManager {
 
     private DefaultBridgeConfigManager(String bridgeId,String bridgeProductKey, String bridgeDeviceName, String bridgeDeviceSecret, String http2Endpoint,String authEndpoint,Map<String, Object> configMaps, PopClientConfiguration popConfiguration) {
         this.bridgeId=bridgeId;
-        this.macId="";
+        this.macId=createMacId(bridgeId);
         this.bridgeProductKey = bridgeProductKey;
         this.bridgeDeviceName = bridgeDeviceName;
         this.bridgeDeviceSecret = bridgeDeviceSecret;
@@ -73,11 +73,6 @@ public class DefaultBridgeConfigManager implements BridgeConfigManager {
         this.popConfiguration = popConfiguration;
     }
 
-    public static void main(String[] args) throws UnknownHostException {
-        System.out.println(createMacId("123"));
-        System.out.println(createMacId("123456789420"));
-    }
-
     private static String createMacId(String id){
         String result="%s%s-%s%s-%s%s-%s%s-%s%s-%s%s";
         String[] params=new String[12];
@@ -149,8 +144,6 @@ public class DefaultBridgeConfigManager implements BridgeConfigManager {
     }
     @Override
     public String getMacAddress() {
-//        return String.valueOf(HashUtil.rsHash(originalProductId));
-//        return getMac(getInetAddress());
         return macId;
     }
     @Override

jetlinks-manager/bridge-manager/src/main/java/org/jetlinks/community/bridge/entity/AliIotBridgeEntity.java

@@ -63,7 +63,6 @@ public class AliIotBridgeEntity extends GenericEntity<String> {
     @Comment("阿里云实例id")
     @Column(name = "iotInstanceId")
     @Schema(description = "阿里云实例id")
-    @Deprecated
     private String iotInstanceId;
 
     @Column(name = "state",length = 16,nullable = false)

jetlinks-manager/bridge-manager/src/main/java/org/jetlinks/community/bridge/server/aliyun/AliBridgeGateway.java

@@ -1,5 +1,6 @@
 package org.jetlinks.community.bridge.server.aliyun;
 
+import cn.hutool.core.util.StrUtil;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
@@ -31,6 +32,8 @@ import reactor.core.scheduler.Schedulers;
 
 import javax.annotation.PostConstruct;
 import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Objects;
 
 /**
@@ -38,12 +41,13 @@ import java.util.Objects;
  * @version 1.0.0
  * @ClassName AliBridgeGateway.java
  * @Description 阿里云网桥网关
+ *              处理网桥服务，解码/编码
  * @createTime 2021年11月30日 08:35:00
  */
 @Slf4j
 @Service
 public class AliBridgeGateway{
-    private volatile AliBridgeServer aliBridgeServer;
+    private Map<String,AliBridgeServer> bridgeMap=new HashMap<>();
     private final EmitterProcessor<Message> messageProcessor = EmitterProcessor.create(false);
 
     private final FluxSink<Message> sink = messageProcessor.sink(FluxSink.OverflowStrategy.BUFFER);
@@ -81,17 +85,17 @@ public class AliBridgeGateway{
                     MessageType messageType = message.getMessageType();
                     switch (messageType) {
                         case del:
-                            return delBridgeServer((String) message.getMsg());
+                            return delBridgeServer(clusterId,message.getBridgeId());
                         case init:
                             return initBridge((AliIotBridgeEntity) message.getMsg());
                         case update:
-                            return replaceBridgeServer(clusterId, (AliIotBridgeEntity) message.getMsg());
+                            return replaceBridgeServer(clusterId, message.getBridgeId(),(AliIotBridgeEntity) message.getMsg());
                         case register:
-                            return registerDevice(clusterId, (AliIotBridgeDeviceConfig) message.getMsg());
+                            return registerDevice(clusterId, message.getBridgeId(),(AliIotBridgeDeviceConfig) message.getMsg());
                         case unregister:
-                            return unregisterDevice(clusterId, (String) message.getMsg());
+                            return unregisterDevice(clusterId,message.getBridgeId(), (String) message.getMsg());
                         case restart:
-                            return reconnect(clusterId);
+                            return reconnect(clusterId,message.getBridgeId());
                         default:
                             return Mono.empty();
                     }
@@ -101,20 +105,32 @@ public class AliBridgeGateway{
     }
 
 
+    private AliBridgeServer getServer(String serverId){
+        return this.bridgeMap.get(serverId);
+    }
 
-    public Mono<Void> reconnect(String serverId){
-        return handleClusterOperation(serverId,MessageType.restart,serverId)
-            .switchIfEmpty(aliBridgeServer.reconnect());
+    private AliBridgeServer putServer(String serverId,AliBridgeServer server){
+        return this.bridgeMap.put(serverId,server);
+    }
+    /**
+     * 重启网桥
+     * @param bridgeId 网桥id
+     * @return
+     */
+    public Mono<Void> reconnect(String nodeId,String bridgeId){
+        return handleClusterOperation(nodeId,bridgeId,MessageType.restart,bridgeId)
+            .switchIfEmpty(getServer(bridgeId).reconnect());
     }
     /**
      * 注册设备
-     * @param serverId 节点id
+     * @param nodeId 节点id
+     * @param bridgeId 网桥id
      * @param config 设备配置
      * @return
      */
-    public Mono<Void> registerDevice(String serverId, AliIotBridgeDeviceConfig config){
-        return handleClusterOperation(serverId,MessageType.register,config)
-            .switchIfEmpty(this.aliBridgeServer
+    public Mono<Void> registerDevice(String nodeId,String bridgeId,AliIotBridgeDeviceConfig config){
+        return handleClusterOperation(nodeId,bridgeId,MessageType.register,config)
+            .switchIfEmpty(getServer(bridgeId)
                 .register(config.getOriginalIdentity(),config.getProductKey(),config.getDeviceName(),config.getDeviceSecret()).then());
     }
 
@@ -125,22 +141,22 @@ public class AliBridgeGateway{
      * @param originalIdentity 设备id
      * @return
      */
-    public Mono<Void> unregisterDevice(String bridgeId,String originalIdentity){
-        return handleClusterOperation(bridgeId,MessageType.unregister,originalIdentity)
-            .switchIfEmpty(aliBridgeServer.unRegister(originalIdentity));
+    public Mono<Void> unregisterDevice(String nodeId,String bridgeId,String originalIdentity){
+        return handleClusterOperation(nodeId,bridgeId,MessageType.unregister,originalIdentity)
+            .switchIfEmpty(getServer(bridgeId).unRegister(originalIdentity));
     }
 
     public Mono<Void> initBridge(AliIotBridgeEntity bridgeEntity){
-        return handleClusterOperation(bridgeEntity.getNodeId(),MessageType.init,bridgeEntity)
-            .switchIfEmpty(Mono.justOrEmpty(aliBridgeServer)
+        return handleClusterOperation(bridgeEntity.getNodeId(),
+            bridgeEntity.getId(),MessageType.init,bridgeEntity)
+            .switchIfEmpty(Mono.justOrEmpty(getServer(bridgeEntity.getId()))
                 .then()
                 .switchIfEmpty(AliBridgeServer.create(eventBus,registry,bridgeEntity,clusterId)
                     .doOnNext(server -> {
-                        if(aliBridgeServer==null){
-                            aliBridgeServer=server;
-                        }else {
-                            server.stopBridge().subscribe();
-                            throw new BusinessException("网桥正在运行，请勿重复启动");
+                        AliBridgeServer oldServer =
+                            this.bridgeMap.putIfAbsent(server.getId(), server);
+                        if(oldServer!=null){
+                            oldServer.stopBridge().subscribe();
                         }
                     })
                     .doOnNext(server->{
@@ -153,26 +169,31 @@ public class AliBridgeGateway{
                     .then()
                     .onErrorResume(error->
                         Mono.error(()->{
-                            this.delBridgeServer(bridgeEntity.getNodeId());
+                            this.delBridgeServer(bridgeEntity.getNodeId(),
+                                bridgeEntity.getId());
                             return new BusinessException("网桥配置出错，请确定配置参数是否正确;或请勿重复启动网桥）");
                         })))
                 .then());
     }
 
-    public Mono<Void> delBridgeServer(String serverId){
-        return handleClusterOperation(serverId,MessageType.del,serverId)
-            .switchIfEmpty(Mono.justOrEmpty(aliBridgeServer)
-                .filter(Objects::nonNull)
-                .flatMap(AliBridgeServer::stopBridge));
-    }
 
-    public Mono<Void> replaceBridgeServer(String serverId,AliIotBridgeEntity bridge){
-        return handleClusterOperation(serverId,MessageType.update,bridge)
-            .switchIfEmpty(delBridgeServer(serverId)
+
+    public Mono<Void> replaceBridgeServer(String oldNodeId,String oldBridgeId,AliIotBridgeEntity bridge){
+        return handleClusterOperation(oldNodeId,oldBridgeId,MessageType.update,bridge)
+            .switchIfEmpty(delBridgeServer(oldNodeId,oldBridgeId)
                 .concatWith(initBridge(bridge))
                 .then());
     }
 
+
+    public Mono<Void> delBridgeServer(String nodeId,String bridgeId){
+        return handleClusterOperation(nodeId,bridgeId,MessageType.del,bridgeId)
+            .switchIfEmpty(Mono.justOrEmpty(getServer(bridgeId))
+                .filter(Objects::nonNull)
+                .flatMap(AliBridgeServer::stopBridge));
+    }
+
+
     //解码消息并处理
     private Mono<Void> decodeAndHandleMessage(AliBridgeMessage msg) {
         return Mono.justOrEmpty(AliBridgeCodec.decode(msg,msg.getProductKey(),msg.getDeviceName()))
@@ -217,19 +238,32 @@ public class AliBridgeGateway{
     }
 
 
-    private Mono<Void> handleClusterOperation(String serverId, MessageType messageType, Object msg){
-        if(!this.clusterId.equals(serverId)){
-            return clusterManager.getTopic(String.format(format,serverId))
-                .publish(Mono.just(BridgeMessage.of(clusterId,messageType,msg))).then();
+    /**
+     *
+     * @param nodeId 网桥所在节点id
+     * @param bridgeId 网桥id
+     * @param messageType  消息类型
+     * @param msg 消息
+     * @return
+     */
+    private Mono<Void> handleClusterOperation(String nodeId,String bridgeId, MessageType messageType, Object msg){
+        if(StrUtil.isNullOrUndefined(bridgeId)){
+            log.error("AliBridgeGateway.handleClusterOperation 网桥id不能为空,");
+            return Mono.error(new BusinessException("系统内部错误"));
+        }
+        if(!this.clusterId.equals(nodeId)){
+            return clusterManager.getTopic(String.format(format,nodeId))
+                .publish(Mono.just(BridgeMessage.of(clusterId,messageType,bridgeId,msg))).then();
         }
         return Mono.empty();
     }
 
     @AllArgsConstructor(staticName = "of")
     @Data
-    static class BridgeMessage implements Serializable {
+    private static class BridgeMessage implements Serializable {
         private String from;
         private MessageType messageType;
+        private String bridgeId;
         private Object msg;
     }
 

jetlinks-manager/bridge-manager/src/main/java/org/jetlinks/community/bridge/server/aliyun/AliBridgeServer.java

@@ -4,6 +4,7 @@ import cn.hutool.core.lang.Assert;
 import com.aliyun.iot.as.bridge.core.handler.DownlinkChannelHandler;
 import com.aliyun.iot.as.bridge.core.model.PopClientConfiguration;
 import com.aliyun.iot.as.bridge.core.model.Session;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.hswebframework.web.exception.BusinessException;
 import org.hswebframework.web.logger.ReactiveLogger;
@@ -32,6 +33,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
  * @ClassName AliBridgeServer.java
  * @Description 产品网桥
  *              管理统一产品下所有设备网桥
+ *              网桥以id表示其唯一性（而不是使用mac地址）
+ * @see  DefaultBridgeConfigManager
  * @createTime 2021年11月30日 11:01:00
  */
 @Slf4j
@@ -54,15 +57,18 @@ public class AliBridgeServer  implements BridgeServer {
 
     private final String clusterId;
 
+    @Getter
+    private final String id;
     private final DeviceRegistry deviceRegistry;
-    private AliBridgeServer(EventBus eventBus,DeviceRegistry deviceRegistry,String clusterId) {
+    private AliBridgeServer(EventBus eventBus,DeviceRegistry deviceRegistry,String clusterId,String id) {
         this.deviceRegistry=deviceRegistry;
         this.eventBus = eventBus;
         this.clusterId=clusterId;
+        this.id=id;
     }
 
     public static Mono<AliBridgeServer> create(EventBus eventBus,DeviceRegistry deviceRegistry, AliIotBridgeEntity bridge,String clusterId) {
-        AliBridgeServer aliBridgeServer = new AliBridgeServer(eventBus,deviceRegistry,clusterId);
+        AliBridgeServer aliBridgeServer = new AliBridgeServer(eventBus,deviceRegistry,clusterId,bridge.getId());
         return  Mono.just(aliBridgeServer)
             .flatMap(server->server.initBridge(bridge))
             .thenReturn(aliBridgeServer);

jetlinks-manager/bridge-manager/src/main/java/org/jetlinks/community/bridge/service/AliBridgeService.java

@@ -39,7 +39,7 @@ public class AliBridgeService extends GenericReactiveCacheSupportCrudService<Ali
                     .where(AliIotBridgeDeviceConfig::getBridgeId,bridgeParam.getId())
                     .fetch()
                     .flatMap(deviceConfig->
-                        bridgeGateway.registerDevice(clusterManager.getCurrentServerId(),deviceConfig)
+                        bridgeGateway.registerDevice(bridgeParam.getNodeId(),bridgeParam.getId(),deviceConfig)
                 )))
             .subscribe();
     }

jetlinks-manager/bridge-manager/src/main/java/org/jetlinks/community/bridge/web/AliBridgeServerController.java

@@ -55,9 +55,10 @@ public class AliBridgeServerController implements
     @DeleteMapping("/delete/{serverId}/{bridgeId}")
     @Operation(summary = "删除网桥")
     @DeleteAction
-    public Mono<Void> deleteBridge(@PathVariable("serverId")String serverId,@PathVariable("bridgeId")String bridgeId,String id){
+    public Mono<Void> deleteBridge(@PathVariable("serverId")String id){
         return bridgeService.findById(id)
-            .flatMap(bridge->Mono.zip(bridgeGateway.delBridgeServer(bridge.getNodeId())
+            .flatMap(bridge->Mono.zip(
+                bridgeGateway.delBridgeServer(bridge.getNodeId(),bridge.getId())
                 ,bridgeService.deleteById(id))
                 .concatWith(bridgeDeviceService.createDelete().where(AliIotBridgeDeviceConfig::getBridgeId,id).execute().then(Mono.empty()))
                 .then()
@@ -68,42 +69,46 @@ public class AliBridgeServerController implements
     @PostMapping("/register/{serverId}")
     @Operation(summary = "注册网桥设备")
     @CreateAction
-    public Mono<Void> register(@PathVariable("serverId") String serverId,@RequestBody AliIotBridgeDeviceConfig config){
-        return Mono.zip(
-            //保存网桥设备信息
-            bridgeDeviceService.save(config),
-            //注册网桥设备
-            bridgeGateway.registerDevice(serverId,config))
-            .then()
-            .onErrorResume(DuplicateKeyException.class,e->Mono.error(new BusinessException("设备不可重复添加")));
+    public Mono<Void> register(@PathVariable("serverId") String id,@RequestBody AliIotBridgeDeviceConfig config){
+        return bridgeService.findById(id)
+            .flatMap(bridge->
+                Mono.zip(
+                    //保存网桥设备信息
+                    bridgeDeviceService.save(config),
+                    //注册网桥设备
+                    bridgeGateway.registerDevice(bridge.getNodeId(),bridge.getId(),config))
+                    .then()
+                    .onErrorResume(DuplicateKeyException.class,e->Mono.error(new BusinessException("设备不可重复添加")))
+            );
     }
 
     @PostMapping("/pause/{serverId}")
     @Operation(summary = "暂停网桥")
     @CreateAction
-    public Mono<Void> pause(@PathVariable("serverId") String serverId){
-        return bridgeGateway.delBridgeServer(serverId)
-            .then();
+    public Mono<Void> pause(@PathVariable("serverId") String id){
+        return bridgeService.findById(id)
+            .flatMap(server->bridgeGateway.delBridgeServer(server.getNodeId(),server.getId()));
     }
 
     @PostMapping("/start/{serverId}")
     @Operation(summary = "重启网桥")
     @CreateAction
-    public Mono<Void> startBridge(@PathVariable("serverId")String serverId){
-        return bridgeGateway.reconnect(serverId);
+    public Mono<Void> startBridge(@PathVariable("serverId")String id){
+        return bridgeService.findById(id)
+            .flatMap(bridge-> bridgeGateway.reconnect(bridge.getNodeId(),id));
     }
 
     @PostMapping("/unregister/{serverId}")
     @Operation(summary = "取消注册网桥设备")
     @DeleteAction
-    public Mono<Void> unRegister(@PathVariable("serverId")String serverId,@RequestBody String originalIdentity,String id){
-        return bridgeDeviceService.findById(id)
-            .flatMap(bridgeDevice->
+    public Mono<Void> unRegister(@RequestBody String originalIdentity,@PathVariable("serverId") String id){
+        return bridgeService.findById(id)
+            .flatMap(bridge->
                 Mono.zip(
                     //删除网桥设备
                     bridgeDeviceService.createDelete().where(AliIotBridgeDeviceConfig::getOriginalIdentity,originalIdentity).execute(),
                     //取消注册
-                    bridgeGateway.unregisterDevice(bridgeDevice.getBridgeId(),bridgeDevice.getOriginalIdentity())).then()
+                    bridgeGateway.unregisterDevice(bridge.getNodeId(),bridge.getId(),originalIdentity)).then()
             );
     }
 
@@ -112,25 +117,25 @@ public class AliBridgeServerController implements
      * 1、删除网桥
      * 2、重启网桥
      * 3、注册网桥设备
-     * @param serverId
+     * @param id
      * @param bridge
      * @return
      */
     @PutMapping("/update/{serverId}")
     @Operation(summary = "更新网桥信息")
     @CreateAction
-    public Mono<Void> updateBridge(@PathVariable("serverId")String serverId,@RequestBody AliIotBridgeEntity bridge,String id){
+    public Mono<Void> updateBridge(@RequestBody AliIotBridgeEntity bridge,@PathVariable("serverId") String id){
         //删除设备后重新注册设备
         return bridgeService.findById(id)
             .flatMap(oldBridge->
                 Mono.zip(
-                    bridgeGateway.replaceBridgeServer(oldBridge.getNodeId(),bridge)
+                    bridgeGateway.replaceBridgeServer(oldBridge.getNodeId(),oldBridge.getId(),bridge)
                     , bridgeService.save(bridge))
                     .flatMap(tp2->bridgeDeviceService.createQuery()
                         .where(AliIotBridgeDeviceConfig::getBridgeId,bridge.getId())
                         .fetch()
                         .flatMap(deviceConfig->
-                            bridgeGateway.registerDevice(oldBridge.getNodeId(),deviceConfig)
+                            bridgeGateway.registerDevice(bridge.getNodeId(),bridge.getId(),deviceConfig)
                         ).then()
                     )
             );