[broker-15] working on test coverage

JavaSaBr · JavaSaBr · commit 0ffaf7b4f37f · 2019-11-21T09:48:30.000+03:00
diff --git a/src/main/java/com/ss/mqtt/broker/config/MqttConnectionConfig.java b/src/main/java/com/ss/mqtt/broker/config/MqttConnectionConfig.java
@@ -1,7 +1,9 @@
 package com.ss.mqtt.broker.config;
 
 import com.ss.mqtt.broker.model.QoS;
+import lombok.Data;
 import lombok.Getter;
+import lombok.NoArgsConstructor;
 import lombok.RequiredArgsConstructor;
 import org.jetbrains.annotations.NotNull;
 
@@ -24,4 +26,7 @@ public class MqttConnectionConfig {
     private final boolean wildcardSubscriptionAvailable;
     private final boolean subscriptionIdAvailable;
     private final boolean sharedSubscriptionAvailable;
+
+
+    @NoArgsConstructor @Data class MyClass { final String field = "";}
 }
diff --git a/src/main/java/com/ss/mqtt/broker/handler/packet/in/PublishAckInPacketHandler.java b/src/main/java/com/ss/mqtt/broker/handler/packet/in/PublishAckInPacketHandler.java
@@ -10,6 +10,6 @@ public class PublishAckInPacketHandler extends AbstractPacketHandler<UnsafeMqttC
 
     @Override
     protected void handleImpl(@NotNull UnsafeMqttClient client, @NotNull PublishAckInPacket packet) {
-        client.getSession().unregisterPendingPacket(client, packet);
+        client.getSession().updatePendingPacket(client, packet);
     }
 }
