add coap网络组件

jetlinks-components/network-component/coap-component/src/main/java/org/jetlinks/community/network/coap/gateway/CoapServerDeviceGateway.java

@@ -139,23 +139,22 @@ public class CoapServerDeviceGateway implements DeviceGateway, MonitorSupportDev
             .filter(ignore->started.get())
             .publishOn(Schedulers.parallel())
             .doOnNext(operator->gatewayMonitor.receivedMessage())
+            .doOnNext(operator->{
+                counter.increment();
+                CoapConnectionSession newSession = new CoapConnectionSession(exchangeMessage.getDeviceId(),operator,DefaultTransport.CoAP,gatewayMonitor);
+                DeviceSession session = sessionManager.getSession(operator.getDeviceId());
+                if (null == session) {
+                    sessionManager.register(newSession);
+                } else if (session instanceof ReplaceableDeviceSession) {
+                    ((ReplaceableDeviceSession) session).replaceWith(newSession);
+                }else{
+                    sessionManager.register(newSession);
+                }
+                gatewayMonitor.connected();
+                gatewayMonitor.totalConnection(counter.sum());
+            })
             .flatMap(operator->
                 this.decodeAndHandleMessage(operator,exchangeMessage)
-                    .doOnNext(ignore->{
-                            counter.increment();
-                            CoapConnectionSession newSession = new CoapConnectionSession(exchangeMessage.getDeviceId(),operator,DefaultTransport.CoAP,gatewayMonitor);
-                            DeviceSession session = sessionManager.getSession(operator.getDeviceId());
-                            if (null == session) {
-                                sessionManager.register(newSession);
-                            } else if (session instanceof ReplaceableDeviceSession) {
-                                ((ReplaceableDeviceSession) session).replaceWith(newSession);
-                            }else{
-                                sessionManager.register(newSession);
-                            }
-                            gatewayMonitor.connected();
-                            gatewayMonitor.totalConnection(counter.sum());
-                        }
-                    )
             )
             .doOnSuccess(s->{
                 exchangeMessage.response(CoAP.ResponseCode.CONTENT);
@@ -170,7 +169,7 @@ public class CoapServerDeviceGateway implements DeviceGateway, MonitorSupportDev
         return operator
             .getProtocol()
             .flatMap(protocol -> protocol.getMessageCodec(getTransport()))
-            .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of(null, exchangeMessage, registry)))
+            .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of(sessionManager.getSession(operator.getDeviceId()), exchangeMessage, registry)))
             .cast(DeviceMessage.class)
             .flatMap(msg -> {
                 if (messageProcessor.hasDownstreams()) {
@@ -191,7 +190,7 @@ public class CoapServerDeviceGateway implements DeviceGateway, MonitorSupportDev
                 return handleMessage(operator, msg);
             })
             .then()
-            .doOnEach(ReactiveLogger.onError(err -> log.error("处理MQTT连接[{}]消息失败:{}", operator.getDeviceId(), exchangeMessage, err)))
+            .doOnEach(ReactiveLogger.onError(err -> log.error("处理Coap连接[{}]消息失败:{}", operator.getDeviceId(), exchangeMessage, err)))
             .onErrorResume((err) -> Mono.empty());
     }
 

jetlinks-components/network-component/coap-component/src/main/java/org/jetlinks/community/network/coap/resources/AbstractCoapResource.java

