Minor, used same const in 2 different test classes

broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java

@@ -22,12 +22,14 @@ import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.impl.security.PermitAllAuthorizator;
 import io.moquette.spi.impl.subscriptions.*;
+import io.moquette.spi.security.IAuthorizator;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.mqtt.*;
 
 import java.util.*;
 
+import static io.moquette.spi.impl.ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR;
 import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;
 import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
 import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE;
@@ -44,9 +46,7 @@ abstract class AbstractProtocolProcessorCommonUtils {
     static final String TEST_USER = "fakeuser";
     static final String TEST_PWD = "fakepwd";
 
-    static final List<InterceptHandler> EMPTY_OBSERVERS = Collections.emptyList();
+    private static final String HELLO_WORLD_MQTT = "Hello world MQTT!!";
-    static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS);
-    public static final String HELLO_WORLD_MQTT = "Hello world MQTT!!";
 
     EmbeddedChannel m_channel;
     ProtocolProcessor m_processor;
@@ -55,7 +55,7 @@ abstract class AbstractProtocolProcessorCommonUtils {
     ISessionsStore m_sessionStore;
     ISubscriptionsDirectory subscriptions;
     MockAuthenticator m_mockAuthenticator;
-    protected SessionsRepository sessionsRepository;
+    SessionsRepository sessionsRepository;
 
     protected void initializeProcessorAndSubsystems() {
         m_channel = new EmbeddedChannel();
@@ -81,10 +81,10 @@ abstract class AbstractProtocolProcessorCommonUtils {
         subscriptions.init(sessionsRepository);
         m_processor = new ProtocolProcessor();
         m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true,
-            new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, sessionsRepository);
+                         new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, sessionsRepository);
     }
 
-    protected void verifyNoPublishIsReceived() {
+    void verifyNoPublishIsReceived() {
         final Object messageReceived = m_channel.readOutbound();
         assertNull("Received an out message from processor while not expected", messageReceived);
     }
@@ -107,6 +107,16 @@ abstract class AbstractProtocolProcessorCommonUtils {
         verifySubscriptionExists(m_sessionStore, expectedSubscription);
     }
 
+    protected void subscribeWithFail(EmbeddedChannel channel, String topic, MqttQoS desiredQos) {
+        MqttSubscribeMessage subscribe = MqttMessageBuilders.subscribe()
+            .addSubscription(desiredQos, topic)
+            .messageId(1)
+            .build();
+        this.m_processor.processSubscribe(channel, subscribe);
+        MqttSubAckMessage subAck = channel.readOutbound();
+        assertEquals(MqttQoS.FAILURE.value(), (int) subAck.payload().grantedQoSLevels().get(0));
+    }
+
     protected static void verifySubscriptionExists(ISessionsStore sessionsStore, Subscription expectedSubscription) {
         final Subscription subscription = sessionsStore.subscriptionStore().reload(expectedSubscription);
         assertEquals(expectedSubscription, subscription);

broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java

@@ -59,16 +59,16 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
     static final String EVIL_TEST_USER = "eviluser";
     static final String EVIL_TEST_PWD = "unsecret";
 
-    static final List<InterceptHandler> EMPTY_OBSERVERS = Collections.emptyList();
+    public static final List<InterceptHandler> EMPTY_OBSERVERS = Collections.emptyList();
-    static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS);
+    public static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS);
 
     @Before
-    public void setUp() throws InterruptedException {
+    public void setUp() {
         initializeProcessorAndSubsystems();
     }
 
     @Test
-    public void testPublishToItself() throws InterruptedException {
+    public void testPublishToItself() {
         final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE);
 
         // subscriptions.matches(topic) redefine the method to return true
@@ -101,7 +101,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
     }
 
     @Test
-    public void testPublishToMultipleSubscribers() throws InterruptedException {
+    public void testPublishToMultipleSubscribers() {
         final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE);
         final Subscription subscriptionClient2 = new Subscription(
                 FAKE_CLIENT_ID2,

broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_InternalPublish_Test.java

@@ -16,6 +16,9 @@
 package io.moquette.spi.impl;
 
 import io.moquette.server.netty.NettyUtils;
+import io.moquette.spi.impl.security.PermitAllAuthorizator;
+import io.moquette.spi.impl.subscriptions.Topic;
+import io.moquette.spi.security.IAuthorizator;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import org.junit.Before;
@@ -27,7 +30,7 @@ import static org.junit.Assert.assertEquals;
 public class ProtocolProcessor_InternalPublish_Test extends AbstractProtocolProcessorCommonUtils {
 
     @Before
-    public void setUp() throws InterruptedException {
+    public void setUp() {
         initializeProcessorAndSubsystems();
         connect();
     }

broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java

@@ -27,7 +27,7 @@ import static org.junit.Assert.assertFalse;
 public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProcessorCommonUtils {
 
     @Before
-    public void setUp() throws InterruptedException {
+    public void setUp() {
         initializeProcessorAndSubsystems();
     }