From 176b986c4725b2a27b8ba885f0b9adcba2e28ffc Mon Sep 17 00:00:00 2001 From: andsel Date: Fri, 2 Nov 2018 16:49:50 +0100 Subject: [PATCH] Introduced interceptors notification into new broker --- .../java/io/moquette/broker/PostOffice.java | 39 ++-- .../main/java/io/moquette/broker/Server.java | 189 ++++++++++++++++-- .../io/moquette/broker/SessionRegistry.java | 3 +- .../main/java/io/moquette/server/Server.java | 3 + .../moquette/spi/impl/BrokerInterceptor.java | 6 +- .../moquette/broker/ConnectionTestUtils.java | 9 + .../broker/MQTTConnectionConnectTest.java | 5 +- .../broker/MQTTConnectionPublishTest.java | 5 +- .../broker/PostOfficeInternalPublishTest.java | 7 +- .../broker/PostOfficePublishTest.java | 5 +- .../broker/PostOfficeSubscribeTest.java | 7 +- .../broker/PostOfficeUnsubscribeTest.java | 5 +- .../moquette/broker/SessionRegistryTest.java | 5 +- 13 files changed, 238 insertions(+), 50 deletions(-) diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/broker/src/main/java/io/moquette/broker/PostOffice.java index 5d511e33..b80cd6a2 100644 --- a/broker/src/main/java/io/moquette/broker/PostOffice.java +++ b/broker/src/main/java/io/moquette/broker/PostOffice.java @@ -1,5 +1,7 @@ package io.moquette.broker; +import io.moquette.server.netty.NettyUtils; +import io.moquette.spi.impl.BrokerInterceptor; import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; @@ -26,13 +28,15 @@ class PostOffice { private final ISubscriptionsDirectory subscriptions; private final IRetainedRepository retainedRepository; private SessionRegistry sessionRegistry; + private BrokerInterceptor interceptor; PostOffice(ISubscriptionsDirectory subscriptions, IAuthorizatorPolicy authorizatorPolicy, - IRetainedRepository retainedRepository, SessionRegistry sessionRegistry) { + IRetainedRepository retainedRepository, SessionRegistry sessionRegistry, BrokerInterceptor interceptor) { this.authorizator = new Authorizator(authorizatorPolicy); this.subscriptions = subscriptions; this.retainedRepository = retainedRepository; this.sessionRegistry = sessionRegistry; + this.interceptor = interceptor; } public void init(SessionRegistry sessionRegistry) { @@ -72,10 +76,9 @@ class PostOffice { publishRetainedMessagesForSubscriptions(clientID, newSubscriptions); - // TODO notify the Observables -// for (Subscription subscription : newSubscriptions) { -// m_interceptor.notifyTopicSubscribed(newSubscription, username); -// } + for (Subscription subscription : newSubscriptions) { + interceptor.notifyTopicSubscribed(subscription, username); + } } private void publishRetainedMessagesForSubscriptions(String clientID, List newSubscriptions) { @@ -132,9 +135,8 @@ class PostOffice { // TODO remove the subscriptions to Session // clientSession.unsubscribeFrom(topic); - //TODO notify interceptors -// String username = NettyUtils.userName(channel); -// m_interceptor.notifyTopicUnsubscribed(topic.toString(), clientID, username); + String username = NettyUtils.userName(mqttConnection.channel); + interceptor.notifyTopicUnsubscribed(topic.toString(), clientID, username); } // ack the client @@ -152,8 +154,11 @@ class PostOffice { // QoS == 0 && retain => clean old retained retainedRepository.cleanRetained(topic); } -// TODO -// m_interceptor.notifyTopicPublished(msg, clientID, username); + + MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, AT_MOST_ONCE, false, 0); + MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic.toString(), 0); + MqttPublishMessage msg = new MqttPublishMessage(fixedHeader, varHeader, payload.retainedDuplicate()); + interceptor.notifyTopicPublished(msg, clientID, username); } void receivedPublishQos1(MQTTConnection connection, Topic topic, String username, ByteBuf payload, int messageID, @@ -183,8 +188,7 @@ class PostOffice { retainedRepository.retain(topic, msg); } } -//TODO -// m_interceptor.notifyTopicPublished(msg, clientID, username); + interceptor.notifyTopicPublished(msg, clientId, username); } private void publish2Subscribers(ByteBuf origPayload, Topic topic, MqttQoS publishingQos) { @@ -230,8 +234,9 @@ class PostOffice { } } - //TODO here we should notify to the listeners - //m_interceptor.notifyTopicPublished(msg, clientID, username); + String clientID = connection.getClientId(); + String username = NettyUtils.userName(connection.channel); + interceptor.notifyTopicPublished(mqttPublishMessage, clientID, username); } static MqttQoS lowerQosToTheSubscriptionDesired(Subscription sub, MqttQoS qos) { @@ -249,11 +254,9 @@ class PostOffice { * where it's publishing. * * @param msg - * the message to publish. - * @param clientId - * the clientID + * the message to publish */ - public void internalPublish(MqttPublishMessage msg, final String clientId) { + public void internalPublish(MqttPublishMessage msg) { final MqttQoS qos = msg.fixedHeader().qosLevel(); final Topic topic = new Topic(msg.variableHeader().topicName()); final ByteBuf payload = msg.payload(); diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java index e36bcc85..9c358642 100644 --- a/broker/src/main/java/io/moquette/broker/Server.java +++ b/broker/src/main/java/io/moquette/broker/Server.java @@ -3,11 +3,9 @@ package io.moquette.broker; import io.moquette.BrokerConstants; import io.moquette.interception.InterceptHandler; import io.moquette.persistence.MemoryStorageService; -import io.moquette.server.config.FileResourceLoader; -import io.moquette.server.config.IConfig; -import io.moquette.server.config.IResourceLoader; -import io.moquette.server.config.ResourceLoaderConfig; +import io.moquette.server.config.*; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.impl.BrokerInterceptor; import io.moquette.spi.impl.SessionsRepository; import io.moquette.spi.impl.security.AcceptAllAuthenticator; import io.moquette.spi.impl.security.PermitAllAuthorizatorPolicy; @@ -16,13 +14,17 @@ import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory; import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizatorPolicy; import io.moquette.spi.security.ISslContextCreator; +import io.netty.handler.codec.mqtt.MqttPublishMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; +import java.lang.reflect.InvocationTargetException; +import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Properties; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -30,17 +32,19 @@ import static io.moquette.logging.LoggingUtils.getInterceptorIds; public class Server { - private static final Logger LOG = LoggerFactory.getLogger(io.moquette.server.Server.class); + private static final Logger LOG = LoggerFactory.getLogger(io.moquette.broker.Server.class); private ScheduledExecutorService scheduler; private NewNettyAcceptor acceptor; private volatile boolean initialized; + private PostOffice dispatcher; + private BrokerInterceptor interceptor; public static void main(String[] args) throws IOException { final Server server = new Server(); server.startServer(); System.out.println("Server started, version 0.12-SNAPSHOT"); - //Bind a shutdown hook + //Bind a shutdown hook Runtime.getRuntime().addShutdownHook(new Thread(server::stopServer)); } @@ -57,11 +61,61 @@ public class Server { startServer(config); } + private static File defaultConfigFile() { + String configPath = System.getProperty("moquette.path", null); + return new File(configPath, IConfig.DEFAULT_CONFIG); + } + + /** + * Starts Moquette bringing the configuration from the given file + * + * @param configFile text file that contains the configuration. + * @throws IOException in case of any IO Error. + */ + public void startServer(File configFile) throws IOException { + LOG.info("Starting Moquette server. Configuration file path: {}", configFile.getAbsolutePath()); + IResourceLoader filesystemLoader = new FileResourceLoader(configFile); + final IConfig config = new ResourceLoaderConfig(filesystemLoader); + startServer(config); + } + + /** + * Starts the server with the given properties. + *

+ * Its suggested to at least have the following properties: + *

+ * + * @param configProps the properties map to use as configuration. + * @throws IOException in case of any IO Error. + */ + public void startServer(Properties configProps) throws IOException { + LOG.debug("Starting Moquette server using properties object"); + final IConfig config = new MemoryConfig(configProps); + startServer(config); + } + + /** + * Starts Moquette bringing the configuration files from the given Config implementation. + * + * @param config the configuration to use to start the broker. + * @throws IOException in case of any IO Error. + */ public void startServer(IConfig config) throws IOException { LOG.debug("Starting Moquette server using IConfig instance"); startServer(config, null); } + /** + * Starts Moquette with config provided by an implementation of IConfig class and with the set + * of InterceptHandler. + * + * @param config the configuration to use to start the broker. + * @param handlers the handlers to install in the broker. + * @throws IOException in case of any IO Error. + */ public void startServer(IConfig config, List handlers) throws IOException { LOG.debug("Starting moquette server using IConfig instance and intercept handlers"); startServer(config, handlers, null, null, null); @@ -86,6 +140,7 @@ public class Server { // m_processorBootstrapper = new ProtocolProcessorBootstrapper(); // final ProtocolProcessor processor = m_processorBootstrapper.init(config, handlers, authenticator, authorizator, // this); + initInterceptors(config, handlers); LOG.debug("Initialized MQTT protocol processor"); // if (sslCtxCreator == null) { // LOG.info("Using default SSL context creator"); @@ -103,11 +158,12 @@ public class Server { ISessionsStore sessionStore = memStorage.sessionsStore(); SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null); // DBG - + ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); subscriptions.init(sessionsRepository); - SessionRegistry sessions = new SessionRegistry(subscriptions); - final PostOffice dispatcher = new PostOffice(subscriptions, authorizatorPolicy, new MemoryRetainedRepository(), sessions); + SessionRegistry sessions = new SessionRegistry(subscriptions, interceptor); + dispatcher = new PostOffice(subscriptions, authorizatorPolicy, new MemoryRetainedRepository(), sessions, + interceptor); final BrokerConfiguration brokerConfig = new BrokerConfiguration(config); MQTTConnectionFactory connectionFactory = new MQTTConnectionFactory(brokerConfig, authenticator, sessions, dispatcher); @@ -121,6 +177,76 @@ public class Server { initialized = true; } + private void initInterceptors(IConfig props, List embeddedObservers) { + LOG.info("Configuring message interceptors..."); + + List observers = new ArrayList<>(embeddedObservers); + String interceptorClassName = props.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME); + if (interceptorClassName != null && !interceptorClassName.isEmpty()) { + InterceptHandler handler = loadClass(interceptorClassName, InterceptHandler.class, io.moquette.broker.Server.class, this); + if (handler != null) { + observers.add(handler); + } + } + interceptor = new BrokerInterceptor(props, observers); + } + + @SuppressWarnings("unchecked") + private T loadClass(String className, Class intrface, Class constructorArgClass, U props) { + T instance = null; + try { + // check if constructor with constructor arg class parameter + // exists + LOG.info("Invoking constructor with {} argument. ClassName={}, interfaceName={}", + constructorArgClass.getName(), className, intrface.getName()); + instance = this.getClass().getClassLoader() + .loadClass(className) + .asSubclass(intrface) + .getConstructor(constructorArgClass) + .newInstance(props); + } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) { + LOG.warn("Unable to invoke constructor with {} argument. ClassName={}, interfaceName={}, cause={}, " + + "errorMessage={}", constructorArgClass.getName(), className, intrface.getName(), ex.getCause(), + ex.getMessage()); + return null; + } catch (NoSuchMethodException | InvocationTargetException e) { + try { + LOG.info("Invoking default constructor. ClassName={}, interfaceName={}", className, intrface.getName()); + // fallback to default constructor + instance = this.getClass().getClassLoader() + .loadClass(className) + .asSubclass(intrface) + .getDeclaredConstructor().newInstance(); + } catch (InstantiationException | IllegalAccessException | ClassNotFoundException | + NoSuchMethodException | InvocationTargetException ex) { + LOG.error("Unable to invoke default constructor. ClassName={}, interfaceName={}, cause={}, " + + "errorMessage={}", className, intrface.getName(), ex.getCause(), ex.getMessage()); + return null; + } + } + + return instance; + } + + /** + * Use the broker to publish a message. It's intended for embedding applications. It can be used + * only after the server is correctly started with startServer. + * + * @param msg the message to forward. + * @param clientId the id of the sending server. + * @throws IllegalStateException if the server is not yet started + */ + public void internalPublish(MqttPublishMessage msg, final String clientId) { + final int messageID = msg.variableHeader().messageId(); + if (!initialized) { + LOG.error("Moquette is not started, internal message cannot be published. CId={}, messageId={}", clientId, + messageID); + throw new IllegalStateException("Can't publish on a server is not yet started"); + } + LOG.debug("Publishing message. CId={}, messageId={}", clientId, messageID); + dispatcher.internalPublish(msg); + } + public void stopServer() { LOG.info("Unbinding server from the configured ports"); acceptor.close(); @@ -135,8 +261,47 @@ public class Server { LOG.info("Moquette server has been stopped."); } - private static File defaultConfigFile() { - String configPath = System.getProperty("moquette.path", null); - return new File(configPath, IConfig.DEFAULT_CONFIG); + /** + * SPI method used by Broker embedded applications to get list of subscribers. Returns null if + * the broker is not started. + * + * @return list of subscriptions. + */ +// TODO reimplement this +// public List getSubscriptions() { +// if (m_processorBootstrapper == null) { +// return null; +// } +// return this.subscriptionsStore.listAllSubscriptions(); +// } + + /** + * SPI method used by Broker embedded applications to add intercept handlers. + * + * @param interceptHandler the handler to add. + */ + public void addInterceptHandler(InterceptHandler interceptHandler) { + if (!initialized) { + LOG.error("Moquette is not started, MQTT message interceptor cannot be added. InterceptorId={}", + interceptHandler.getID()); + throw new IllegalStateException("Can't register interceptors on a server that is not yet started"); + } + LOG.info("Adding MQTT message interceptor. InterceptorId={}", interceptHandler.getID()); + interceptor.addInterceptHandler(interceptHandler); + } + + /** + * SPI method used by Broker embedded applications to remove intercept handlers. + * + * @param interceptHandler the handler to remove. + */ + public void removeInterceptHandler(InterceptHandler interceptHandler) { + if (!initialized) { + LOG.error("Moquette is not started, MQTT message interceptor cannot be removed. InterceptorId={}", + interceptHandler.getID()); + throw new IllegalStateException("Can't deregister interceptors from a server that is not yet started"); + } + LOG.info("Removing MQTT message interceptor. InterceptorId={}", interceptHandler.getID()); + interceptor.removeInterceptHandler(interceptHandler); } } diff --git a/broker/src/main/java/io/moquette/broker/SessionRegistry.java b/broker/src/main/java/io/moquette/broker/SessionRegistry.java index 0611878e..e10a16ab 100644 --- a/broker/src/main/java/io/moquette/broker/SessionRegistry.java +++ b/broker/src/main/java/io/moquette/broker/SessionRegistry.java @@ -1,6 +1,7 @@ package io.moquette.broker; import io.moquette.broker.Session.SessionStatus; +import io.moquette.spi.impl.BrokerInterceptor; import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; @@ -47,7 +48,7 @@ class SessionRegistry { private final ISubscriptionsDirectory subscriptionsDirectory; private final ConcurrentMap> queues = new ConcurrentHashMap<>(); - SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory) { + SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory, BrokerInterceptor m_interceptor) { this.subscriptionsDirectory = subscriptionsDirectory; } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 36fcefd8..a3778d26 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -43,7 +43,10 @@ import static io.moquette.logging.LoggingUtils.getInterceptorIds; /** * Launch a configured version of the server. + * + * @deprecated use io.moquette.broker.Server */ +@Deprecated public class Server { private static final Logger LOG = LoggerFactory.getLogger(Server.class); diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index 9c380786..766e43f6 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -39,7 +39,7 @@ import static io.moquette.logging.LoggingUtils.getInterceptorIds; /** * An interceptor that execute the interception tasks asynchronously. */ -final class BrokerInterceptor implements Interceptor { +public final class BrokerInterceptor implements Interceptor { private static final Logger LOG = LoggerFactory.getLogger(BrokerInterceptor.class); private final Map, List> handlers; @@ -62,14 +62,14 @@ final class BrokerInterceptor implements Interceptor { * * @param handlers */ - BrokerInterceptor(List handlers) { + public BrokerInterceptor(List handlers) { this(1, handlers); } /** * Configures a broker interceptor using the pool size specified in the IConfig argument. */ - BrokerInterceptor(IConfig props, List handlers) { + public BrokerInterceptor(IConfig props, List handlers) { this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers); } diff --git a/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java b/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java index b28c1e15..e0d6dae8 100644 --- a/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java +++ b/broker/src/test/java/io/moquette/broker/ConnectionTestUtils.java @@ -1,14 +1,23 @@ package io.moquette.broker; +import io.moquette.interception.InterceptHandler; +import io.moquette.spi.impl.BrokerInterceptor; import io.moquette.spi.impl.DebugUtils; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.mqtt.*; +import java.util.Collections; +import java.util.List; + import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED; import static org.junit.Assert.*; class ConnectionTestUtils { + public static final List EMPTY_OBSERVERS = Collections.emptyList(); + public static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); + + static void assertConnectAccepted(EmbeddedChannel channel) { MqttConnAckMessage connAck = channel.readOutbound(); final MqttConnectReturnCode connAckReturnCode = connAck.variableHeader().connectReturnCode(); diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java index 308dfb8f..6cb585fa 100644 --- a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java +++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java @@ -53,8 +53,9 @@ public class MQTTConnectionConnectTest { SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null); subscriptions.init(sessionsRepository); - sessionRegistry = new SessionRegistry(subscriptions); - postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), sessionRegistry); + sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), + sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); sut = createMQTTConnection(CONFIG); channel = (EmbeddedChannel) sut.channel; diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java index 15ebee4e..97a9ab1d 100644 --- a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java +++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java @@ -57,9 +57,10 @@ public class MQTTConnectionPublishTest { SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null); subscriptions.init(sessionsRepository); - sessionRegistry = new SessionRegistry(subscriptions); + sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); final PostOffice postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), - new MemoryRetainedRepository(), sessionRegistry); + new MemoryRetainedRepository(), sessionRegistry, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); return new MQTTConnection(channel, config, mockAuthenticator, sessionRegistry, postOffice); } diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java index f24e0bcd..4fc989b6 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java @@ -74,8 +74,9 @@ public class PostOfficeInternalPublishTest { subscriptions.init(sessionsRepository); retainedRepository = new MemoryRetainedRepository(); - SessionRegistry sessionRegistry = new SessionRegistry(subscriptions); - sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry); + SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); return sessionRegistry; } @@ -93,7 +94,7 @@ public class PostOfficeInternalPublishTest { .retained(retained) .qos(qos) .payload(Unpooled.copiedBuffer(PAYLOAD.getBytes(UTF_8))).build(); - sut.internalPublish(publish, "INTRPUBL"); + sut.internalPublish(publish); } @Test diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java index 2e6b3f61..f4b14c7c 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java @@ -81,8 +81,9 @@ public class PostOfficePublishTest { subscriptions.init(sessionsRepository); retainedRepository = new MemoryRetainedRepository(); - SessionRegistry sessionRegistry = new SessionRegistry(subscriptions); - sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry); + SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); return sessionRegistry; } diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java index dd59534a..4c5cd4d7 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java @@ -79,9 +79,9 @@ public class PostOfficeSubscribeTest { SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null); subscriptions.init(sessionsRepository); - sessionRegistry = new SessionRegistry(subscriptions); + sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), - sessionRegistry); + sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); } private MQTTConnection createMQTTConnection(BrokerConfiguration config, Channel channel) { @@ -153,7 +153,8 @@ public class PostOfficeSubscribeTest { when(prohibitReadOnNewsTopic.canRead(eq(new Topic(NEWS_TOPIC)), eq(FAKE_USER_NAME), eq(FAKE_CLIENT_ID))) .thenReturn(false); - sut = new PostOffice(subscriptions, prohibitReadOnNewsTopic, new MemoryRetainedRepository(), sessionRegistry); + sut = new PostOffice(subscriptions, prohibitReadOnNewsTopic, new MemoryRetainedRepository(), sessionRegistry, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); connection.processConnect(connectMessage); ConnectionTestUtils.assertConnectAccepted(channel); diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java index a8711a20..88e58371 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java @@ -69,8 +69,9 @@ public class PostOfficeUnsubscribeTest { SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null); subscriptions.init(sessionsRepository); - sessionRegistry = new SessionRegistry(subscriptions); - sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), sessionRegistry); + sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), + sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); } private MQTTConnection createMQTTConnection(BrokerConfiguration config, Channel channel) { diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java index 6e499ece..29893b80 100644 --- a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java +++ b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java @@ -69,9 +69,10 @@ public class SessionRegistryTest { SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null); subscriptions.init(sessionsRepository); - sut = new SessionRegistry(subscriptions); + sut = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); final PostOffice postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), - new MemoryRetainedRepository(), sut); + new MemoryRetainedRepository(), sut, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); return new MQTTConnection(channel, config, mockAuthenticator, sut, postOffice); }