Merge branch '20230627' into 'master_dev'

refactor: UDP协议功能完善

See merge request yunteng/thingskit!201

+/**
+ * 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
+ *
+ *     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
+ * limitations under the License.
+ */
+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_OPEN;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.mqtt.*;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.DeviceProfile;
+import org.thingsboard.server.common.data.DeviceTransportType;
+import org.thingsboard.server.common.data.device.profile.TkTcpDeviceProfileTransportConfiguration;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.rpc.RpcStatus;
+import org.thingsboard.server.common.data.yunteng.utils.ByteUtils;
+import org.thingsboard.server.common.msg.tools.TbRateLimitsException;
+import org.thingsboard.server.common.transport.SessionMsgListener;
+import org.thingsboard.server.common.transport.TransportService;
+import org.thingsboard.server.common.transport.TransportServiceCallback;
+import org.thingsboard.server.common.transport.adaptor.AdaptorException;
+import org.thingsboard.server.common.transport.auth.SessionInfoCreator;
+import org.thingsboard.server.common.transport.auth.TransportDeviceInfo;
+import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
+import org.thingsboard.server.common.transport.service.DefaultTransportService;
+import org.thingsboard.server.common.transport.service.SessionMetaData;
+import org.thingsboard.server.gen.transport.TransportProtos;
+import org.thingsboard.server.transport.tcp.TcpTransportContext;
+import org.thingsboard.server.transport.tcp.adaptors.TcpAuthEntry;
+import org.thingsboard.server.transport.tcp.adaptors.TcpTransportAdaptor;
+import org.thingsboard.server.transport.tcp.adaptors.TcpUpEntry;
+import org.thingsboard.server.transport.tcp.script.TkScriptFactory;
+import org.thingsboard.server.transport.tcp.script.TkScriptInvokeService;
+import org.thingsboard.server.transport.tcp.session.TcpDeviceSessionCtx;
+import org.thingsboard.server.transport.tcp.session.TcpGatewaySessionHandler;
+
+/**
+ * @author Andrew Shvayka
+ */
+@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 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
+
+    }
+
+
+    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);
+        });
+
+    }
+
+
+    /**
+     * 上行脚本解析结果是否包含数据
+     * @param datas 数据集合
+     * @return
+     */
+    private boolean hasDatas(Map<String, Object> datas) {
+        if (datas == null || datas.isEmpty()) {
+            return false;
+        }
+        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 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 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 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();
+        }
+        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.
+                }
+
+                @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());
+
+//            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 {
+      adaptor
+          .convertToPublish(deviceSessionCtx, rpcRequest)
+          .ifPresent(
+              payload -> {
+                deviceSessionCtx.rpcRequesting(payload.getIdentifier(), rpcRequest);
+                Optional.ofNullable(pushDeviceMsg(deviceSessionCtx.getChannel(), payload.getDatas())).ifPresent(cf->{
+                    cf.addListener(
+                            result -> {
+                                if (result.cause() == null) {
+                                    transportService.process(
+                                            deviceSessionCtx.getSessionInfo(),
+                                            rpcRequest,
+                                            RpcStatus.DELIVERED,
+                                            TransportServiceCallback.EMPTY);
+                                } else {
+                                    // 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);
+        }
+    }
+
+
+    @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 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();
+    }
+}

@@ -16,14 +16,7 @@
 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_OPEN;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
+
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFuture;
@@ -31,79 +24,40 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.socket.DatagramPacket;
 import io.netty.channel.socket.DatagramPacket;
 import io.netty.handler.codec.mqtt.*;
 import io.netty.handler.codec.mqtt.*;
-import io.netty.handler.ssl.SslHandler;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.concurrent.GenericFutureListener;
-import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.Map;
-import java.util.Optional;
 import java.util.UUID;
 import java.util.UUID;
-import java.util.concurrent.atomic.AtomicInteger;
-
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.checkerframework.checker.nullness.qual.Nullable;
-import org.thingsboard.common.util.JacksonUtil;
-import org.thingsboard.server.common.data.DataConstants;
-import org.thingsboard.server.common.data.Device;
-import org.thingsboard.server.common.data.DeviceProfile;
-import org.thingsboard.server.common.data.DeviceTransportType;
-import org.thingsboard.server.common.data.device.profile.TkTcpDeviceProfileTransportConfiguration;
-import org.thingsboard.server.common.data.id.DeviceId;
-import org.thingsboard.server.common.data.rpc.RpcStatus;
 import org.thingsboard.server.common.data.yunteng.utils.ByteUtils;
 import org.thingsboard.server.common.data.yunteng.utils.ByteUtils;
-import org.thingsboard.server.common.msg.tools.TbRateLimitsException;
-import org.thingsboard.server.common.transport.SessionMsgListener;
-import org.thingsboard.server.common.transport.TransportService;
-import org.thingsboard.server.common.transport.TransportServiceCallback;
-import org.thingsboard.server.common.transport.adaptor.AdaptorException;
-import org.thingsboard.server.common.transport.auth.SessionInfoCreator;
-import org.thingsboard.server.common.transport.auth.TransportDeviceInfo;
-import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
-import org.thingsboard.server.common.transport.service.DefaultTransportService;
-import org.thingsboard.server.common.transport.service.SessionMetaData;
-import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.transport.tcp.TcpTransportContext;
 import org.thingsboard.server.transport.tcp.TcpTransportContext;
-import org.thingsboard.server.transport.tcp.adaptors.TcpAuthEntry;
-import org.thingsboard.server.transport.tcp.adaptors.TcpTransportAdaptor;
-import org.thingsboard.server.transport.tcp.adaptors.TcpUpEntry;
-import org.thingsboard.server.transport.tcp.script.TkScriptFactory;
-import org.thingsboard.server.transport.tcp.script.TkScriptInvokeService;
-import org.thingsboard.server.transport.tcp.session.TcpDeviceSessionCtx;
-import org.thingsboard.server.transport.tcp.session.TcpGatewaySessionHandler;
 import org.thingsboard.server.transport.tcp.util.ByteBufUtils;
 import org.thingsboard.server.transport.tcp.util.ByteBufUtils;
 
 
 /**
 /**
  * @author Andrew Shvayka
  * @author Andrew Shvayka
  */
  */
 @Slf4j
 @Slf4j
-      public class UdpTransportHandler extends ChannelInboundHandlerAdapter implements GenericFutureListener<Future<? super Void>>, SessionMsgListener {
+      public class UdpTransportHandler extends ChannelInboundHandlerAdapter implements GenericFutureListener<Future<? super Void>> {
 
 
     private final UUID sessionId;
     private final UUID sessionId;
     private final TcpTransportContext context;
     private final TcpTransportContext context;
-    private final TransportService transportService;
-    private final SslHandler sslHandler;
 
 
 
 
     /**
     /**
      * 需要处理的消息队列，例如：需要下发给设备的，设备上传的。
      * 需要处理的消息队列，例如：需要下发给设备的，设备上传的。
      */
      */
-    final TcpDeviceSessionCtx deviceSessionCtx;
-    volatile InetSocketAddress address;
+    private final ConcurrentMap<InetSocketAddress, UdpDatagramDataHandler> deviceSessiones = new ConcurrentHashMap<>();
 
 
-    volatile TcpGatewaySessionHandler gatewaySessionHandler;
-    private final AtomicInteger authedCounter = new AtomicInteger();
 
 
-
-    UdpTransportHandler(TcpTransportContext context, SslHandler sslHandler) {
+    UdpTransportHandler(TcpTransportContext context) {
         super();
         super();
         this.sessionId = UUID.randomUUID();
         this.sessionId = UUID.randomUUID();
         this.context = context;
         this.context = context;
-        this.transportService = context.getTransportService();
-        this.sslHandler = sslHandler;
-        this.deviceSessionCtx = new TcpDeviceSessionCtx(sessionId, context);
     }
     }
 
 
     @Override
     @Override
@@ -123,24 +77,16 @@ import org.thingsboard.server.transport.tcp.util.ByteBufUtils;
         log.error("【{}】 Processing msg: 【{}】", sessionId, msg);
         log.error("【{}】 Processing msg: 【{}】", sessionId, msg);
         try {
         try {
             if(!context.isReady()){
             if(!context.isReady()){
-                ctx.close();
                 return;
                 return;
             }
             }
-            if (address == null) {
-                address = getAddress(ctx);
-            }
             if (msg instanceof DatagramPacket) {
             if (msg instanceof DatagramPacket) {
                 DatagramPacket message = (DatagramPacket) msg;
                 DatagramPacket message = (DatagramPacket) msg;
+                InetSocketAddress address = getAddress(message,ctx);
                 byte[] byteMsg = ByteBufUtils.buf2Bytes(message.content());
                 byte[] byteMsg = ByteBufUtils.buf2Bytes(message.content());
                 String msgStr = ByteUtils.bytesToStr(byteMsg);
                 String msgStr = ByteUtils.bytesToStr(byteMsg);
-                log.error("会话【{}】收到设备【{}】来自【{}】数据【{}】", sessionId, deviceSessionCtx.getDeviceId(), address, msgStr);
-                deviceSessionCtx.setChannel(ctx);
-                if (deviceSessionCtx.getDeviceInfo() == null || deviceSessionCtx.getDeviceProfile() == null) {
-                    processConnect(ctx, msgStr);
-                } else {
-                    enqueueRegularSessionMsg(ctx, msgStr);
-                }
-
+                log.error("UDP服务【{}】收到来自【{}】数据【{}】", sessionId, address, msgStr);
+                UdpDatagramDataHandler dataHandler = deviceSessiones.computeIfAbsent(address,k->new UdpDatagramDataHandler(ctx,context,address));
+                dataHandler.dealDeviceMsg(ctx,msgStr);
             } else {
             } else {
                 log.debug("【{}】 Received non tcp message: 【{}】", sessionId, msg.getClass().getSimpleName());
                 log.debug("【{}】 Received non tcp message: 【{}】", sessionId, msg.getClass().getSimpleName());
                 ctx.close();
                 ctx.close();
@@ -150,8 +96,8 @@ import org.thingsboard.server.transport.tcp.util.ByteBufUtils;
         }
         }
     }
     }
 
 
-    InetSocketAddress getAddress(ChannelHandlerContext ctx) {
-        var address = ctx.channel().attr(UdpTransportService.ADDRESS).get();
+    InetSocketAddress getAddress(DatagramPacket msg,ChannelHandlerContext ctx) {
+        var address = msg.sender();
         if (address == null) {
         if (address == null) {
             log.trace("[{}] Received empty address.", ctx.channel().id());
             log.trace("[{}] Received empty address.", ctx.channel().id());
             InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
             InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
@@ -164,42 +110,9 @@ import org.thingsboard.server.transport.tcp.util.ByteBufUtils;
     }
     }
 
 
 
 
-    void enqueueRegularSessionMsg(ChannelHandlerContext ctx, String msg) {
-        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
-
-    }
 
 
 
 
-    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);
-        });
-
-    }
 
 
 
 
     /**
     /**
@@ -215,129 +128,7 @@ import org.thingsboard.server.transport.tcp.util.ByteBufUtils;
     }
     }
 
 
 
 
-    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 <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 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);
-        }
-    }
 
 
 
 
 
 
@@ -359,159 +150,14 @@ import org.thingsboard.server.transport.tcp.util.ByteBufUtils;
 
 
 
 
 
 
-
-    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
     @Override
     public void operationComplete(Future<? super Void> future) throws Exception {
     public void operationComplete(Future<? super Void> future) throws Exception {
         log.trace("[{}] Channel closed!", sessionId);
         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();
-        }
-        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(), UdpTransportHandler.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());
-
-            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 {
-      adaptor
-          .convertToPublish(deviceSessionCtx, rpcRequest)
-          .ifPresent(
-              payload -> {
-                deviceSessionCtx.rpcRequesting(payload.getIdentifier(), rpcRequest);
-                Optional.ofNullable(pushDeviceMsg(deviceSessionCtx.getChannel(), payload.getDatas())).ifPresent(cf->{
-                    cf.addListener(
-                            result -> {
-                                if (result.cause() == null) {
-                                    transportService.process(
-                                            deviceSessionCtx.getSessionInfo(),
-                                            rpcRequest,
-                                            RpcStatus.DELIVERED,
-                                            TransportServiceCallback.EMPTY);
-                                } else {
-                                    // 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);
-        }
-    }
-
-
-    @Override
-    public void onToServerRpcResponse(TransportProtos.ToServerRpcResponseMsg rpcResponse) {
-        log.debug("[{}] 服务端响应设备的RPC请求", sessionId);
-    }
 
 
     /**
     /**
      * 往设备推送消息
      * 往设备推送消息
@@ -532,21 +178,4 @@ import org.thingsboard.server.transport.tcp.util.ByteBufUtils;
         return ctx.writeAndFlush(buff);
         return ctx.writeAndFlush(buff);
     }
     }
 
 
-
-    @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();
-    }
 }
 }

@@ -39,7 +39,7 @@ public class UdpTransportServerInitializer extends ChannelInitializer<Channel> {
         ChannelPipeline pipeline = ch.pipeline();
         ChannelPipeline pipeline = ch.pipeline();
         SslHandler sslHandler = null;
         SslHandler sslHandler = null;
 
 
-        UdpTransportHandler handler = new UdpTransportHandler(context, sslHandler);
+        UdpTransportHandler handler = new UdpTransportHandler(context);
 //        NettyUdpServerHandler handler = new NettyUdpServerHandler();
 //        NettyUdpServerHandler handler = new NettyUdpServerHandler();
 
 
         pipeline.addLast(handler);
         pipeline.addLast(handler);