@@ -29,7 +29,7 @@ import java.util.function.Function;
 public abstract class AbstractCoapResource extends CoapResource {
     private final EmitterProcessor<CoapExchangeMessage> processor;
 
-    public static final String prefixTopicName="{productId}/{deviceId}";
+    public static final String prefixTopicName="{productId}/{deviceId}/**";
 
     public static final AntPathMatcher matcher = new AntPathMatcher(File.separator);
     public AbstractCoapResource(EmitterProcessor<CoapExchangeMessage> processor) {
@@ -54,8 +54,15 @@ public abstract class AbstractCoapResource extends CoapResource {
 
     public void handlePOST(DefaultCoapExchange exchange) {
         Map<String, String> map = matcher.extractUriTemplateVariables(prefixTopicName, this.getPath());
-        exchange.setDeviceId(String.valueOf(map.get("deviceId")));
-        exchange.setProductId(String.valueOf(map.get("productId")));
+        String path = this.getPath();
+        path=path.startsWith("/")?path.substring(1):path;
+        String[] split = path.split("/");
+        if(split.length<2){
+            log.warn("Coap连接,deviceId:[{}],productId:[{}],不能为空",exchange.getDeviceId(),exchange.getProductId());
+            exchange.reject();
+        }
+        exchange.setDeviceId(split[1]);
+        exchange.setProductId(split[0]);
         Map<String, String> elseMap = new HashMap<>();
         map.forEach((k,v)->{
             if(!"deviceId".equals(k)&&!"productId".equals(k)){

jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/support/JetLinksExtendCoapDeviceMessageCodec.java

@@ -1,5 +1,6 @@
 package org.jetlinks.community.support;
 
+import cn.hutool.core.util.StrUtil;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import io.netty.buffer.ByteBuf;
@@ -41,7 +42,12 @@ public class JetLinksExtendCoapDeviceMessageCodec extends JetlinksExtendTopicMes
                     byte[] req = new byte[byteBuf.readableBytes()];
                     byteBuf.readBytes(req);
                     byteBuf.resetReaderIndex();
-                    String payload = new String(ciphers.decrypt(req, secureKey));
+                    String payload =null;
+                    if(StrUtil.isEmpty(secureKey)){
+                        payload = new String(req);
+                    }else {
+                        payload = new String(ciphers.decrypt(req, secureKey));
+                    }
                     //解码
                     return Mono.just(decode(path, decode(payload)).getMessage());
                 });

jetlinks-components/network-component/network-core/src/main/java/org/jetlinks/community/support/JetLinksExtendProtocolSupportProvider.java

@@ -38,7 +38,7 @@ public class JetLinksExtendProtocolSupportProvider implements ProtocolSupportPro
             "encrypt(payload,secureKey);")
         .add("encAlg", "加密算法", "加密算法",
             new EnumType().addElement(EnumType.Element.of("AES", "AES加密(ECB,PKCS#5)", "加密模式:ECB,填充方式:PKCS#5")))
-        .add("secureKey", "密钥", "16位密钥KEY", new PasswordType());
+        .add("secureKey", "密钥", "信息传输加密16位密钥KEY", new PasswordType());
 
     private static final DefaultConfigMetadata coapDTLSConfig = new DefaultConfigMetadata(
         "CoAP DTLS配置",

jetlinks-manager/device-manager/src/main/java/org/jetlinks/community/device/web/DeviceInstanceController.java

@@ -11,6 +11,7 @@ import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.elasticsearch.search.aggregations.bucket.filter.Filter;
 import org.hswebframework.ezorm.core.param.Term;
+import org.hswebframework.ezorm.rdb.exception.DuplicateKeyException;
 import org.hswebframework.ezorm.rdb.mapping.ReactiveRepository;
 import org.hswebframework.ezorm.rdb.mapping.defaults.SaveResult;
 import org.hswebframework.reactor.excel.ReactorExcel;
@@ -56,7 +57,6 @@ import org.jetlinks.core.metadata.*;
 import org.jetlinks.core.metadata.types.GeoPoint;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
-import org.springframework.dao.DuplicateKeyException;
 import org.springframework.data.util.Lazy;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
@@ -224,8 +224,8 @@ public class DeviceInstanceController implements
         return Mono
             .zip(payload, Authentication.currentReactive(), this::applyAuthentication)
             .flatMap(entity -> service.insert(Mono.just(entity)).thenReturn(entity))
-            .onErrorMap(DuplicateKeyException.class, err -> new BusinessException("设备ID已存在", err))
-            .onErrorMap(e->new BusinessException("服务器繁忙，请稍后重试",e));
+            .onErrorResume(DuplicateKeyException.class,err->Mono.error(new BusinessException("设备ID已存在")));
+//            .onErrorResume(Exception.class,e->Mono.error(new BusinessException("服务器繁忙，请稍后重试",e)));
     }
 
     /**
@@ -872,9 +872,9 @@ public class DeviceInstanceController implements
     public Mono<?> queryPageGeo(@RequestBody GeoRequestParam param, @PathVariable("hasGeo") boolean hasGeo) {
         Mono<PagerResult<DeviceInstanceEntity>> resultMono = null;
         if(!hasGeo){
-           /**
-            * 没有查询区域
-            */
+            /**
+             * 没有查询区域
+             */
             param.getQuery().setIncludes(CollectionUtil.newHashSet("id","name","location","productId","productName"));
             Term term = new Term();
             term.setColumn("location");