PublishMsg rename qosLevel to qos

thingsboard · Nov 18, 2024 · 7c701b9 · 7c701b9
1 parent fbfa6ee
commit 7c701b9
Show file tree

Hide file tree

Showing 13 changed files with 25 additions and 25 deletions.
diff --git a/...n/java/org/thingsboard/mqtt/broker/actors/client/service/handlers/MqttPublishHandler.java b/...n/java/org/thingsboard/mqtt/broker/actors/client/service/handlers/MqttPublishHandler.java
@@ -112,10 +112,10 @@ public void process(ClientSessionCtx ctx, MqttPublishMsg msg, TbActorRef actorRe
 
         MqttMsgWrapper mqttMsgWrapper = null; // for QoS 0
         try {
-            if (MqttQoS.EXACTLY_ONCE.value() == publishMsg.getQosLevel()) {
+            if (MqttQoS.EXACTLY_ONCE.value() == publishMsg.getQos()) {
                 mqttMsgWrapper = processExactlyOnce(ctx, msgId);
                 if (ensureMsgPersistedAwaitingPubRel(ctx, actorRef, mqttMsgWrapper)) return;
-            } else if (MqttQoS.AT_LEAST_ONCE.value() == publishMsg.getQosLevel()) {
+            } else if (MqttQoS.AT_LEAST_ONCE.value() == publishMsg.getQos()) {
                 mqttMsgWrapper = processAtLeastOnce(ctx, msgId);
             }
         } catch (FullMsgQueueException e) {
@@ -180,9 +180,9 @@ private void handleMsgPersistenceFailure(ClientSessionCtx ctx, PublishMsg publis
 
     private void handleMqtt5ErrorResponse(ClientSessionCtx ctx, PublishMsg publishMsg,
                                           MqttReasonCodes.PubRec pubRecCode, MqttReasonCodes.PubAck pubAckCode) {
-        if (publishMsg.getQosLevel() == 2) {
+        if (publishMsg.getQos() == 2) {
             pushPubRecErrorResponseWithReasonCode(ctx, publishMsg, pubRecCode);
-        } else if (publishMsg.getQosLevel() == 1) {
+        } else if (publishMsg.getQos() == 1) {
             pushPubAckErrorResponseWithReasonCode(ctx, publishMsg, pubAckCode);
         } else {
             // QoS=0 - do nothing
@@ -218,7 +218,7 @@ public void onSuccess(TbQueueMsgMetadata metadata) {
                     if (isTraceEnabled) {
                         log.trace("[{}][{}] Successfully acknowledged msg: {}", ctx.getClientId(), ctx.getSessionId(), publishMsg.getPacketId());
                     }
-                    sendPubResponseEventToActor(actorRef, ctx.getSessionId(), mqttMsgWrapper, MqttQoS.valueOf(publishMsg.getQosLevel()));
+                    sendPubResponseEventToActor(actorRef, ctx.getSessionId(), mqttMsgWrapper, MqttQoS.valueOf(publishMsg.getQos()));
                 });
             }
 

diff --git a/application/src/main/java/org/thingsboard/mqtt/broker/adaptor/NettyMqttConverter.java b/application/src/main/java/org/thingsboard/mqtt/broker/adaptor/NettyMqttConverter.java
@@ -39,9 +39,9 @@
 import org.thingsboard.mqtt.broker.actors.client.messages.mqtt.MqttPublishMsg;
 import org.thingsboard.mqtt.broker.actors.client.messages.mqtt.MqttSubscribeMsg;
 import org.thingsboard.mqtt.broker.actors.client.messages.mqtt.MqttUnsubscribeMsg;
+import org.thingsboard.mqtt.broker.common.data.BrokerConstants;
 import org.thingsboard.mqtt.broker.common.data.subscription.SubscriptionOptions;
 import org.thingsboard.mqtt.broker.common.data.subscription.TopicSubscription;
-import org.thingsboard.mqtt.broker.common.data.BrokerConstants;
 import org.thingsboard.mqtt.broker.service.mqtt.PublishMsg;
 import org.thingsboard.mqtt.broker.session.ClientSessionCtx;
 import org.thingsboard.mqtt.broker.session.DisconnectReason;
@@ -185,7 +185,7 @@ private static PublishMsg extractPublishMsg(MqttPublishMessage mqttPublishMessag
         return PublishMsg.builder()
                 .packetId(mqttPublishMessage.variableHeader().packetId())
                 .topicName(mqttPublishMessage.variableHeader().topicName())
-                .qosLevel(mqttPublishMessage.fixedHeader().qosLevel().value())
+                .qos(mqttPublishMessage.fixedHeader().qosLevel().value())
                 .isRetained(mqttPublishMessage.fixedHeader().isRetain())
                 .isDup(mqttPublishMessage.fixedHeader().isDup())
                 .byteBuf(byteBuf)
@@ -200,7 +200,7 @@ private static PublishMsg extractLastWillPublishMsg(MqttConnectMessage msg) {
                 .payload(msg.payload().willMessageInBytes())
                 .isRetained(msg.variableHeader().isWillRetain())
                 .isDup(false)
-                .qosLevel(msg.variableHeader().willQos())
+                .qos(msg.variableHeader().willQos())
                 .properties(msg.payload().willProperties())
                 .build();
     }

diff --git a/application/src/main/java/org/thingsboard/mqtt/broker/adaptor/ProtoConverter.java b/application/src/main/java/org/thingsboard/mqtt/broker/adaptor/ProtoConverter.java
@@ -60,7 +60,7 @@ public static QueueProtos.PublishMsgProto convertToPublishMsgProto(SessionInfo s
         QueueProtos.PublishMsgProto.Builder builder = QueueProtos.PublishMsgProto.newBuilder()
                 .setPacketId(publishMsg.getPacketId())
                 .setTopicName(publishMsg.getTopicName())
-                .setQos(publishMsg.getQosLevel())
+                .setQos(publishMsg.getQos())
                 .setRetain(publishMsg.isRetained())
                 .addAllUserProperties(toUserPropertyProtos(userProperties))
                 .setClientId(sessionInfo.getClientInfo().getClientId());
@@ -155,7 +155,7 @@ public static PublishMsg convertToPublishMsg(QueueProtos.PublishMsgProto msg, in
                 .payload(msg.getPayload().toByteArray())
                 .properties(properties)
                 .packetId(packetId)
-                .qosLevel(qos)
+                .qos(qos)
                 .isDup(isDup)
                 .build();
     }

diff --git a/application/src/main/java/org/thingsboard/mqtt/broker/server/MqttSessionHandler.java b/application/src/main/java/org/thingsboard/mqtt/broker/server/MqttSessionHandler.java
@@ -223,7 +223,7 @@ private void processPublish(MqttMessage msg) {
                 rateLimitBatchProcessor.addMessage(mqttPublishMsg,
                         mqttMsg -> clientMqttActorManager.processMqttMsg(clientId, mqttMsg),
                         mqttMsg -> {
-                            processMsgOnRateLimits(mqttMsg.getPublishMsg().getPacketId(), mqttMsg.getPublishMsg().getQosLevel(), "Total rate limits detected");
+                            processMsgOnRateLimits(mqttMsg.getPublishMsg().getPacketId(), mqttMsg.getPublishMsg().getQos(), "Total rate limits detected");
                             mqttMsg.release();
                         });
                 return;

diff --git a/...ion/src/main/java/org/thingsboard/mqtt/broker/service/mqtt/DefaultMqttMessageCreator.java b/...ion/src/main/java/org/thingsboard/mqtt/broker/service/mqtt/DefaultMqttMessageCreator.java
@@ -202,7 +202,7 @@ public MqttMessage createPubCompMsg(int msgId, MqttReasonCodes.PubComp code) {
 
     @Override
     public MqttPublishMessage createPubMsg(PublishMsg pubMsg) {
-        return getMqttPublishMessage(pubMsg.isDup(), pubMsg.getQosLevel(), pubMsg.isRetained(),
+        return getMqttPublishMessage(pubMsg.isDup(), pubMsg.getQos(), pubMsg.isRetained(),
                 pubMsg.getTopicName(), pubMsg.getPacketId(), pubMsg.getPayload(), pubMsg.getProperties());
     }
 

diff --git a/.../main/java/org/thingsboard/mqtt/broker/service/mqtt/DefaultPublishMsgDeliveryService.java b/.../main/java/org/thingsboard/mqtt/broker/service/mqtt/DefaultPublishMsgDeliveryService.java
@@ -135,7 +135,7 @@ public void sendPublishMsgToClientWithoutFlush(ClientSessionCtx sessionCtx, Publ
         pubMsg = sessionCtx.getTopicAliasCtx().createPublishMsgUsingTopicAlias(pubMsg, minTopicNameLengthForAliasReplacement);
         MqttPublishMessage mqttPubMsg = mqttMessageGenerator.createPubMsg(pubMsg);
         tbMessageStatsReportClient.reportStats(OUTGOING_MSGS);
-        tbMessageStatsReportClient.reportClientReceiveStats(sessionCtx.getClientId(), pubMsg.getQosLevel());
+        tbMessageStatsReportClient.reportClientReceiveStats(sessionCtx.getClientId(), pubMsg.getQos());
         sendPublishMsgWithoutFlushToClient(sessionCtx, mqttPubMsg);
     }
 

diff --git a/application/src/main/java/org/thingsboard/mqtt/broker/service/mqtt/PublishMsg.java b/application/src/main/java/org/thingsboard/mqtt/broker/service/mqtt/PublishMsg.java
@@ -33,17 +33,17 @@ public class PublishMsg {
     private final String topicName;
     private final byte[] payload;
     private final ByteBuf byteBuf;
-    private final int qosLevel;
+    private final int qos;
     private final boolean isRetained;
     private final boolean isDup;
     private final MqttProperties properties;
 
-    public PublishMsg(int packetId, String topicName, byte[] payload, int qosLevel, boolean isRetained, boolean isDup) {
+    public PublishMsg(int packetId, String topicName, byte[] payload, int qos, boolean isRetained, boolean isDup) {
         this.packetId = packetId;
         this.topicName = topicName;
         this.payload = payload;
         this.byteBuf = Unpooled.wrappedBuffer(payload);
-        this.qosLevel = qosLevel;
+        this.qos = qos;
         this.isRetained = isRetained;
         this.isDup = isDup;
         this.properties = MqttProperties.NO_PROPERTIES;

diff --git a/.../broker/service/mqtt/persistence/application/processing/ApplicationPackProcessingCtx.java b/.../broker/service/mqtt/persistence/application/processing/ApplicationPackProcessingCtx.java
@@ -61,7 +61,7 @@ public ApplicationPackProcessingCtx(ApplicationSubmitStrategy submitStrategy, Ap
             switch (persistedMsg.getPacketType()) {
                 case PUBLISH:
                     PersistedPublishMsg persistedPublishMsg = (PersistedPublishMsg) persistedMsg;
-                    if (persistedPublishMsg.isSharedSubscriptionMsg() && persistedPublishMsg.getPublishMsg().getQosLevel() == 0) {
+                    if (persistedPublishMsg.isSharedSubscriptionMsg() && persistedPublishMsg.getPublishMsg().getQos() == 0) {
                         break;
                     }
                     publishPendingMsgMap.put(persistedMsg.getPacketId(), persistedPublishMsg);

diff --git a/...c/main/java/org/thingsboard/mqtt/broker/service/mqtt/retain/RetainedMsgProcessorImpl.java b/...c/main/java/org/thingsboard/mqtt/broker/service/mqtt/retain/RetainedMsgProcessorImpl.java
@@ -54,7 +54,7 @@ private boolean payloadIsEmpty(PublishMsg publishMsg) {
 
     private RetainedMsg newRetainedMsg(PublishMsg publishMsg) {
         MqttProperties properties = MqttPropertiesUtil.getMqttProperties(publishMsg);
-        return new RetainedMsg(publishMsg.getTopicName(), publishMsg.getPayload(), publishMsg.getQosLevel(), properties);
+        return new RetainedMsg(publishMsg.getTopicName(), publishMsg.getPayload(), publishMsg.getQos(), properties);
     }
 
 }
diff --git a/...rc/main/java/org/thingsboard/mqtt/broker/service/processing/MsgDispatcherServiceImpl.java b/...rc/main/java/org/thingsboard/mqtt/broker/service/processing/MsgDispatcherServiceImpl.java
@@ -30,8 +30,8 @@
 import org.thingsboard.mqtt.broker.common.data.ClientType;
 import org.thingsboard.mqtt.broker.common.data.MqttQoS;
 import org.thingsboard.mqtt.broker.common.data.SessionInfo;
-import org.thingsboard.mqtt.broker.common.data.util.StringUtils;
 import org.thingsboard.mqtt.broker.common.data.subscription.SubscriptionOptions;
+import org.thingsboard.mqtt.broker.common.data.util.StringUtils;
 import org.thingsboard.mqtt.broker.common.stats.MessagesStats;
 import org.thingsboard.mqtt.broker.gen.queue.QueueProtos.PublishMsgProto;
 import org.thingsboard.mqtt.broker.queue.TbQueueCallback;
@@ -107,7 +107,7 @@ public void persistPublishMsg(SessionInfo sessionInfo, PublishMsg publishMsg, Tb
         PublishMsgProto publishMsgProto = ProtoConverter.convertToPublishMsgProto(sessionInfo, publishMsg);
         producerStats.incrementTotal();
         tbMessageStatsReportClient.reportStats(INCOMING_MSGS);
-        tbMessageStatsReportClient.reportClientSendStats(sessionInfo.getClientId(), publishMsg.getQosLevel());
+        tbMessageStatsReportClient.reportClientSendStats(sessionInfo.getClientId(), publishMsg.getQos());
         callback = statsManager.wrapTbQueueCallback(callback, producerStats);
 
         DefaultTbQueueMsgHeaders headers = createHeaders(publishMsg);

diff --git a/application/src/test/java/org/thingsboard/mqtt/broker/adaptor/NettyMqttConverterTest.java b/application/src/test/java/org/thingsboard/mqtt/broker/adaptor/NettyMqttConverterTest.java
@@ -50,10 +50,10 @@
 import org.thingsboard.mqtt.broker.actors.client.messages.mqtt.MqttPublishMsg;
 import org.thingsboard.mqtt.broker.actors.client.messages.mqtt.MqttSubscribeMsg;
 import org.thingsboard.mqtt.broker.actors.client.messages.mqtt.MqttUnsubscribeMsg;
+import org.thingsboard.mqtt.broker.common.data.BrokerConstants;
 import org.thingsboard.mqtt.broker.common.data.SessionInfo;
 import org.thingsboard.mqtt.broker.common.data.subscription.SubscriptionOptions;
 import org.thingsboard.mqtt.broker.common.data.subscription.TopicSubscription;
-import org.thingsboard.mqtt.broker.common.data.BrokerConstants;
 import org.thingsboard.mqtt.broker.service.mqtt.PublishMsg;
 import org.thingsboard.mqtt.broker.session.ClientSessionCtx;
 import org.thingsboard.mqtt.broker.session.DisconnectReasonType;
@@ -224,7 +224,7 @@ public void testCreateMqttPublishMsg() {
         PublishMsg publishMsg = mqttPublishMsg.getPublishMsg();
         Assert.assertEquals(1, publishMsg.getPacketId());
         Assert.assertEquals("topic", publishMsg.getTopicName());
-        Assert.assertEquals(MqttQoS.AT_LEAST_ONCE.value(), publishMsg.getQosLevel());
+        Assert.assertEquals(MqttQoS.AT_LEAST_ONCE.value(), publishMsg.getQos());
         Assert.assertFalse(publishMsg.isRetained());
         Assert.assertFalse(publishMsg.isDup());
     }

diff --git a/application/src/test/java/org/thingsboard/mqtt/broker/adaptor/ProtoConverterTest.java b/application/src/test/java/org/thingsboard/mqtt/broker/adaptor/ProtoConverterTest.java
@@ -367,7 +367,7 @@ public void givenPubMsg_whenExecuteConvertToPublishMsgProto_thenGetExpectedPubMs
         PublishMsg publishMsg = PublishMsg.builder()
                 .topicName("topic")
                 .packetId(1)
-                .qosLevel(2)
+                .qos(2)
                 .properties(properties)
                 .payload("p".getBytes(StandardCharsets.UTF_8))
                 .build();
@@ -391,7 +391,7 @@ public void givenPubMsgProto_whenExecuteConvertToPublishMsg_thenGetExpectedPubMs
 
         assertEquals("t", publishMsg.getTopicName());
         assertEquals(1, publishMsg.getPacketId());
-        assertEquals(1, publishMsg.getQosLevel());
+        assertEquals(1, publishMsg.getQos());
         assertNotNull(publishMsg.getProperties().getProperty(BrokerConstants.PAYLOAD_FORMAT_INDICATOR_PROP_ID));
         assertNull(publishMsg.getProperties().getProperty(BrokerConstants.CONTENT_TYPE_PROP_ID));
         assertNull(publishMsg.getProperties().getProperty(BrokerConstants.SUBSCRIPTION_IDENTIFIER_PROP_ID));

diff --git a/...ce/mqtt/persistence/application/processing/ApplicationPersistedMsgCtxServiceImplTest.java b/...ce/mqtt/persistence/application/processing/ApplicationPersistedMsgCtxServiceImplTest.java
@@ -129,7 +129,7 @@ private PublishMsg buildPubMsg() {
                 .packetId(1)
                 .topicName("topic")
                 .payload("data".getBytes())
-                .qosLevel(1)
+                .qos(1)
                 .isRetained(false)
                 .isDup(false)
                 .build();