diff --git a/src/main/java/com/ss/mqtt/broker/model/MqttPropertyConstants.java b/src/main/java/com/ss/mqtt/broker/model/MqttPropertyConstants.java
@@ -25,8 +25,8 @@ public interface MqttPropertyConstants {
 
     boolean PAYLOAD_FORMAT_INDICATOR_DEFAULT = false;
 
-    long MESSAGE_EXPIRY_INTERVAL_DEFAULT = 0;
     long MESSAGE_EXPIRY_INTERVAL_UNDEFINED = -1;
+    long MESSAGE_EXPIRY_INTERVAL_INFINITY = 0;
 
     int TOPIC_ALIAS_MAXIMUM_UNDEFINED = -1;
     int TOPIC_ALIAS_MAXIMUM_DISABLED = 0;
diff --git a/src/main/java/com/ss/mqtt/broker/model/MqttSession.java b/src/main/java/com/ss/mqtt/broker/model/MqttSession.java
@@ -36,6 +36,8 @@ interface PendingPacketHandler {
     void removeExpiredPackets();
     void resendPendingPacketsAsync(@NotNull MqttClient client, int retryInterval);
 
+    boolean hasPendingPackets();
+
     void registerPendingPublish(@NotNull PublishInPacket publish, @NotNull PendingPacketHandler handler, int packetId);
-    void unregisterPendingPacket(@NotNull MqttClient client, @NotNull HasPacketId response);
+    void updatePendingPacket(@NotNull MqttClient client, @NotNull HasPacketId response);
 }
diff --git a/src/main/java/com/ss/mqtt/broker/model/impl/DefaultMqttSession.java b/src/main/java/com/ss/mqtt/broker/model/impl/DefaultMqttSession.java
@@ -74,6 +74,11 @@ public void registerPendingPublish(
         pendingPublishes.runInWriteLock(pendingPublish, Array::add);
     }
 
+    @Override
+    public boolean hasPendingPackets() {
+        return !pendingPublishes.isEmpty();
+    }
+
     @Override
     public void removeExpiredPackets() {
 
@@ -93,13 +98,15 @@ public void removeExpiredPackets() {
                 var publish = pendingPublish.publish;
                 var messageExpiryInterval = publish.getMessageExpiryInterval();
 
-                if (messageExpiryInterval == MqttPropertyConstants.MESSAGE_EXPIRY_INTERVAL_UNDEFINED) {
+                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--;
@@ -113,19 +120,26 @@ public void resendPendingPacketsAsync(@NotNull MqttClient client, int retryInter
         var currentTime = System.currentTimeMillis();
         var stamp = pendingPublishes.readLock();
         try {
+
             for (var pendingPublish : pendingPublishes) {
-                if (currentTime - pendingPublish.lastAttemptTime > retryInterval) {
-                    pendingPublish.lastAttemptTime = currentTime;
-                    pendingPublish.handler.retryAsync(client, pendingPublish.publish, pendingPublish.packetId);
+
+                if (currentTime - pendingPublish.lastAttemptTime <= retryInterval) {
+                    continue;
                 }
+
+                log.debug("Re-try to send publish {}", pendingPublish.publish);
+
+                pendingPublish.lastAttemptTime = currentTime;
+                pendingPublish.handler.retryAsync(client, pendingPublish.publish, pendingPublish.packetId);
             }
+
         } finally {
             pendingPublishes.readUnlock(stamp);
         }
     }
 
     @Override
-    public void unregisterPendingPacket(
+    public void updatePendingPacket(
         @NotNull MqttClient client,
         @NotNull HasPacketId response
     ) {
diff --git a/src/main/java/com/ss/mqtt/broker/network/packet/in/PublishInPacket.java b/src/main/java/com/ss/mqtt/broker/network/packet/in/PublishInPacket.java
@@ -265,7 +265,7 @@ public class PublishInPacket extends MqttReadablePacket {
 
     private @NotNull byte[] correlationData;
 
-    private long messageExpiryInterval = MqttPropertyConstants.MESSAGE_EXPIRY_INTERVAL_DEFAULT;
+    private long messageExpiryInterval = MqttPropertyConstants.MESSAGE_EXPIRY_INTERVAL_UNDEFINED;
     private int topicAlias = MqttPropertyConstants.TOPIC_ALIAS_DEFAULT;
     private boolean payloadFormatIndicator = MqttPropertyConstants.PAYLOAD_FORMAT_INDICATOR_DEFAULT;
 
@@ -315,8 +315,8 @@ protected void applyProperty(@NotNull PacketProperty property, long value) {
                 topicAlias = NumberUtils.validate(
                     (int) value,
                     MqttPropertyConstants.TOPIC_ALIAS_MIN,
-                    MqttPropertyConstants.TOPIC_ALIAS_MAX)
-                ;
+                    MqttPropertyConstants.TOPIC_ALIAS_MAX
+                );
                 break;
             case MESSAGE_EXPIRY_INTERVAL:
                 messageExpiryInterval = value;
diff --git a/src/main/java/com/ss/mqtt/broker/network/packet/out/Publish5OutPacket.java b/src/main/java/com/ss/mqtt/broker/network/packet/out/Publish5OutPacket.java
@@ -176,7 +176,7 @@ protected void writeProperties(@NotNull ByteBuffer buffer) {
         writeProperty(buffer,
             PacketProperty.MESSAGE_EXPIRY_INTERVAL,
             0,
-            MqttPropertyConstants.MESSAGE_EXPIRY_INTERVAL_DEFAULT
+            MqttPropertyConstants.MESSAGE_EXPIRY_INTERVAL_UNDEFINED
         );
         writeProperty(buffer, PacketProperty.TOPIC_ALIAS, topicAlias, MqttPropertyConstants.TOPIC_ALIAS_DEFAULT);
         writeNotEmptyProperty(buffer, PacketProperty.RESPONSE_TOPIC, responseTopic);
diff --git a/src/main/java/com/ss/mqtt/broker/service/PublishRetryService.java b/src/main/java/com/ss/mqtt/broker/service/PublishRetryService.java
@@ -8,4 +8,6 @@ public interface PublishRetryService {
     void register(@NotNull MqttClient client);
 
     void unregister(@NotNull MqttClient client);
+
+    boolean exist(@NotNull String clientId);
 }
diff --git a/src/main/java/com/ss/mqtt/broker/service/impl/DefaultPublishRetryService.java b/src/main/java/com/ss/mqtt/broker/service/impl/DefaultPublishRetryService.java
@@ -6,10 +6,12 @@
 import com.ss.rlib.common.util.array.Array;
 import com.ss.rlib.common.util.array.ArrayFactory;
 import com.ss.rlib.common.util.array.ConcurrentArray;
+import lombok.extern.log4j.Log4j2;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.Closeable;
 
+@Log4j2
 public class DefaultPublishRetryService implements PublishRetryService, Closeable {
 
     private final @NotNull ConcurrentArray<MqttClient> registeredClients;
@@ -39,6 +41,11 @@ public void unregister(@NotNull MqttClient client) {
         registeredClients.runInWriteLock(client, Array::fastRemove);
     }
 
+    @Override
+    public boolean exist(@NotNull String clientId) {
+        return registeredClients.anyMatchInReadLock(clientId, (id, client) -> id.equals(client.getClientId()));
+    }
+
     private void checkPendingPackets() {
 
         var toCheck = ArrayFactory.newArray(MqttClient.class);
@@ -53,6 +60,8 @@ private void checkPendingPackets() {
 
             registeredClients.runInReadLock(toCheck, Array::copyTo);
 
+            log.debug("Check pending packets in {} client(s)", toCheck.size());
+
             for (var client : toCheck) {
 
                 var session = client.getSession();
@@ -68,7 +77,7 @@ private void checkPendingPackets() {
             }
 
             if (toUnregister.isEmpty()) {
-                return;
+                continue;
             }
 
             // unregister closed clients
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/UnitSpecification.groovy b/src/test/groovy/com/ss/mqtt/broker/test/UnitSpecification.groovy
@@ -0,0 +1,6 @@
+package com.ss.mqtt.broker.test
+
+import spock.lang.Specification
+
+class UnitSpecification extends Specification {
+}
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/integration/ConnectSubscribePublishTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/integration/ConnectSubscribePublishTest.groovy
@@ -8,7 +8,7 @@ import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAckReasonCo
 
 import java.util.concurrent.atomic.AtomicReference
 
-class ConnectSubscribePublishTest extends MqttBrokerTest {
+class ConnectSubscribePublishTest extends IntegrationSpecification {
     
     def "publisher should publish message QoS 0"() {
         given:
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/integration/ConnectionTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/integration/ConnectionTest.groovy
@@ -8,7 +8,7 @@ import org.springframework.beans.factory.annotation.Autowired
 
 import java.util.concurrent.CompletionException
 
-class ConnectionTest extends MqttBrokerTest {
+class ConnectionTest extends IntegrationSpecification {
     
     @Autowired
     InetSocketAddress deviceNetworkAddress
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/integration/IntegrationSpecification.groovy b/src/test/groovy/com/ss/mqtt/broker/test/integration/IntegrationSpecification.groovy
@@ -10,7 +10,7 @@ import spock.lang.Specification
 import java.nio.charset.StandardCharsets
 
 @ContextConfiguration(classes = MqttBrokerTestConfig)
-class MqttBrokerTest extends Specification {
+class IntegrationSpecification extends Specification {
 
     @Autowired
     InetSocketAddress deviceNetworkAddress
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/integration/service/ClientIdRegistryTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/integration/service/ClientIdRegistryTest.groovy
@@ -2,11 +2,11 @@ package com.ss.mqtt.broker.test.integration.service
 
 import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAckReasonCode
 import com.ss.mqtt.broker.service.ClientIdRegistry
-import com.ss.mqtt.broker.test.integration.MqttBrokerTest
+import com.ss.mqtt.broker.test.integration.IntegrationSpecification
 import com.ss.rlib.common.util.StringUtils
 import org.springframework.beans.factory.annotation.Autowired
 
-class ClientIdRegistryTest extends MqttBrokerTest {
+class ClientIdRegistryTest extends IntegrationSpecification {
     
     @Autowired
     ClientIdRegistry clientIdRegistry
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/integration/service/MqttSessionServiceTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/integration/service/MqttSessionServiceTest.groovy
@@ -4,10 +4,10 @@ import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAckReasonCo
 import com.ss.mqtt.broker.config.MqttConnectionConfig
 import com.ss.mqtt.broker.service.ClientIdRegistry
 import com.ss.mqtt.broker.service.MqttSessionService
-import com.ss.mqtt.broker.test.integration.MqttBrokerTest
+import com.ss.mqtt.broker.test.integration.IntegrationSpecification
 import org.springframework.beans.factory.annotation.Autowired
 
-class MqttSessionServiceTest extends MqttBrokerTest {
+class MqttSessionServiceTest extends IntegrationSpecification {
     
     @Autowired
     ClientIdRegistry clientIdRegistry
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/integration/service/PublishRetryServiceTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/integration/service/PublishRetryServiceTest.groovy
@@ -0,0 +1,105 @@
+package com.ss.mqtt.broker.test.integration.service
+
+import com.ss.mqtt.broker.model.MqttSession
+import com.ss.mqtt.broker.network.client.MqttClient
+import com.ss.mqtt.broker.network.packet.in.PublishInPacket
+import com.ss.mqtt.broker.service.MqttSessionService
+import com.ss.mqtt.broker.service.PublishRetryService
+import com.ss.mqtt.broker.test.integration.IntegrationSpecification
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.beans.factory.annotation.Value
+
+import java.util.concurrent.atomic.AtomicInteger
+
+class PublishRetryServiceTest extends IntegrationSpecification {
+    
+    @Autowired
+    PublishRetryService publishRetryService
+    
+    @Autowired
+    MqttSessionService mqttSessionService
+    
+    @Value('${publish.pending.check.interval}')
+    int checkInterval
+    
+    @Value('${publish.retry.interval}')
+    int retryInterval
+    
+    def "client should be presented in re-try service after connecting"() {
+        given:
+            def client = buildClient()
+            def config = client.getConfig()
+            def identifier = config.getClientIdentifier()
+        when:
+            client.connect().join()
+        then:
+            noExceptionThrown()
+            publishRetryService.exist(identifier.get().toString())
+        when:
+            client.disconnect().join()
+            Thread.sleep(500)
+        then:
+            noExceptionThrown()
+            !publishRetryService.exist(identifier.get().toString())
+    }
+    
+    def "service should check expired and pending messages from time to time"() {
+        given:
+            
+            def session = Mock(MqttSession)
+            def client = Stub(MqttClient) {
+                getSession() >> session
+            }
+        
+        when:
+            publishRetryService.register(client)
+            Thread.sleep(checkInterval)
+        then:
+            1 * session.removeExpiredPackets()
+            1 * session.resendPendingPacketsAsync(client, retryInterval)
+        cleanup:
+            publishRetryService.unregister(client)
+    }
+    
+    def "service should remove expired and retry pending messages from time to time"() {
+        given:
+    
+            def session = mqttSessionService.create(UUID.randomUUID().toString()).block()
+            def retryAttempts = new AtomicInteger()
+        
+            def handler = Stub(MqttSession.PendingPacketHandler) {
+                retryAsync(_,_,_) >> {
+                    retryAttempts.incrementAndGet()
+                }
+            }
+            def client = Stub(MqttClient) {
+                getSession() >> session
+            }
+    
+            publishRetryService.register(client)
+        when:
+            def publish = Stub(PublishInPacket) {
+                getMessageExpiryInterval() >> 1L
+            }
+        
+            session.registerPendingPublish(publish, handler, 1)
+        
+            Thread.sleep(1000 + checkInterval)
+        then:
+            retryAttempts.get() == 0
+            !session.hasPendingPackets()
+        when:
+            publish = Stub(PublishInPacket) {
+                getMessageExpiryInterval() >> 1000L
+            }
+        
+            session.registerPendingPublish(publish, handler, 2)
+    
+            Thread.sleep(retryInterval + checkInterval)
+        then:
+            retryAttempts.get() == 1
+            session.hasPendingPackets()
+        cleanup:
+            publishRetryService.unregister(client)
+    }
+}
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/network/NetworkUnitSpecification.groovy b/src/test/groovy/com/ss/mqtt/broker/test/network/NetworkUnitSpecification.groovy
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/network/in/BaseInPacketTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/network/in/BaseInPacketTest.groovy
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/network/in/PublishInPacketTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/network/in/PublishInPacketTest.groovy
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/network/out/BaseOutPacketTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/network/out/BaseOutPacketTest.groovy
diff --git a/src/test/groovy/com/ss/mqtt/broker/test/util/MqttDataUtilsTest.groovy b/src/test/groovy/com/ss/mqtt/broker/test/util/MqttDataUtilsTest.groovy
diff --git a/src/test/resources/application-test.properties b/src/test/resources/application-test.properties

Original file line number	Diff line number	Diff line change
`@@ -10,6 +10,6 @@ public class PublishAckInPacketHandler extends AbstractPacketHandler<UnsafeMqttC`
`10`	`10`
`11`	`11`	`@Override`
`12`	`12`	`protected void handleImpl(@NotNull UnsafeMqttClient client, @NotNull PublishAckInPacket packet) {`
`13`		`- client.getSession().unregisterPendingPacket(client, packet);`
	`13`	`+ client.getSession().updatePendingPacket(client, packet);`
`14`	`14`	`}`
`15`	`15`	`}`
Original file line number	Diff line number	Diff line change
`@@ -36,6 +36,8 @@ interface PendingPacketHandler {`
`36`	`36`	`void removeExpiredPackets();`
`37`	`37`	`void resendPendingPacketsAsync(@NotNull MqttClient client, int retryInterval);`
`38`	`38`
	`39`	`+ boolean hasPendingPackets();`
	`40`	`+`
`39`	`41`	`void registerPendingPublish(@NotNull PublishInPacket publish, @NotNull PendingPacketHandler handler, int packetId);`
`40`		`- void unregisterPendingPacket(@NotNull MqttClient client, @NotNull HasPacketId response);`
	`42`	`+ void updatePendingPacket(@NotNull MqttClient client, @NotNull HasPacketId response);`
`41`	`43`	`}`
Original file line number	Diff line number	Diff line change
`@@ -8,4 +8,6 @@ public interface PublishRetryService {`
`8`	`8`	`void register(@NotNull MqttClient client);`
`9`	`9`
`10`	`10`	`void unregister(@NotNull MqttClient client);`
	`11`	`+`
	`12`	`+ boolean exist(@NotNull String clientId);`
`11`	`13`	`}`