feat: UDP协议接入设备功能

+/**
+ * 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.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.socket.DatagramPacket;
+import io.netty.handler.codec.mqtt.*;
+import io.netty.handler.ssl.SslHandler;
+import io.netty.util.ReferenceCountUtil;
+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;
+import org.thingsboard.server.transport.tcp.util.ByteBufUtils;
+
+/**
+ * @author Andrew Shvayka
+ */
+@Slf4j
+      public class UdpTransportHandler extends ChannelInboundHandlerAdapter implements GenericFutureListener<Future<? super Void>>, SessionMsgListener {
+
+    private final UUID sessionId;
+    private final TcpTransportContext context;
+    private final TransportService transportService;
+    private final SslHandler sslHandler;
+
+
+    /**
+     * 需要处理的消息队列，例如：需要下发给设备的，设备上传的。
+     */
+    final TcpDeviceSessionCtx deviceSessionCtx;
+    volatile InetSocketAddress address;
+
+    volatile TcpGatewaySessionHandler gatewaySessionHandler;
+    private final AtomicInteger authedCounter = new AtomicInteger();
+
+
+    UdpTransportHandler(TcpTransportContext context, SslHandler sslHandler) {
+        super();
+        this.sessionId = UUID.randomUUID();
+        this.context = context;
+        this.transportService = context.getTransportService();
+        this.sslHandler = sslHandler;
+        this.deviceSessionCtx = new TcpDeviceSessionCtx(sessionId, context);
+    }
+
+    @Override
+    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
+        super.channelRegistered(ctx);
+        context.channelRegistered();
+    }
+
+    @Override
+    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
+        super.channelUnregistered(ctx);
+        context.channelUnregistered();
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        log.error("【{}】 Processing msg: 【{}】", sessionId, msg);
+        try {
+            if(!context.isReady()){
+                ctx.close();
+                return;
+            }
+            if (address == null) {
+                address = getAddress(ctx);
+            }
+            if (msg instanceof DatagramPacket) {
+                DatagramPacket message = (DatagramPacket) msg;
+                byte[] byteMsg = ByteBufUtils.buf2Bytes(message.content());
+                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);
+                }
+
+            } else {
+                log.debug("【{}】 Received non tcp message: 【{}】", sessionId, msg.getClass().getSimpleName());
+                ctx.close();
+            }
+        } finally {
+            ReferenceCountUtil.safeRelease(msg);
+        }
+    }
+
+    InetSocketAddress getAddress(ChannelHandlerContext ctx) {
+        var address = ctx.channel().attr(UdpTransportService.ADDRESS).get();
+        if (address == null) {
+            log.trace("[{}] Received empty address.", ctx.channel().id());
+            InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
+            log.trace("[{}] Going to use address: {}", ctx.channel().id(), remoteAddress);
+            return remoteAddress;
+        } else {
+            log.trace("[{}] Received address: {}", ctx.channel().id(), address);
+        }
+        return address;
+    }
+
+
+    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);
+        });
+
+    }
+
+
+    /**
+     * 上行脚本解析结果是否包含数据
+     * @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);
+        }
+    }
+
+
+
+
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) {
+        ctx.flush();
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        log.error("[{}] Unexpected Exception", sessionId, cause);
+        ctx.close();
+        if (cause instanceof OutOfMemoryError) {
+            log.error("Received critical error. Going to shutdown the service.");
+            System.exit(1);
+        }
+    }
+
+
+
+
+    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(), 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);
+    }
+
+    /**
+     * 往设备推送消息
+     *
+     * @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();
+    }
+}

+/**
+ * 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 io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.ssl.SslHandler;
+import org.thingsboard.server.transport.tcp.TcpTransportContext;
+
+/**
+ * @author Andrew Shvayka
+ */
+public class UdpTransportServerInitializer extends ChannelInitializer<Channel> {
+
+    private final TcpTransportContext context;
+    private final boolean sslEnabled;
+
+    public UdpTransportServerInitializer(TcpTransportContext context, boolean sslEnabled) {
+        this.context = context;
+        this.sslEnabled = sslEnabled;
+    }
+
+    @Override
+    public void initChannel(Channel ch) {
+        ChannelPipeline pipeline = ch.pipeline();
+        SslHandler sslHandler = null;
+
+        UdpTransportHandler handler = new UdpTransportHandler(context, sslHandler);
+//        NettyUdpServerHandler handler = new NettyUdpServerHandler();
+
+        pipeline.addLast(handler);
+        ch.closeFuture().addListener(handler);
+    }
+
+}

