Merge branch '20230705' into 'master_dev'

20230705

See merge request yunteng/thingskit!203

 /**
 /**
  * Copyright © 2016-2022 The Thingsboard Authors
  * Copyright © 2016-2022 The Thingsboard Authors
  *
  *
- * 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
+ * <p>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
+ * <p>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
+ * <p>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.
  * limitations under the License.
  */
  */
 package org.thingsboard.server.transport.udp;
 package org.thingsboard.server.transport.udp;
 
 
-
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_CLOSED;
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_CLOSED;
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_OPEN;
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_OPEN;
 
 
@@ -25,9 +22,11 @@ import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.MoreExecutors;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.socket.DatagramPacket;
 import io.netty.handler.codec.mqtt.*;
 import io.netty.handler.codec.mqtt.*;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.concurrent.GenericFutureListener;
@@ -73,411 +72,472 @@ import org.thingsboard.server.transport.tcp.session.TcpGatewaySessionHandler;
  * @author Andrew Shvayka
  * @author Andrew Shvayka
  */
  */
 @Slf4j
 @Slf4j
-      public class UdpDatagramDataHandler   implements GenericFutureListener<Future<? super Void>>, SessionMsgListener {
-
-    private final UUID sessionId;
-    private final TcpTransportContext context;
-    private final TransportService transportService;
-
-
-    /**
-     * 需要处理的消息队列，例如：需要下发给设备的，设备上传的。
-     */
-    final TcpDeviceSessionCtx deviceSessionCtx;
-    volatile InetSocketAddress address;
-
-    volatile TcpGatewaySessionHandler gatewaySessionHandler;
-    private final AtomicInteger authedCounter = new AtomicInteger();
-
-
-    UdpDatagramDataHandler(ChannelHandlerContext ctx,TcpTransportContext context, InetSocketAddress devAddress) {
-        super();
-        this.sessionId = UUID.randomUUID();
-        this.context = context;
-        this.address = devAddress;
-        this.transportService = context.getTransportService();
-        this.deviceSessionCtx = new TcpDeviceSessionCtx(sessionId, context);
+public class UdpDatagramDataHandler
+    implements GenericFutureListener<Future<? super Void>>, SessionMsgListener {
+
+  private final UUID sessionId;
+  private final TcpTransportContext context;
+  private final TransportService transportService;
+
+  /** 需要处理的消息队列，例如：需要下发给设备的，设备上传的。 */
+  final TcpDeviceSessionCtx deviceSessionCtx;
+
+  volatile InetSocketAddress address;
+
+  volatile TcpGatewaySessionHandler gatewaySessionHandler;
+  private final AtomicInteger authedCounter = new AtomicInteger();
+
+  UdpDatagramDataHandler(
+      ChannelHandlerContext ctx, TcpTransportContext context, InetSocketAddress devAddress) {
+    super();
+    this.sessionId = UUID.randomUUID();
+    this.context = context;
+    this.address = devAddress;
+    this.transportService = context.getTransportService();
+    this.deviceSessionCtx = new TcpDeviceSessionCtx(sessionId, context);
+  }
+
+  public void dealDeviceMsg(ChannelHandlerContext ctx, String msg) {
+    deviceSessionCtx.setChannel(ctx);
+    if (deviceSessionCtx.getDeviceInfo() == null || deviceSessionCtx.getDeviceProfile() == null) {
+      processConnect(ctx, msg);
+      return;
     }
     }
-
-
-
-
-
-
-    public void dealDeviceMsg(ChannelHandlerContext ctx, String msg) {
-        deviceSessionCtx.setChannel(ctx);
-        if (deviceSessionCtx.getDeviceInfo() == null || deviceSessionCtx.getDeviceProfile() == null) {
-            processConnect(ctx, msg);
-            return;
-        }
-        final int queueSize = deviceSessionCtx.getMsgQueueSize();
-        if (queueSize >= context.getMessageQueueSizePerDeviceLimit()) {
-            log.info("Closing current session because msq queue size for device {} exceed limit {} with msgQueueSize counter {} and actual queue size {}",
-                    deviceSessionCtx.getDeviceId(), context.getMessageQueueSizePerDeviceLimit(), queueSize, deviceSessionCtx.getMsgQueueSize());
-//            ctx.close();
-            return;
-        }
-        deviceSessionCtx.addToQueue(msg);
-        processQueueMessage(ctx); //Under the normal conditions the msg queue will contain 0 messages. Many messages will be processed on device connect event in separate thread pool
-
+    final int queueSize = deviceSessionCtx.getMsgQueueSize();
+    if (queueSize >= context.getMessageQueueSizePerDeviceLimit()) {
+      log.info(
+          "Closing current session because msq queue size for device {} exceed limit {} with msgQueueSize counter {} and actual queue size {}",
+          deviceSessionCtx.getDeviceId(),
+          context.getMessageQueueSizePerDeviceLimit(),
+          queueSize,
+          deviceSessionCtx.getMsgQueueSize());
+      //            ctx.close();
+      return;
     }
     }
-
-
-    void processQueueMessage(ChannelHandlerContext ctx) {
-        if (!deviceSessionCtx.isConnected()) {
-            log.trace("[{}][{}] Postpone processing msg due to device is not connected. Msg queue size is {}", sessionId, deviceSessionCtx.getDeviceId(), deviceSessionCtx.getMsgQueueSize());
-            return;
-        }
-        deviceSessionCtx.tryProcessQueuedMsgs(msg -> processDeviceSessionMsg(ctx, msg));
+    deviceSessionCtx.addToQueue(msg);
+    processQueueMessage(
+        ctx); // Under the normal conditions the msg queue will contain 0 messages. Many messages
+              // will be processed on device connect event in separate thread pool
+  }
+
+  void processQueueMessage(ChannelHandlerContext ctx) {
+    if (!deviceSessionCtx.isConnected()) {
+      log.trace(
+          "[{}][{}] Postpone processing msg due to device is not connected. Msg queue size is {}",
+          sessionId,
+          deviceSessionCtx.getDeviceId(),
+          deviceSessionCtx.getMsgQueueSize());
+      return;
     }
     }
+    deviceSessionCtx.tryProcessQueuedMsgs(msg -> processDeviceSessionMsg(ctx, msg));
+  }
 
 
-
-    private void processDeviceSessionMsg(ChannelHandlerContext ctx, String tcpMessage) {
-        if (!checkConnected(ctx, tcpMessage)) {
-            return;
-        }
-        deviceSessionCtx.doUpScript(tcpMessage, r -> {
-            if (gatewaySessionHandler != null) {
-                processGatewayDeviceMsg(ctx, r);
-            }
-            processDirectDeviceMsg(ctx, r);
+  private void processDeviceSessionMsg(ChannelHandlerContext ctx, String tcpMessage) {
+    if (!checkConnected(ctx, tcpMessage)) {
+      return;
+    }
+    deviceSessionCtx.doUpScript(
+        tcpMessage,
+        r -> {
+          if (gatewaySessionHandler != null) {
+            processGatewayDeviceMsg(ctx, r);
+          }
+          processDirectDeviceMsg(ctx, r);
         });
         });
-
+  }
+
+  /**
+   * 上行脚本解析结果是否包含数据
+   *
+   * @param datas 数据集合
+   * @return
+   */
+  private boolean hasDatas(Map<String, Object> datas) {
+    if (datas == null || datas.isEmpty()) {
+      return false;
     }
     }
-
-
-    /**
-     * 上行脚本解析结果是否包含数据
-     * @param datas 数据集合
-     * @return
-     */
-    private boolean hasDatas(Map<String, Object> datas) {
-        if (datas == null || datas.isEmpty()) {
-            return false;
-        }
-        return true;
+    return true;
+  }
+
+  private void processGatewayDeviceMsg(ChannelHandlerContext ctx, TcpUpEntry tcpMessage) {
+    log.trace(
+        "[{}][{}] Processing publish msg [{}]!",
+        sessionId,
+        deviceSessionCtx.getDeviceId(),
+        tcpMessage);
+    try {
+      Map<String, Object> datas = tcpMessage.getDatas();
+      if (hasDatas(datas)) {
+        datas.forEach(
+            (devName, param) -> {
+              if (TkScriptFactory.ORIGINAL_DATA_FILED.equals(devName)) {
+                return;
+              }
+              if (tcpMessage.getTelemetry()) {
+                gatewaySessionHandler.onDeviceTelemetry(
+                    devName, tcpMessage.getRequestId(), param.toString());
+              } else {
+                //                        gatewaySessionHandler.onDeviceRpcResponse(devName,
+                // tcpMessage.getRequestId(), param.toString());
+              }
+            });
+      } else {
+        transportService.reportActivity(deviceSessionCtx.getSessionInfo());
+        pushDeviceMsg(ctx, tcpMessage.getAckMsg());
+      }
+    } catch (Exception e) {
+      log.debug("[{}] Failed to process publish msg [{}][{}]", sessionId, tcpMessage, e);
+      //            ctx.close();
     }
     }
-
-
-    private void processGatewayDeviceMsg(ChannelHandlerContext ctx, TcpUpEntry tcpMessage) {
-        log.trace("[{}][{}] Processing publish msg [{}]!", sessionId, deviceSessionCtx.getDeviceId(), tcpMessage);
-        try {
-            Map<String, Object> datas = tcpMessage.getDatas();
-            if (hasDatas(datas)) {
-                datas.forEach((devName, param) -> {
-                    if (TkScriptFactory.ORIGINAL_DATA_FILED.equals(devName)) {
-                        return;
-                    }
-                    if (tcpMessage.getTelemetry()) {
-                        gatewaySessionHandler.onDeviceTelemetry(devName, tcpMessage.getRequestId(), param.toString());
-                    } else {
-//                        gatewaySessionHandler.onDeviceRpcResponse(devName, tcpMessage.getRequestId(), param.toString());
-                    }
-
-                });
-            } else {
-                transportService.reportActivity(deviceSessionCtx.getSessionInfo());
-                pushDeviceMsg(ctx, tcpMessage.getAckMsg());
-            }
-        } catch (Exception e) {
-            log.debug("[{}] Failed to process publish msg [{}][{}]", sessionId, tcpMessage, e);
-//            ctx.close();
+  }
+
+  private void processDirectDeviceMsg(ChannelHandlerContext ctx, TcpUpEntry tcpMessage) {
+    log.trace(
+        "[{}][{}] Processing publish msg [{}]!",
+        sessionId,
+        deviceSessionCtx.getDeviceId(),
+        tcpMessage);
+    try {
+      TcpTransportAdaptor payloadAdaptor = deviceSessionCtx.getPayloadAdaptor();
+      Map<String, Object> datas = tcpMessage.getDatas();
+      if (hasDatas(datas)) {
+        String dataStr = JacksonUtil.toString(datas);
+        if (tcpMessage.getTelemetry()) {
+          TransportProtos.PostTelemetryMsg postTelemetryMsg =
+              payloadAdaptor.convertToPostTelemetry(deviceSessionCtx, dataStr);
+          transportService.process(
+              deviceSessionCtx.getSessionInfo(),
+              postTelemetryMsg,
+              getPubAckCallback(ctx, tcpMessage));
+        } else {
+          TransportProtos.ToDeviceRpcResponseMsg postRpcMsg =
+              payloadAdaptor.convertToDeviceRpcResponse(deviceSessionCtx, tcpMessage);
+          transportService.process(
+              deviceSessionCtx.getSessionInfo(), postRpcMsg, getPubAckCallback(ctx, tcpMessage));
         }
         }
+      } else {
+        transportService.reportActivity(deviceSessionCtx.getSessionInfo());
+        pushDeviceMsg(ctx, tcpMessage.getAckMsg());
+      }
+    } catch (AdaptorException e) {
+      log.debug("[{}] Failed to process publish msg [{}][{}]", sessionId, tcpMessage, e);
+      //            ctx.close();
     }
     }
-
-    private void processDirectDeviceMsg(ChannelHandlerContext ctx, TcpUpEntry tcpMessage) {
-        log.trace("[{}][{}] Processing publish msg [{}]!", sessionId, deviceSessionCtx.getDeviceId(), tcpMessage);
-        try {
-            TcpTransportAdaptor payloadAdaptor = deviceSessionCtx.getPayloadAdaptor();
-            Map<String, Object> datas = tcpMessage.getDatas();
-            if (hasDatas(datas)) {
-                String dataStr = JacksonUtil.toString(datas);
-                if (tcpMessage.getTelemetry()) {
-                    TransportProtos.PostTelemetryMsg postTelemetryMsg = payloadAdaptor.convertToPostTelemetry(deviceSessionCtx, dataStr);
-                    transportService.process(deviceSessionCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(ctx, tcpMessage));
-                } else {
-                    TransportProtos.ToDeviceRpcResponseMsg postRpcMsg = payloadAdaptor.convertToDeviceRpcResponse(deviceSessionCtx, tcpMessage);
-                    transportService.process(deviceSessionCtx.getSessionInfo(), postRpcMsg, getPubAckCallback(ctx, tcpMessage));
-                }
-            } else {
-                transportService.reportActivity(deviceSessionCtx.getSessionInfo());
-                pushDeviceMsg(ctx, tcpMessage.getAckMsg());
-            }
-        } catch (AdaptorException e) {
-            log.debug("[{}] Failed to process publish msg [{}][{}]", sessionId, tcpMessage, e);
-//            ctx.close();
+  }
+
+  private <T> TransportServiceCallback<Void> getPubAckCallback(
+      final ChannelHandlerContext ctx, final TcpUpEntry msg) {
+    return new TransportServiceCallback<>() {
+      @Override
+      public void onSuccess(Void dummy) {
+        log.trace("[{}] Published msg: {}", sessionId, msg);
+        if (StringUtils.isNotEmpty(msg.getAckMsg())) {
+          pushDeviceMsg(ctx, msg.getAckMsg());
         }
         }
+      }
+
+      @Override
+      public void onError(Throwable e) {
+        log.trace("[{}] Failed to publish msg: {}", sessionId, msg, e);
+        //                ctx.close();
+      }
+    };
+  }
+
+  void processConnect(ChannelHandlerContext ctx, String accessToken) {
+    log.debug("[{}][{}] Processing connect msg for client: {}!", address, sessionId, accessToken);
+
+    if (DataConstants.PROVISION.equals(accessToken)
+        || DataConstants.PROVISION.equals(accessToken)) {
+      deviceSessionCtx.setProvisionOnly(true);
+      pushDeviceMsg(ctx, "CONNECTION_ACCEPTED");
+    } else {
+      TkScriptInvokeService.authScripts.forEach(
+          id -> {
+            ListenableFuture item = context.getJsEngine().invokeFunction(id, accessToken);
+            Futures.addCallback(
+                item,
+                new FutureCallback<String>() {
+                  @Override
+                  public void onSuccess(@Nullable String result) {
+                    processAuthTokenConnect(
+                        ctx,
+                        id,
+                        JacksonUtil.fromString(result.replace("\\", "\\\\"), TcpAuthEntry.class));
+                  }
+
+                  @Override
+                  public void onFailure(Throwable t) {
+                    onValidateFailed(
+                        ctx, MqttConnectReturnCode.CONNECTION_REFUSED_PAYLOAD_FORMAT_INVALID);
+                  }
+                },
+                MoreExecutors.directExecutor());
+          });
     }
     }
-
-
-    private <T> TransportServiceCallback<Void> getPubAckCallback(final ChannelHandlerContext ctx, final TcpUpEntry msg) {
-        return new TransportServiceCallback<>() {
+  }
+
+  private void processAuthTokenConnect(
+      ChannelHandlerContext ctx, UUID scriptId, TcpAuthEntry accessToken) {
+
+    log.debug(
+        "[{}][{}] Processing connect msg for client with user name: {}!",
+        address,
+        sessionId,
+        accessToken);
+    if (null != accessToken.getClientId()) {}
+    if (null != accessToken.getUserName()) {}
+    String token = accessToken.getPassword();
+    if (StringUtils.isNotEmpty(token)) {
+      TransportProtos.ValidateDeviceTokenRequestMsg.Builder request =
+          TransportProtos.ValidateDeviceTokenRequestMsg.newBuilder();
+      request.setToken(token);
+      transportService.process(
+          DeviceTransportType.TCP,
+          request.build(),
+          new TransportServiceCallback<>() {
             @Override
             @Override
-            public void onSuccess(Void dummy) {
-                log.trace("[{}] Published msg: {}", sessionId, msg);
-                if (StringUtils.isNotEmpty(msg.getAckMsg())) {
-                    pushDeviceMsg(ctx, msg.getAckMsg());
-                }
+            public void onSuccess(ValidateDeviceCredentialsResponse msg) {
+              onValidateDeviceResponse(msg, ctx, accessToken, scriptId);
             }
             }
 
 
             @Override
             @Override
             public void onError(Throwable e) {
             public void onError(Throwable e) {
-                log.trace("[{}] Failed to publish msg: {}", sessionId, msg, e);
-//                ctx.close();
+              log.trace("[{}] Failed to process credentials: {}", address, accessToken, e);
+              onValidateFailed(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
             }
             }
-        };
+          });
+    } else {
+      onValidateFailed(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_PAYLOAD_FORMAT_INVALID);
     }
     }
-
-
-    void processConnect(ChannelHandlerContext ctx, String accessToken) {
-        log.debug("[{}][{}] Processing connect msg for client: {}!", address, sessionId, accessToken);
-
-        if (DataConstants.PROVISION.equals(accessToken) || DataConstants.PROVISION.equals(accessToken)) {
-            deviceSessionCtx.setProvisionOnly(true);
-            pushDeviceMsg(ctx,"CONNECTION_ACCEPTED");
-        } else {
-            TkScriptInvokeService.authScripts.forEach(id -> {
-                ListenableFuture item = context.getJsEngine().invokeFunction(id, accessToken);
-                Futures.addCallback(item, new FutureCallback<String>() {
-                    @Override
-                    public void onSuccess(@Nullable String result) {
-                        processAuthTokenConnect(ctx, id, JacksonUtil.fromString(result.replace("\\","\\\\"), TcpAuthEntry.class));
-                    }
-
-                    @Override
-                    public void onFailure(Throwable t) {
-                        onValidateFailed(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_PAYLOAD_FORMAT_INVALID);
-                    }
-                }, MoreExecutors.directExecutor());
-
-            });
-        }
+  }
+
+  private boolean checkConnected(ChannelHandlerContext ctx, String msg) {
+    if (deviceSessionCtx.isConnected()) {
+      return true;
+    } else {
+      log.info("[{}] Closing current session due to invalid msg order: {}", sessionId, msg);
+      return false;
     }
     }
-
-    private void processAuthTokenConnect(ChannelHandlerContext ctx, UUID scriptId, TcpAuthEntry accessToken) {
-
-        log.debug("[{}][{}] Processing connect msg for client with user name: {}!", address, sessionId, accessToken);
-        if (null != accessToken.getClientId()) {
-        }
-        if (null != accessToken.getUserName()) {
-        }
-        String token = accessToken.getPassword();
-        if(StringUtils.isNotEmpty(token)){
-            TransportProtos.ValidateDeviceTokenRequestMsg.Builder request = TransportProtos.ValidateDeviceTokenRequestMsg.newBuilder();
-            request.setToken(token);
-            transportService.process(DeviceTransportType.TCP, request.build(),
-                    new TransportServiceCallback<>() {
-                        @Override
-                        public void onSuccess(ValidateDeviceCredentialsResponse msg) {
-                            onValidateDeviceResponse(msg, ctx, accessToken, scriptId);
-                        }
-
-                        @Override
-                        public void onError(Throwable e) {
-                            log.trace("[{}] Failed to process credentials: {}", address, accessToken, e);
-                            onValidateFailed(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
-                        }
-                    });
-        }else{
-            onValidateFailed(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_PAYLOAD_FORMAT_INVALID);
+  }
+
+  private void checkGatewaySession(SessionMetaData sessionMetaData) {
+    TransportDeviceInfo device = deviceSessionCtx.getDeviceInfo();
+    try {
+      JsonNode infoNode = context.getMapper().readTree(device.getAdditionalInfo());
+      if (infoNode != null) {
+        JsonNode gatewayNode = infoNode.get("gateway");
+        if (gatewayNode != null && gatewayNode.asBoolean()) {
+          gatewaySessionHandler = new TcpGatewaySessionHandler(deviceSessionCtx, sessionId);
+          if (infoNode.has(DefaultTransportService.OVERWRITE_ACTIVITY_TIME)
+              && infoNode.get(DefaultTransportService.OVERWRITE_ACTIVITY_TIME).isBoolean()) {
+            sessionMetaData.setOverwriteActivityTime(
+                infoNode.get(DefaultTransportService.OVERWRITE_ACTIVITY_TIME).asBoolean());
+          }
         }
         }
+      }
+    } catch (IOException e) {
+      log.trace(
+          "[{}][{}] Failed to fetch device additional info", sessionId, device.getDeviceName(), e);
     }
     }
-
-
-
-
-
-
-
-
-
-    private boolean checkConnected(ChannelHandlerContext ctx, String msg) {
-        if (deviceSessionCtx.isConnected()) {
-            return true;
-        } else {
-            log.info("[{}] Closing current session due to invalid msg order: {}", sessionId, msg);
-            return false;
-        }
-    }
-
-    private void checkGatewaySession(SessionMetaData sessionMetaData) {
-        TransportDeviceInfo device = deviceSessionCtx.getDeviceInfo();
-        try {
-            JsonNode infoNode = context.getMapper().readTree(device.getAdditionalInfo());
-            if (infoNode != null) {
-                JsonNode gatewayNode = infoNode.get("gateway");
-                if (gatewayNode != null && gatewayNode.asBoolean()) {
-                    gatewaySessionHandler = new TcpGatewaySessionHandler(deviceSessionCtx, sessionId);
-                    if (infoNode.has(DefaultTransportService.OVERWRITE_ACTIVITY_TIME) && infoNode.get(DefaultTransportService.OVERWRITE_ACTIVITY_TIME).isBoolean()) {
-                        sessionMetaData.setOverwriteActivityTime(infoNode.get(DefaultTransportService.OVERWRITE_ACTIVITY_TIME).asBoolean());
-                    }
-                }
-            }
-        } catch (IOException e) {
-            log.trace("[{}][{}] Failed to fetch device additional info", sessionId, device.getDeviceName(), e);
-        }
+  }
+
+  @Override
+  public void operationComplete(Future<? super Void> future) throws Exception {
+    log.trace("[{}] Channel closed!", sessionId);
+    doDisconnect();
+  }
+
+  public void doDisconnect() {
+    if (deviceSessionCtx.isConnected()) {
+      log.debug("[{}] Client disconnected!", sessionId);
+      transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null);
+      transportService.deregisterSession(deviceSessionCtx.getSessionInfo());
+      if (gatewaySessionHandler != null) {
+        gatewaySessionHandler.onGatewayDisconnect();
+      }
+      deviceSessionCtx.setDisconnected();
     }
     }
-
-    @Override
-    public void operationComplete(Future<? super Void> future) throws Exception {
-        log.trace("[{}] Channel closed!", sessionId);
-        doDisconnect();
-    }
-
-    public void doDisconnect() {
-        if (deviceSessionCtx.isConnected()) {
-            log.debug("[{}] Client disconnected!", sessionId);
-            transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null);
-            transportService.deregisterSession(deviceSessionCtx.getSessionInfo());
-            if (gatewaySessionHandler != null) {
-                gatewaySessionHandler.onGatewayDisconnect();
+    deviceSessionCtx.release();
+  }
+
+  private void onValidateDeviceResponse(
+      ValidateDeviceCredentialsResponse msg,
+      ChannelHandlerContext ctx,
+      TcpAuthEntry authEntry,
+      UUID scriptId) {
+    if (!msg.hasDeviceInfo()) {
+      context.onAuthFailure(address);
+      onValidateFailed(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED);
+    } else {
+      DeviceProfile profile = msg.getDeviceProfile();
+      TkTcpDeviceProfileTransportConfiguration tcpConfig =
+          (TkTcpDeviceProfileTransportConfiguration)
+              profile.getProfileData().getTransportConfiguration();
+      if (scriptId != null && !tcpConfig.getAuthScriptId().equals(scriptId.toString())) {
+        authedCounter.incrementAndGet();
+        return;
+      }
+      context.onAuthSuccess(address);
+      deviceSessionCtx.setDeviceInfo(msg.getDeviceInfo());
+      deviceSessionCtx.setDeviceProfile(profile);
+      deviceSessionCtx.setSessionInfo(SessionInfoCreator.create(msg, context, sessionId));
+      transportService.process(
+          deviceSessionCtx.getSessionInfo(),
+          SESSION_EVENT_MSG_OPEN,
+          new TransportServiceCallback<Void>() {
+            @Override
+            public void onSuccess(Void msg) {
+              SessionMetaData sessionMetaData =
+                  transportService.registerAsyncSession(
+                      deviceSessionCtx.getSessionInfo(), UdpDatagramDataHandler.this);
+              checkGatewaySession(sessionMetaData);
+              pushDeviceMsg(ctx, authEntry.getSuccess());
+              deviceSessionCtx.setConnected(true);
+              log.debug("[{}] Client connected!", sessionId);
+
+              transportService.process(
+                  deviceSessionCtx.getSessionInfo(),
+                  TransportProtos.SubscribeToRPCMsg.newBuilder().build(),
+                  null);
+              transportService
+                  .getCallbackExecutor()
+                  .execute(
+                      () ->
+                          processQueueMessage(
+                              ctx)); // this callback will execute in Producer worker thread and
+                                     // hard or blocking work have to be submitted to the separate
+                                     // thread.
             }
             }
-            deviceSessionCtx.setDisconnected();
-        }
-        deviceSessionCtx.release();
-    }
 
 
-
-    private void onValidateDeviceResponse(ValidateDeviceCredentialsResponse msg, ChannelHandlerContext ctx, TcpAuthEntry authEntry, UUID scriptId) {
-        if (!msg.hasDeviceInfo()) {
-            context.onAuthFailure(address);
-            onValidateFailed(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED);
-        } else {
-            DeviceProfile profile = msg.getDeviceProfile();
-            TkTcpDeviceProfileTransportConfiguration tcpConfig = (TkTcpDeviceProfileTransportConfiguration) profile.getProfileData().getTransportConfiguration();
-            if (scriptId != null && !tcpConfig.getAuthScriptId().equals(scriptId.toString())) {
-                authedCounter.incrementAndGet();
-                return;
+            @Override
+            public void onError(Throwable e) {
+              if (e instanceof TbRateLimitsException) {
+                log.trace("[{}] Failed to submit session event: {}", sessionId, e.getMessage());
+              } else {
+                log.warn("[{}] Failed to submit session event", sessionId, e);
+              }
+              onValidateFailed(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
             }
             }
-            context.onAuthSuccess(address);
-            deviceSessionCtx.setDeviceInfo(msg.getDeviceInfo());
-            deviceSessionCtx.setDeviceProfile(profile);
-            deviceSessionCtx.setSessionInfo(SessionInfoCreator.create(msg, context, sessionId));
-            transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_OPEN, new TransportServiceCallback<Void>() {
-                @Override
-                public void onSuccess(Void msg) {
-                    SessionMetaData sessionMetaData = transportService.registerAsyncSession(deviceSessionCtx.getSessionInfo(), UdpDatagramDataHandler.this);
-                    checkGatewaySession(sessionMetaData);
-                    pushDeviceMsg(ctx,authEntry.getSuccess());
-                    deviceSessionCtx.setConnected(true);
-                    log.debug("[{}] Client connected!", sessionId);
-
-                    transportService.process(deviceSessionCtx.getSessionInfo(), TransportProtos.SubscribeToRPCMsg.newBuilder().build(), null);
-                    transportService.getCallbackExecutor().execute(() -> processQueueMessage(ctx)); //this callback will execute in Producer worker thread and hard or blocking work have to be submitted to the separate thread.
-                }
-
-                @Override
-                public void onError(Throwable e) {
-                    if (e instanceof TbRateLimitsException) {
-                        log.trace("[{}] Failed to submit session event: {}", sessionId, e.getMessage());
-                    } else {
-                        log.warn("[{}] Failed to submit session event", sessionId, e);
-                    }
-                    onValidateFailed(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
-                }
-            });
-        }
+          });
     }
     }
+  }
 
 
-    private void onValidateFailed(ChannelHandlerContext ctx, MqttConnectReturnCode msg) {
-        authedCounter.incrementAndGet();
-        if (TkScriptInvokeService.authScripts.size() == authedCounter.intValue()) {
-            pushDeviceMsg(ctx,msg.name());
+  private void onValidateFailed(ChannelHandlerContext ctx, MqttConnectReturnCode msg) {
+    authedCounter.incrementAndGet();
+    if (TkScriptInvokeService.authScripts.size() == authedCounter.intValue()) {
+      pushDeviceMsg(ctx, msg.name());
 
 
-//            ctx.close();
-        }
+      //            ctx.close();
     }
     }
-
-    @Override
-    public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg response) {
-
-    }
-
-    @Override
-    public void onAttributeUpdate(UUID sessionId, TransportProtos.AttributeUpdateNotificationMsg notification) {
-
-    }
-
-    @Override
-    public void onRemoteSessionCloseCommand(UUID sessionId, TransportProtos.SessionCloseNotificationProto sessionCloseNotification) {
-        log.trace("[{}] Received the remote command to close the session: {}", sessionId, sessionCloseNotification.getMessage());
-//        deviceSessionCtx.getChannel().close();
-    }
-
-    @Override
-    public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) {
-        log.debug("【{}】下发RPC命令【{}】给设备【{}】", sessionId, rpcRequest.getParams(), deviceSessionCtx.getDeviceInfo().getDeviceName());
-        TcpTransportAdaptor adaptor = deviceSessionCtx.getPayloadAdaptor();
-        try {
+  }
+
+  @Override
+  public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg response) {}
+
+  @Override
+  public void onAttributeUpdate(
+      UUID sessionId, TransportProtos.AttributeUpdateNotificationMsg notification) {}
+
+  @Override
+  public void onRemoteSessionCloseCommand(
+      UUID sessionId, TransportProtos.SessionCloseNotificationProto sessionCloseNotification) {
+    log.trace(
+        "[{}] Received the remote command to close the session: {}",
+        sessionId,
+        sessionCloseNotification.getMessage());
+    //        deviceSessionCtx.getChannel().close();
+  }
+
+  @Override
+  public void onToDeviceRpcRequest(
+      UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) {
+    log.debug(
+        "【{}】下发RPC命令【{}】给设备【{}】",
+        sessionId,
+        rpcRequest.getParams(),
+        deviceSessionCtx.getDeviceInfo().getDeviceName());
+    TcpTransportAdaptor adaptor = deviceSessionCtx.getPayloadAdaptor();
+    try {
       adaptor
       adaptor
           .convertToPublish(deviceSessionCtx, rpcRequest)
           .convertToPublish(deviceSessionCtx, rpcRequest)
           .ifPresent(
           .ifPresent(
               payload -> {
               payload -> {
                 deviceSessionCtx.rpcRequesting(payload.getIdentifier(), rpcRequest);
                 deviceSessionCtx.rpcRequesting(payload.getIdentifier(), rpcRequest);
-                Optional.ofNullable(pushDeviceMsg(deviceSessionCtx.getChannel(), payload.getDatas())).ifPresent(cf->{
-                    cf.addListener(
-                            result -> {
+                Optional.ofNullable(
+                        pushDeviceMsg(deviceSessionCtx.getChannel(), payload.getDatas()))
+                    .ifPresent(
+                        cf -> {
+                          cf.addListener(
+                              result -> {
                                 if (result.cause() == null) {
                                 if (result.cause() == null) {
-                                    transportService.process(
-                                            deviceSessionCtx.getSessionInfo(),
-                                            rpcRequest,
-                                            RpcStatus.DELIVERED,
-                                            TransportServiceCallback.EMPTY);
+                                  transportService.process(
+                                      deviceSessionCtx.getSessionInfo(),
+                                      rpcRequest,
+                                      RpcStatus.DELIVERED,
+                                      TransportServiceCallback.EMPTY);
                                 } else {
                                 } else {
-                                    // TODO: send error
+                                  // TODO: send error
                                 }
                                 }
-                            });
-                });
+                              });
+                        });
                 ;
                 ;
               });
               });
-        } catch (Exception e) {
-            transportService.process(deviceSessionCtx.getSessionInfo(),
-                    TransportProtos.ToDeviceRpcResponseMsg.newBuilder()
-                            .setRequestId(rpcRequest.getRequestId()).setError("Failed to convert device RPC command to TCP msg").build(), TransportServiceCallback.EMPTY);
-            log.error("[{}] Failed to convert device RPC command to TCP msg", sessionId, e);
-        }
+    } catch (Exception e) {
+      transportService.process(
+          deviceSessionCtx.getSessionInfo(),
+          TransportProtos.ToDeviceRpcResponseMsg.newBuilder()
+              .setRequestId(rpcRequest.getRequestId())
+              .setError("Failed to convert device RPC command to TCP msg")
+              .build(),
+          TransportServiceCallback.EMPTY);
+      log.error("[{}] Failed to convert device RPC command to TCP msg", sessionId, e);
     }
     }
-
-
-    @Override
-    public void onToServerRpcResponse(TransportProtos.ToServerRpcResponseMsg rpcResponse) {
-        log.debug("[{}] 服务端响应设备的RPC请求", sessionId);
-    }
-
-    /**
-     * 往设备推送消息
-     *
-     * @param message
-     * @return
-     */
-    private ChannelFuture pushDeviceMsg(ChannelHandlerContext ctx, String message) {
-        if(StringUtils.isBlank(message)){
-            return null;
-        }
-        ByteBuf buff = Unpooled.buffer();
-        if(!message.matches("-?[0-9a-fA-F]+")){
-            //不满足16进制将字符串转为16进制
-            message = ByteUtils.stringEncodeToHex(message);
-        }
-        buff.writeBytes(ByteUtils.hexToByteArray(message));
-        return ctx.writeAndFlush(buff);
+  }
+
+  @Override
+  public void onToServerRpcResponse(TransportProtos.ToServerRpcResponseMsg rpcResponse) {
+    log.debug("[{}] 服务端响应设备的RPC请求", sessionId);
+  }
+
+  /**
+   * 往设备推送消息
+   *
+   * @param message
+   * @return
+   */
+  private ChannelFuture pushDeviceMsg(ChannelHandlerContext ctx, String message) {
+    if (StringUtils.isBlank(message)) {
+      return null;
     }
     }
-
-
-    @Override
-    public void onDeviceProfileUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceProfile deviceProfile) {
-        deviceSessionCtx.onDeviceProfileUpdate(sessionInfo, deviceProfile);
-    }
-
-    @Override
-    public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) {
-        deviceSessionCtx.onDeviceUpdate(sessionInfo, device, deviceProfileOpt);
-    }
-
-    @Override
-    public void onDeviceDeleted(DeviceId deviceId) {
-        context.onAuthFailure(address);
-        ChannelHandlerContext ctx = deviceSessionCtx.getChannel();
-//        ctx.close();
+    ByteBuf buff = Unpooled.buffer();
+    if (!message.matches("-?[0-9a-fA-F]+")) {
+      // 不满足16进制将字符串转为16进制
+      message = ByteUtils.stringEncodeToHex(message);
     }
     }
