[broker-15] fix for code review

Author: JavaSaBr

src/main/java/com/ss/mqtt/broker/model/MqttPropertyConstants.java

@@ -52,5 +52,4 @@ public interface MqttPropertyConstants {
     boolean SUBSCRIPTION_IDENTIFIER_AVAILABLE_DEFAULT = false;
 
     int PACKET_ID_FOR_QOS_0 = 0;
-    int PACKET_ID_MAX = 0xFFFF;
 }

src/main/java/com/ss/mqtt/broker/model/impl/DefaultMqttSession.java

@@ -81,38 +81,9 @@ public boolean hasPendingPackets() {
 
     @Override
     public void removeExpiredPackets() {
-
-        if (pendingPublishes.isEmpty()) {
-            return;
+        if (!pendingPublishes.isEmpty()) {
+            pendingPublishes.runInWriteLock(this::removeExpiredPackets);
         }
-
-        pendingPublishes.runInWriteLock(publishes -> {
-
-            var currentTime = System.currentTimeMillis();
-            var array = publishes.array();
-
-            for (int i = 0, length = publishes.size(); i < length; i++) {
-
-                var pendingPublish = array[i];
-
-                var publish = pendingPublish.publish;
-                var messageExpiryInterval = publish.getMessageExpiryInterval();
-
-                if (messageExpiryInterval == MqttPropertyConstants.MESSAGE_EXPIRY_INTERVAL_UNDEFINED ||
-                    messageExpiryInterval == MqttPropertyConstants.MESSAGE_EXPIRY_INTERVAL_INFINITY) {
-                    continue;
-                }
-
-                var expiredTime = pendingPublish.registeredTime + (messageExpiryInterval * 1000);
-
-                if (expiredTime < currentTime) {
-                    log.debug("Remove pending publish {} by expiration reason", publish);
-                    publishes.fastRemove(i);
-                    i--;
-                    length--;
-                }
-            }
-        });
     }
 
     @Override
@@ -167,4 +138,32 @@ public void updatePendingPacket(
     public void clear() {
         pendingPublishes.runInWriteLock(Collection::clear);
     }
+
+    private void removeExpiredPackets(@NotNull Array<PendingPublish> publishes) {
+
+        var currentTime = System.currentTimeMillis();
+        var array = publishes.array();
+
+        for (int i = 0, length = publishes.size(); i < length; i++) {
+
+            var pendingPublish = array[i];
+
+            var publish = pendingPublish.publish;
+            var messageExpiryInterval = publish.getMessageExpiryInterval();
+
+            if (messageExpiryInterval == MqttPropertyConstants.MESSAGE_EXPIRY_INTERVAL_UNDEFINED ||
+                messageExpiryInterval == MqttPropertyConstants.MESSAGE_EXPIRY_INTERVAL_INFINITY) {
+                continue;
+            }
+
+            var expiredTime = pendingPublish.registeredTime + (messageExpiryInterval * 1000);
+
+            if (expiredTime < currentTime) {
+                log.debug("Remove pending publish {} by expiration reason", publish);
+                publishes.fastRemove(i);
+                i--;
+                length--;
+            }
+        }
+    }
 }

src/main/java/com/ss/mqtt/broker/service/impl/InMemoryMqttSessionService.java

@@ -64,7 +64,11 @@ public InMemoryMqttSessionService(int cleanInterval) {
     }
 
     @Override
-    public @NotNull Mono<Boolean> store(@NotNull String clientId, @NotNull MqttSession session, long expiryInterval) {
+    public @NotNull Mono<Boolean> store(
+        @NotNull String clientId,
+        @NotNull MqttSession session,
+        long expiryInterval
+    ) {
 
         var unsafe = (UnsafeMqttSession) session;
         unsafe.setExpirationTime(System.currentTimeMillis() + (expiryInterval * 1000));