From aab1161c796d55b369b1f82a5376a61d0faf59c8 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Tue, 3 Sep 2024 10:58:40 +0200 Subject: [PATCH] create pojo for session limits --- .../rpc/MqttClientSideRpcIntegrationTest.java | 58 +++++++++---------- .../transport/mqtt/MqttTransportHandler.java | 41 +++++++------ .../mqtt/limits/GatewaySessionLimits.java | 25 ++++++++ .../transport/mqtt/limits/SessionLimits.java | 30 ++++++++++ 4 files changed, 106 insertions(+), 48 deletions(-) create mode 100644 common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/limits/GatewaySessionLimits.java create mode 100644 common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/limits/SessionLimits.java diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttClientSideRpcIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttClientSideRpcIntegrationTest.java index 6c08b61f09..2a700e484d 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttClientSideRpcIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/rpc/MqttClientSideRpcIntegrationTest.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.transport.mqtt.mqttv3.rpc; -import com.fasterxml.jackson.core.type.TypeReference; import io.netty.handler.codec.mqtt.MqttQoS; import org.junit.Test; import org.springframework.beans.factory.annotation.Value; @@ -26,16 +25,16 @@ import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileCon import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; +import org.thingsboard.server.transport.mqtt.limits.GatewaySessionLimits; +import org.thingsboard.server.transport.mqtt.limits.SessionLimits; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestCallback; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestSubscribeOnTopicCallback; -import java.util.Map; import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_RPC_REQUESTS_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_RPC_RESPONSE_SUB_TOPIC; @@ -62,6 +61,15 @@ public class MqttClientSideRpcIntegrationTest extends AbstractMqttIntegrationTes doPost("/api/tenantProfile", tenantProfile); + var expectedLimits = new SessionLimits(); + var deviceLimits = new SessionLimits.SessionRateLimits(profileConfiguration.getTransportDeviceMsgRateLimit(), + profileConfiguration.getTransportDeviceTelemetryMsgRateLimit(), + profileConfiguration.getTransportDeviceTelemetryDataPointsRateLimit()); + expectedLimits.setRateLimits(deviceLimits); + expectedLimits.setMaxPayloadSize(maxPayloadSize); + expectedLimits.setMaxInflightMessages(maxInflightMessages); + expectedLimits.setPayloadType(TransportPayloadType.JSON); + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() .deviceName("Test Get Service Configuration") .transportPayloadType(TransportPayloadType.JSON) @@ -81,18 +89,8 @@ public class MqttClientSideRpcIntegrationTest extends AbstractMqttIntegrationTes .as("await callback").isTrue(); var payload = callback.getPayloadBytes(); - Map response = JacksonUtil.fromBytes(payload, new TypeReference<>() {}); - - assertNotNull(response); - assertEquals(4, response.size()); - assertEquals(response.get("maxPayloadSize"), maxPayloadSize); - assertEquals(response.get("maxInflightMessages"), maxInflightMessages); - assertEquals(response.get("payloadType"), TransportPayloadType.JSON.name()); - Map rateLimits = (Map) response.get("rateLimits"); - assertEquals(3, rateLimits.size()); - assertEquals(rateLimits.get("messages"), profileConfiguration.getTransportDeviceMsgRateLimit()); - assertEquals(rateLimits.get("telemetryMessages"), profileConfiguration.getTransportDeviceTelemetryMsgRateLimit()); - assertEquals(rateLimits.get("telemetryDataPoints"), profileConfiguration.getTransportDeviceTelemetryDataPointsRateLimit()); + SessionLimits actualLimits = JacksonUtil.fromBytes(payload, SessionLimits.class); + assertEquals(expectedLimits, actualLimits); client.disconnect(); } @@ -113,6 +111,19 @@ public class MqttClientSideRpcIntegrationTest extends AbstractMqttIntegrationTes doPost("/api/tenantProfile", tenantProfile); + var expectedLimits = new GatewaySessionLimits(); + var gatewayLimits = new SessionLimits.SessionRateLimits(profileConfiguration.getTransportGatewayMsgRateLimit(), + profileConfiguration.getTransportGatewayTelemetryMsgRateLimit(), + profileConfiguration.getTransportGatewayTelemetryDataPointsRateLimit()); + var gatewayDeviceLimits = new SessionLimits.SessionRateLimits(profileConfiguration.getTransportGatewayDeviceMsgRateLimit(), + profileConfiguration.getTransportGatewayDeviceTelemetryMsgRateLimit(), + profileConfiguration.getTransportGatewayDeviceTelemetryDataPointsRateLimit()); + expectedLimits.setGatewayRateLimits(gatewayLimits); + expectedLimits.setRateLimits(gatewayDeviceLimits); + expectedLimits.setMaxPayloadSize(maxPayloadSize); + expectedLimits.setMaxInflightMessages(maxInflightMessages); + expectedLimits.setPayloadType(TransportPayloadType.JSON); + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() .gatewayName("Test Get Service Configuration Gateway") .transportPayloadType(TransportPayloadType.JSON) @@ -131,21 +142,8 @@ public class MqttClientSideRpcIntegrationTest extends AbstractMqttIntegrationTes assertTrue(callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)); var payload = callback.getPayloadBytes(); - Map response = JacksonUtil.fromBytes(payload, new TypeReference<>() {}); - - assertNotNull(response); - assertEquals(4, response.size()); - assertEquals(response.get("maxPayloadSize"), maxPayloadSize); - assertEquals(response.get("maxInflightMessages"), maxInflightMessages); - assertEquals(response.get("payloadType"), TransportPayloadType.JSON.name()); - Map rateLimits = (Map) response.get("rateLimits"); - assertEquals(6, rateLimits.size()); - assertEquals(rateLimits.get("messages"), profileConfiguration.getTransportGatewayMsgRateLimit()); - assertEquals(rateLimits.get("telemetryMessages"), profileConfiguration.getTransportGatewayTelemetryMsgRateLimit()); - assertEquals(rateLimits.get("telemetryDataPoints"), profileConfiguration.getTransportGatewayTelemetryDataPointsRateLimit()); - assertEquals(rateLimits.get("deviceMessages"), profileConfiguration.getTransportGatewayDeviceMsgRateLimit()); - assertEquals(rateLimits.get("deviceTelemetryMessages"), profileConfiguration.getTransportGatewayDeviceTelemetryMsgRateLimit()); - assertEquals(rateLimits.get("deviceTelemetryDataPoints"), profileConfiguration.getTransportGatewayDeviceTelemetryDataPointsRateLimit()); + SessionLimits actualLimits = JacksonUtil.fromBytes(payload, GatewaySessionLimits.class); + assertEquals(expectedLimits, actualLimits); client.disconnect(); } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index 0fce2048a4..823e558370 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -79,6 +79,8 @@ import org.thingsboard.server.gen.transport.mqtt.SparkplugBProto; import org.thingsboard.server.queue.scheduler.SchedulerComponent; import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor; +import org.thingsboard.server.transport.mqtt.limits.GatewaySessionLimits; +import org.thingsboard.server.transport.mqtt.limits.SessionLimits; import org.thingsboard.server.transport.mqtt.session.DeviceSessionCtx; import org.thingsboard.server.transport.mqtt.session.GatewaySessionHandler; import org.thingsboard.server.transport.mqtt.session.MqttTopicMatcher; @@ -97,9 +99,7 @@ import java.security.cert.X509Certificate; import java.util.ArrayList; import java.util.Collections; import java.util.Date; -import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.Optional; import java.util.UUID; import java.util.concurrent.Callable; @@ -936,7 +936,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement } } else { log.debug("[{}] Failed to process unsubscription [{}] to [{}] - Subscription not found", sessionId, mqttMsg.variableHeader().messageId(), topicName); - unSubResults.add((short)MqttReasonCodes.UnsubAck.NO_SUBSCRIPTION_EXISTED.byteValue()); + unSubResults.add((short) MqttReasonCodes.UnsubAck.NO_SUBSCRIPTION_EXISTED.byteValue()); } } if (!activityReported) { @@ -1341,25 +1341,30 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement private void onGetSessionLimitsRpc(TransportProtos.SessionInfoProto sessionInfo, ChannelHandlerContext ctx, int msgId, TransportProtos.ToServerRpcRequestMsg rpcRequestMsg) { var tenantProfile = context.getTenantProfileCache().get(deviceSessionCtx.getTenantId()); DefaultTenantProfileConfiguration profile = tenantProfile.getDefaultProfileConfiguration(); - Map sessionLimits = new HashMap<>(); - Map rateLimits = new HashMap<>(); + + SessionLimits sessionLimits; if (sessionInfo.getIsGateway()) { - rateLimits.put("messages", profile.getTransportGatewayMsgRateLimit()); - rateLimits.put("telemetryMessages", profile.getTransportGatewayTelemetryMsgRateLimit()); - rateLimits.put("telemetryDataPoints", profile.getTransportGatewayTelemetryDataPointsRateLimit()); - rateLimits.put("deviceMessages", profile.getTransportGatewayDeviceMsgRateLimit()); - rateLimits.put("deviceTelemetryMessages", profile.getTransportGatewayDeviceTelemetryMsgRateLimit()); - rateLimits.put("deviceTelemetryDataPoints", profile.getTransportGatewayDeviceTelemetryDataPointsRateLimit()); + var gatewaySessionLimits = new GatewaySessionLimits(); + var gatewayLimits = new SessionLimits.SessionRateLimits(profile.getTransportGatewayMsgRateLimit(), + profile.getTransportGatewayTelemetryMsgRateLimit(), + profile.getTransportGatewayTelemetryDataPointsRateLimit()); + var gatewayDeviceLimits = new SessionLimits.SessionRateLimits(profile.getTransportGatewayDeviceMsgRateLimit(), + profile.getTransportGatewayDeviceTelemetryMsgRateLimit(), + profile.getTransportGatewayDeviceTelemetryDataPointsRateLimit()); + gatewaySessionLimits.setGatewayRateLimits(gatewayLimits); + gatewaySessionLimits.setRateLimits(gatewayDeviceLimits); + sessionLimits = gatewaySessionLimits; } else { - rateLimits.put("messages", profile.getTransportDeviceMsgRateLimit()); - rateLimits.put("telemetryMessages", profile.getTransportDeviceTelemetryMsgRateLimit()); - rateLimits.put("telemetryDataPoints", profile.getTransportDeviceTelemetryDataPointsRateLimit()); + var rateLimits = new SessionLimits.SessionRateLimits(profile.getTransportDeviceMsgRateLimit(), + profile.getTransportDeviceTelemetryMsgRateLimit(), + profile.getTransportDeviceTelemetryDataPointsRateLimit()); + sessionLimits = new SessionLimits(); + sessionLimits.setRateLimits(rateLimits); } - sessionLimits.put("rateLimits", rateLimits); - sessionLimits.put("maxPayloadSize", context.getMaxPayloadSize()); - sessionLimits.put("maxInflightMessages", context.getMessageQueueSizePerDeviceLimit()); - sessionLimits.put("payloadType", deviceSessionCtx.getPayloadType()); + sessionLimits.setMaxPayloadSize(context.getMaxPayloadSize()); + sessionLimits.setMaxInflightMessages(context.getMessageQueueSizePerDeviceLimit()); + sessionLimits.setPayloadType(deviceSessionCtx.getPayloadType()); ack(ctx, msgId, MqttReasonCodes.PubAck.SUCCESS); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/limits/GatewaySessionLimits.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/limits/GatewaySessionLimits.java new file mode 100644 index 0000000000..6b59f6bbbd --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/limits/GatewaySessionLimits.java @@ -0,0 +1,25 @@ +/** + * Copyright © 2016-2024 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.mqtt.limits; + +import lombok.Data; + +@Data +public class GatewaySessionLimits extends SessionLimits { + + private SessionRateLimits gatewayRateLimits; + +} diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/limits/SessionLimits.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/limits/SessionLimits.java new file mode 100644 index 0000000000..7651d27386 --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/limits/SessionLimits.java @@ -0,0 +1,30 @@ +/** + * Copyright © 2016-2024 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.mqtt.limits; + +import lombok.Data; +import org.thingsboard.server.common.data.TransportPayloadType; + +@Data +public class SessionLimits { + + private int maxPayloadSize; + private int maxInflightMessages; + private TransportPayloadType payloadType; + private SessionRateLimits rateLimits; + + public record SessionRateLimits(String messages, String telemetryMessages, String telemetryDataPoints) {} +}