+/**
+ * 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 io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.util.AttributeKey;
+import io.netty.util.ResourceLeakDetector;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.TbTransportService;
+import org.thingsboard.server.transport.tcp.TcpTransportContext;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.net.InetSocketAddress;
+
+/**
+ * @author Andrew Shvayka
+ */
+@Service("UdpTransportService")
+@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.tcp.enabled}'=='true')")
+@Slf4j
+public class UdpTransportService implements TbTransportService {
+
+    public static AttributeKey<InetSocketAddress> ADDRESS = AttributeKey.newInstance("UDP_SRC_ADDRESS");
+
+    @Value("${transport.tcp.bind_address}")
+    private String host;
+    @Value("${transport.tcp.bind_port}")
+    private Integer port;
+
+    @Value("${transport.tcp.ssl.enabled}")
+    private boolean sslEnabled;
+
+    @Value("${transport.tcp.ssl.bind_address}")
+    private String sslHost;
+    @Value("${transport.tcp.ssl.bind_port}")
+    private Integer sslPort;
+
+    @Value("${transport.tcp.netty.leak_detector_level}")
+    private String leakDetectorLevel;
+    @Value("${transport.tcp.netty.boss_group_thread_count}")
+    private Integer bossGroupThreadCount;
+    @Value("${transport.tcp.netty.worker_group_thread_count}")
+    private Integer workerGroupThreadCount;
+    @Value("${transport.tcp.netty.so_keep_alive}")
+    private boolean keepAlive;
+
+    @Autowired
+    private TcpTransportContext context;
+
+    private Channel serverChannel;
+    private Channel sslServerChannel;
+    private EventLoopGroup bossGroup;
+
+    @PostConstruct
+    public void init() throws Exception {
+        log.info("Setting resource leak detector level to {}", leakDetectorLevel);
+        ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.valueOf(leakDetectorLevel.toUpperCase()));
+
+        log.info("Starting TCP transport...");
+        bossGroup = new NioEventLoopGroup(bossGroupThreadCount);
+        Bootstrap b = new Bootstrap();
+    b.group(bossGroup)
+        .channel(NioDatagramChannel.class)
+            .option(ChannelOption.SO_BROADCAST, true)
+            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
+            .handler(new UdpTransportServerInitializer(context, false));
+        serverChannel = b.bind(host, port).sync().channel();
+        if (sslEnabled) {
+            b = new Bootstrap();
+      b.group(bossGroup)
+          .channel(NioDatagramChannel.class)
+          .handler(new UdpTransportServerInitializer(context, false))
+          .option(ChannelOption.SO_BROADCAST, true);
+            sslServerChannel = b.bind(sslHost, sslPort).sync().channel();
+        }
+        log.info("TCP transport started!");
+    }
+
+    @PreDestroy
+    public void shutdown() throws InterruptedException {
+        log.info("Stopping TCP transport!");
+        try {
+            serverChannel.close().sync();
+            if (sslEnabled) {
+                sslServerChannel.close().sync();
+            }
+        } finally {
+            bossGroup.shutdownGracefully();
+        }
+        log.info("TCP transport stopped!");
+    }
+
+    @Override
+    public String getName() {
+        return DataConstants.TCP_TRANSPORT_NAME;
+    }
+}