+    buff.writeBytes(ByteBufUtil.decodeHexDump(message));
+    DatagramPacket packet = new DatagramPacket(buff, address);
+    return ctx.writeAndFlush(packet);
+  }
+
+  @Override
+  public void onDeviceProfileUpdate(
+      TransportProtos.SessionInfoProto sessionInfo, DeviceProfile deviceProfile) {
+    deviceSessionCtx.onDeviceProfileUpdate(sessionInfo, deviceProfile);
+  }
+
+  @Override
+  public void onDeviceUpdate(
+      TransportProtos.SessionInfoProto sessionInfo,
+      Device device,
+      Optional<DeviceProfile> deviceProfileOpt) {
+    deviceSessionCtx.onDeviceUpdate(sessionInfo, device, deviceProfileOpt);
+  }
+
+  @Override
+  public void onDeviceDeleted(DeviceId deviceId) {
+    context.onAuthFailure(address);
+    ChannelHandlerContext ctx = deviceSessionCtx.getChannel();
+    //        ctx.close();
+  }
 }
 }

 package org.thingsboard.server.dao.yunteng.impl;
 package org.thingsboard.server.dao.yunteng.impl;
 
 
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.common.data.yunteng.dto.DoActionDTO;
 import org.thingsboard.server.common.data.yunteng.dto.DoActionDTO;
