Merge branch '20230420' into 'master_dev'

fix: 设备鉴权前校验凭证有效性

See merge request yunteng/thingskit!179

@@ -71,7 +71,9 @@ import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Pattern;
 
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;
 import static io.netty.handler.codec.mqtt.MqttMessageType.PUBACK;
@@ -219,27 +221,22 @@ public class TcpTransportHandler extends ChannelInboundHandlerAdapter implements
         deviceSessionCtx.doUpScript(tcpMessage, r -> {
             if (gatewaySessionHandler != null) {
                 processGatewayDeviceMsg(ctx, r);
-
-                if (!hasDatas(r.getDatas(), true)) {
-                    transportService.reportActivity(deviceSessionCtx.getSessionInfo());
-                    return;
-                }
             }
-
-
             processDirectDeviceMsg(ctx, r);
         });
 
     }
 
 
-    private boolean hasDatas(Map<String, Object> datas, boolean includeSource) {
+    /**
+     * 上行脚本解析结果是否包含数据
+     * @param datas 数据集合
+     * @return
+     */
+    private boolean hasDatas(Map<String, Object> datas) {
         if (datas == null || datas.isEmpty()) {
             return false;
         }
-        if (includeSource && !datas.containsKey(TkScriptFactory.ORIGINAL_DATA_FILED)) {
-            return false;
-        }
         return true;
     }
 
@@ -248,7 +245,7 @@ public class TcpTransportHandler extends ChannelInboundHandlerAdapter implements
         log.trace("[{}][{}] Processing publish msg [{}]!", sessionId, deviceSessionCtx.getDeviceId(), tcpMessage);
         try {
             Map<String, Object> datas = tcpMessage.getDatas();
-            if (hasDatas(datas, false)) {
+            if (hasDatas(datas)) {
                 datas.forEach((devName, param) -> {
                     if (TkScriptFactory.ORIGINAL_DATA_FILED.equals(devName)) {
                         return;
@@ -275,7 +272,7 @@ public class TcpTransportHandler extends ChannelInboundHandlerAdapter implements
         try {
             TcpTransportAdaptor payloadAdaptor = deviceSessionCtx.getPayloadAdaptor();
             Map<String, Object> datas = tcpMessage.getDatas();
-            if (hasDatas(datas, false)) {
+            if (hasDatas(datas)) {
                 String dataStr = JacksonUtil.toString(datas);
                 if (tcpMessage.getTelemetry()) {
                     TransportProtos.PostTelemetryMsg postTelemetryMsg = payloadAdaptor.convertToPostTelemetry(deviceSessionCtx, dataStr);
@@ -347,11 +344,12 @@ public class TcpTransportHandler extends ChannelInboundHandlerAdapter implements
         }
         if (null != accessToken.getUserName()) {
         }
-        if (null == accessToken.getPassword()) {
+        String token = accessToken.getPassword();
+    if (null == token || Pattern.compile(".*\\s+.*").matcher(token).matches()) {
             onValidateFailed(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_PAYLOAD_FORMAT_INVALID);
             return;
         }
-        request.setToken(accessToken.getPassword());
+        request.setToken(token);
 
         transportService.process(DeviceTransportType.TCP, request.build(),
                 new TransportServiceCallback<>() {
@@ -504,26 +502,12 @@ public class TcpTransportHandler extends ChannelInboundHandlerAdapter implements
 
     @Override
     public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg response) {
-//        log.trace("[{}] Received get attributes response", sessionId);
-//        String topicBase = attrReqTopicType.getAttributesResponseTopicBase();
-//        TcpTransportAdaptor adaptor = deviceSessionCtx.getAdaptor(attrReqTopicType);
-//        try {
-//            adaptor.convertToPublish(deviceSessionCtx, response, topicBase).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
-//        } catch (Exception e) {
-//            log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e);
-//        }
+
     }
 
     @Override
     public void onAttributeUpdate(UUID sessionId, TransportProtos.AttributeUpdateNotificationMsg notification) {
-//        log.trace("[{}] Received attributes update notification to device", sessionId);
-//        String topic = attrSubTopicType.getAttributesSubTopic();
-//        TcpTransportAdaptor adaptor = deviceSessionCtx.getAdaptor(attrSubTopicType);
-//        try {
-//            adaptor.convertToPublish(deviceSessionCtx, notification, topic).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
-//        } catch (Exception e) {
-//            log.trace("[{}] Failed to convert device attributes update to MQTT msg", sessionId, e);
-//        }
+
     }
 
     @Override
@@ -538,23 +522,23 @@ public class TcpTransportHandler extends ChannelInboundHandlerAdapter implements
         TcpTransportAdaptor adaptor = deviceSessionCtx.getPayloadAdaptor();
         try {
             adaptor.convertToPublish(deviceSessionCtx, rpcRequest).ifPresent(payload -> {
-//                int msgId = ((MqttPublishMessage) payload).variableHeader().packetId();
-//                if (isAckExpected(payload)) {
-//                    rpcAwaitingAck.put(msgId, rpcRequest);
-//                    context.getScheduler().schedule(() -> {
-//                        TransportProtos.ToDeviceRpcRequestMsg msg = rpcAwaitingAck.remove(msgId);
-//                        if (msg != null) {
-//                            transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
-//                        }
-//                    }, Math.max(0, Math.min(deviceSessionCtx.getContext().getTimeout(), rpcRequest.getExpirationTime() - System.currentTimeMillis())), TimeUnit.MILLISECONDS);
-//                }
+                int msgId = rpcRequest.getRequestId();
+                boolean oneWay = rpcRequest.getOneway();
+                if (!oneWay) {
+                    rpcAwaitingAck.put(msgId, rpcRequest);
+                    context.getScheduler().schedule(() -> {
+                        TransportProtos.ToDeviceRpcRequestMsg msg = rpcAwaitingAck.remove(msgId);
+                        if (msg != null) {
+                            transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
+                        }
+                    }, Math.max(0, Math.min(deviceSessionCtx.getContext().getTimeout(), rpcRequest.getExpirationTime() - System.currentTimeMillis())), TimeUnit.MILLISECONDS);
+                }
                 var cf = pushDeviceMsg(deviceSessionCtx.getChannel(), payload.getDatas());
                 cf.addListener(result -> {
                     if (result.cause() == null) {
-//                        if (!isAckExpected(payload)) {
-//                            transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
-//                        } else
-                        if (rpcRequest.getPersisted()) {
+                        if (oneWay) {
+                            transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
+                        } else if (rpcRequest.getPersisted()) {
                             transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.SENT, TransportServiceCallback.EMPTY);
                         }
                     } else {
@@ -573,13 +557,6 @@ public class TcpTransportHandler extends ChannelInboundHandlerAdapter implements
     @Override
     public void onToServerRpcResponse(TransportProtos.ToServerRpcResponseMsg rpcResponse) {
         log.debug("[{}] 服务端响应设备的RPC请求", sessionId);
-//        String baseTopic = toServerRpcSubTopicType.getRpcResponseTopicBase();
-//        TcpTransportAdaptor adaptor = deviceSessionCtx.getAdaptor(toServerRpcSubTopicType);
-//        try {
-//            adaptor.convertToPublish(deviceSessionCtx, rpcResponse, baseTopic).ifPresent(deviceSessionCtx.getChannel()::writeAndFlush);
-//        } catch (Exception e) {
-//            log.trace("[{}] Failed to convert device RPC command to MQTT msg", sessionId, e);
-//        }
     }
 
     /**

@@ -93,11 +93,15 @@ public class TcpGatewayDeviceSessionCtx extends TcpDeviceWareSessionContext impl
         try {
             parent.getPayloadAdaptor().convertToPublish(this, request).ifPresent(
                     payload -> {
+                        boolean oneWay = request.getOneway();
                         ChannelFuture channelFuture = parent.pushDeviceMsg(payload.getDatas());
                         if (request.getPersisted()) {
                             channelFuture.addListener(result -> {
-                                if (result.cause() == null) {
+                                if (oneWay) {
+                                    transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
+                                } else if (request.getPersisted()) {
                                     transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY);
+
                                 }
                             });
                         }