-import org.thingsboard.server.common.data.yunteng.enums.DeviceTypeEnum;
 import org.thingsboard.server.common.data.yunteng.enums.ScopeEnum;
 import org.thingsboard.server.common.data.yunteng.enums.ScopeEnum;
 import org.thingsboard.server.dao.yunteng.entities.TenantBaseEntity;
 import org.thingsboard.server.dao.yunteng.entities.TenantBaseEntity;
 import org.thingsboard.server.dao.yunteng.entities.TkDeviceEntity;
 import org.thingsboard.server.dao.yunteng.entities.TkDeviceEntity;
@@ -14,44 +17,46 @@ import org.thingsboard.server.dao.yunteng.mapper.DoActionMapper;
 import org.thingsboard.server.dao.yunteng.service.AbstractBaseService;
 import org.thingsboard.server.dao.yunteng.service.AbstractBaseService;
 import org.thingsboard.server.dao.yunteng.service.DoActionService;
 import org.thingsboard.server.dao.yunteng.service.DoActionService;
 
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-/** @Description @Author cxy @Date 2021/12/6 20:23 */
+/**
+ * @Description @Author cxy @Date 2021/12/6 20:23
+ */
 @Service
 @Service
 @RequiredArgsConstructor
 @RequiredArgsConstructor
 public class TkDoActionServiceImpl extends AbstractBaseService<DoActionMapper, TkDoActionEntity>
 public class TkDoActionServiceImpl extends AbstractBaseService<DoActionMapper, TkDoActionEntity>
     implements DoActionService {
     implements DoActionService {
   private final DeviceMapper deviceMapper;
   private final DeviceMapper deviceMapper;
+
   @Override
   @Override
   public List<TkDoActionEntity> getActionsByAll(String sceneId) {
   public List<TkDoActionEntity> getActionsByAll(String sceneId) {
     LambdaQueryWrapper filter =
     LambdaQueryWrapper filter =
-        new LambdaQueryWrapper<TkDoActionEntity>().eq(TkDoActionEntity::getSceneLinkageId, sceneId).eq(TkDoActionEntity::getEntityType, ScopeEnum.ALL);
+        new LambdaQueryWrapper<TkDoActionEntity>()
+            .eq(TkDoActionEntity::getSceneLinkageId, sceneId)
+            .eq(TkDoActionEntity::getEntityType, ScopeEnum.ALL);
     return baseMapper.selectList(filter);
     return baseMapper.selectList(filter);
   }
   }
 
 
-
   @Override
   @Override
   public List<TkDoActionEntity> getActionsByPart(String sceneId) {
   public List<TkDoActionEntity> getActionsByPart(String sceneId) {
     LambdaQueryWrapper filter =
     LambdaQueryWrapper filter =
-            new LambdaQueryWrapper<TkDoActionEntity>().eq(TkDoActionEntity::getSceneLinkageId, sceneId).eq(TkDoActionEntity::getEntityType, ScopeEnum.PART);
+        new LambdaQueryWrapper<TkDoActionEntity>()
+            .eq(TkDoActionEntity::getSceneLinkageId, sceneId)
+            .eq(TkDoActionEntity::getEntityType, ScopeEnum.PART);
     List<TkDoActionEntity> source = baseMapper.selectList(filter);
     List<TkDoActionEntity> source = baseMapper.selectList(filter);
-    return source.stream().map(t ->{
-      List<TkDeviceEntity> partDevices = deviceMapper.selectList(new LambdaQueryWrapper<TkDeviceEntity>().in(TkDeviceEntity::getTbDeviceId,t.getDeviceId()));
-      List<String> deviceId = new ArrayList<>();
-      for(TkDeviceEntity item : partDevices){
-        if(DeviceTypeEnum.SENSOR.equals(item.getDeviceType())){
-          deviceId.add(item.getGatewayId());
-        }else{
-          deviceId.add(item.getTbDeviceId());
-        }
-      }
-      t.setDeviceId(deviceId);
-      return t;
-    }).collect(Collectors.toList());
-
+    return source.stream()
+        .map(
+            t -> {
+              List<TkDeviceEntity> partDevices =
+                  deviceMapper.selectList(
+                      new LambdaQueryWrapper<TkDeviceEntity>()
+                          .in(TkDeviceEntity::getTbDeviceId, t.getDeviceId()));
+              List<String> deviceId = new ArrayList<>();
+              for (TkDeviceEntity item : partDevices) {
+                deviceId.add(item.getTbDeviceId());
+              }
+              t.setDeviceId(deviceId);
+              return t;
+            })
+        .collect(Collectors.toList());
   }
   }
 
 
   @Override
   @Override