diff --git a/ChangeLog.txt b/ChangeLog.txt index ef9af28e..3b9acaa0 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -1,4 +1,6 @@ Version 0.10: + Separated in memory and MapDB storage implementations + Cleanly separated inflight part (inbound and outbound) Version 0.9: diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java index f2ce2305..9da0f251 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java @@ -1,29 +1,35 @@ - -package io.moquette.interception.messages; - -import static io.moquette.spi.IMessagesStore.StoredMessage; - -public class InterceptAcknowledgedMessage implements InterceptMessage { - - final private StoredMessage msg; - private final String username; - private final String topic; - - public InterceptAcknowledgedMessage(final StoredMessage msg, final String topic, final String username) { - this.msg = msg; - this.username = username; - this.topic = topic; - } - - public StoredMessage getMsg() { - return msg; - } - - public String getUsername() { - return username; - } - - public String getTopic() { - return topic; - } -} + +package io.moquette.interception.messages; + +import static io.moquette.spi.IMessagesStore.StoredMessage; + +public class InterceptAcknowledgedMessage implements InterceptMessage { + + final private StoredMessage msg; + private final String username; + private final String topic; + private final int packetID; + + public InterceptAcknowledgedMessage(StoredMessage msg, String topic, String username, int packetID) { + this.msg = msg; + this.username = username; + this.topic = topic; + this.packetID = packetID; + } + + public StoredMessage getMsg() { + return msg; + } + + public String getUsername() { + return username; + } + + public String getTopic() { + return topic; + } + + public int getPacketID() { + return packetID; + } +} diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index da1812be..ad0f7e25 100644 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -35,7 +35,6 @@ public class MemoryMessagesStore implements IMessagesStore { private Map m_retainedStore = new HashMap<>(); private Map m_persistentMessageStore = new HashMap<>(); - private Map> m_messageToGuids = new HashMap<>(); MemoryMessagesStore() { } @@ -72,21 +71,13 @@ public class MemoryMessagesStore implements IMessagesStore { MessageGUID guid = new MessageGUID(UUID.randomUUID().toString()); storedMessage.setGuid(guid); m_persistentMessageStore.put(guid, storedMessage); - final HashMap emptyGuids = new HashMap<>(); - Map guids = defaultGet(m_messageToGuids, storedMessage.getClientID(), emptyGuids); - guids.put(storedMessage.getMessageID(), guid); - m_messageToGuids.put(storedMessage.getClientID(), guids); return guid; } @Override - public void dropMessagesInSession(String clientID) { - Map messageGUIDMap = m_messageToGuids.get(clientID); - if (messageGUIDMap == null || messageGUIDMap.isEmpty()) { - return; - } + public void dropInFlightMessagesInSession(Collection pendingAckMessages) { //remove all guids from retained - Collection messagesToRemove = new HashSet<>(messageGUIDMap.values()); + Collection messagesToRemove = new HashSet<>(pendingAckMessages); messagesToRemove.removeAll(m_retainedStore.values()); for (MessageGUID guid : messagesToRemove) { @@ -103,20 +94,4 @@ public class MemoryMessagesStore implements IMessagesStore { public void cleanRetained(Topic topic) { m_retainedStore.remove(topic); } - - @Override - public int getPendingPublishMessages(String clientID) { - Map messageToGuids = m_messageToGuids.get(clientID); - if (messageToGuids == null) - return 0; - else - return messageToGuids.size(); - } - - @Override - public MessageGUID mapToGuid(String clientID, int messageID) { - final HashMap emptyGuids = new HashMap<>(); - Map guids = Utils.defaultGet(m_messageToGuids, clientID, emptyGuids); - return guids.get(messageID); - } } diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index 1af19c8f..f59a8e9f 100644 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -31,6 +31,8 @@ import java.util.*; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; +import static io.moquette.spi.impl.Utils.defaultGet; + /** * * @author andrea @@ -45,12 +47,15 @@ public class MemorySessionStore implements ISessionsStore { // maps clientID->[MessageId -> guid] private Map> m_inflightStore = new HashMap<>(); - // private Map> m_inflightIDs = new HashMap<>(); // maps clientID->BlockingQueue private Map> queues = new HashMap<>(); // maps clientID->[MessageId -> guid] private Map> m_secondPhaseStore = new HashMap<>(); + private Map> outboundFlightMessageToGuid = new HashMap<>(); + + private Map> inboundFlightMessageToGuid = new HashMap<>(); + private final IMessagesStore m_messagesStore; public MemorySessionStore(IMessagesStore messagesStore) { @@ -176,6 +181,11 @@ public class MemorySessionStore implements ISessionsStore { } m.put(messageID, guid); this.m_inflightStore.put(clientID, m); + + final HashMap emptyGuids = new HashMap<>(); + Map guids = defaultGet(outboundFlightMessageToGuid, clientID, emptyGuids); + guids.put(messageID, guid); + outboundFlightMessageToGuid.put(clientID, guids); } /** @@ -251,9 +261,25 @@ public class MemorySessionStore implements ISessionsStore { return inflightMessages.size(); } + @Override + public StoredMessage inboundInflight(String clientID, int messageID) { + final HashMap emptyGuids = new HashMap<>(); + Map guids = Utils.defaultGet(inboundFlightMessageToGuid, clientID, emptyGuids); + final MessageGUID guid = guids.get(messageID); + return m_messagesStore.getMessageByGuid(guid); + } + + @Override + public void markAsInboundInflight(String clientID, int messageID, MessageGUID guid) { + final HashMap emptyGuids = new HashMap<>(); + Map guids = Utils.defaultGet(inboundFlightMessageToGuid, clientID, emptyGuids); + guids.put(messageID, guid); + inboundFlightMessageToGuid.put(clientID, guids); + } + @Override public int getPendingPublishMessagesNo(String clientID) { - return m_messagesStore.getPendingPublishMessages(clientID); + return queues.get(clientID).size(); } @Override @@ -265,4 +291,13 @@ public class MemorySessionStore implements ISessionsStore { return pendingAcks.size(); } + @Override + public Collection pendingAck(String clientID) { + Map messageGUIDMap = outboundFlightMessageToGuid.get(clientID); + if (messageGUIDMap == null || messageGUIDMap.isEmpty()) { + return Collections.emptyList(); + } + + return new ArrayList<>(messageGUIDMap.values()); + } } diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java index 5d61dde6..0a75cca4 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java @@ -43,16 +43,10 @@ public class ConnectionDescriptorStore implements IConnectionsManager { ConnectionDescriptor descriptor = connectionDescriptors.get(clientID); if (descriptor == null) { if (messageID != null) { - LOG.error( - "Client has just disconnected. The {} message could not be sent. CId=<{}>, messageId={}", - messageType, - clientID, - messageID); + LOG.error("Client has just disconnected. {} message could not be sent. CId=<{}>, messageId={}", + messageType, clientID, messageID); } else { - LOG.error( - "Client has just disconnected. The {} could not be sent. CId=<{}>", - messageType, - clientID); + LOG.error("Client has just disconnected. {} could not be sent. CId=<{}>", messageType, clientID); } /* * If the client has just disconnected, its connection descriptor will be null. We @@ -103,11 +97,8 @@ public class ConnectionDescriptorStore implements IConnectionsManager { public boolean closeConnection(String clientID, boolean closeImmediately) { ConnectionDescriptor descriptor = connectionDescriptors.get(clientID); if (descriptor == null) { - LOG.error( - "The connection descriptor doesn't exist. " - + "The MQTT connection cannot be closed. CId=<{}>, closeImmediately = {}.", - clientID, - closeImmediately); + LOG.error("Connection descriptor doesn't exist. MQTT connection cannot be closed. CId=<{}>, " + + "closeImmediately={}", clientID, closeImmediately); return false; } if (closeImmediately) { @@ -123,7 +114,7 @@ public class ConnectionDescriptorStore implements IConnectionsManager { LOG.info("Retrieving status of session. CId=<{}>", clientID); ClientSession session = sessionsStore.sessionForClient(clientID); if (session == null) { - LOG.error("The given MQTT client ID doesn't have an associated session. CId=<{}>", clientID); + LOG.error("MQTT client ID doesn't have an associated session. CId=<{}>", clientID); return null; } return buildMqttSession(session); @@ -143,12 +134,8 @@ public class ConnectionDescriptorStore implements IConnectionsManager { MqttSession result = new MqttSession(); Collection mqttSubscriptions = new ArrayList<>(); for (Subscription subscription : session.getSubscriptions()) { - mqttSubscriptions.add( - new MqttSubscription( - subscription.getRequestedQos().toString(), - subscription.getClientId(), - subscription.getTopicFilter().toString(), - subscription.isActive())); + mqttSubscriptions.add(new MqttSubscription(subscription.getRequestedQos().toString(), + subscription.getClientId(), subscription.getTopicFilter().toString(), subscription.isActive())); } result.setActiveSubscriptions(mqttSubscriptions); result.setCleanSession(session.isCleanSession()); @@ -157,19 +144,14 @@ public class ConnectionDescriptorStore implements IConnectionsManager { result.setConnectionEstablished(true); BytesMetrics bytesMetrics = descriptor.getBytesMetrics(); MessageMetrics messageMetrics = descriptor.getMessageMetrics(); - result.setConnectionMetrics( - new MqttConnectionMetrics( - bytesMetrics.readBytes(), - bytesMetrics.wroteBytes(), - messageMetrics.messagesRead(), - messageMetrics.messagesWrote())); + result.setConnectionMetrics(new MqttConnectionMetrics(bytesMetrics.readBytes(), bytesMetrics.wroteBytes(), + messageMetrics.messagesRead(), messageMetrics.messagesWrote())); } else { result.setConnectionEstablished(false); } result.setPendingPublishMessagesNo(session.getPendingPublishMessagesNo()); result.setSecondPhaseAckPendingMessages(session.getSecondPhaseAckPendingMessages()); result.setInflightMessages(session.getInflightMessagesNo()); - LOG.info("The status of the session has been retrieved successfully. CId=<{}>", session.clientID); return result; } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 17920991..5dce61dd 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -95,7 +95,7 @@ public class Server { */ public void startServer() throws IOException { File defaultConfigurationFile = defaultConfigFile(); - LOG.info("Starting Moquette server. Configuration file path = {}.", defaultConfigurationFile.getAbsolutePath()); + LOG.info("Starting Moquette server. Configuration file path={}", defaultConfigurationFile.getAbsolutePath()); IResourceLoader filesystemLoader = new FileResourceLoader(defaultConfigurationFile); final IConfig config = new ResourceLoaderConfig(filesystemLoader); startServer(config); @@ -115,7 +115,7 @@ public class Server { * in case of any IO Error. */ public void startServer(File configFile) throws IOException { - LOG.info("Starting Moquette server. Configuration file path = {}.", configFile.getAbsolutePath()); + LOG.info("Starting Moquette server. Configuration file path={}", configFile.getAbsolutePath()); IResourceLoader filesystemLoader = new FileResourceLoader(configFile); final IConfig config = new ResourceLoaderConfig(filesystemLoader); startServer(config); @@ -136,7 +136,7 @@ public class Server { * in case of any IO Error. */ public void startServer(Properties configProps) throws IOException { - LOG.info("Starting Moquette server using properties object..."); + LOG.info("Starting Moquette server using properties object"); final IConfig config = new MemoryConfig(configProps); startServer(config); } @@ -166,7 +166,7 @@ public class Server { * in case of any IO Error. */ public void startServer(IConfig config, List handlers) throws IOException { - LOG.info("Starting moquette server using IConfig instance and intercept handlers..."); + LOG.info("Starting moquette server using IConfig instance and intercept handlers"); startServer(config, handlers, null, null, null); } @@ -175,7 +175,7 @@ public class Server { if (handlers == null) { handlers = Collections.emptyList(); } - LOG.info("Starting Moquette Server. MQTT message interceptors = {}.", getInterceptorIds(handlers)); + LOG.info("Starting Moquette Server. MQTT message interceptors={}", getInterceptorIds(handlers)); scheduler = Executors.newScheduledThreadPool(1); @@ -184,30 +184,28 @@ public class Server { config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp); } configureCluster(config); - LOG.info( - "Configuring Using persistent store file. Path = {}.", - config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME)); + final String persistencePath = config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME); + LOG.info("Configuring Using persistent store file, path={}", persistencePath); m_processorBootstrapper = new ProtocolProcessorBootstrapper(); - LOG.info("Initializing MQTT protocol processor..."); - final ProtocolProcessor processor = m_processorBootstrapper - .init(config, handlers, authenticator, authorizator, this); - + final ProtocolProcessor processor = m_processorBootstrapper.init(config, handlers, authenticator, authorizator, + this); + LOG.info("Initialized MQTT protocol processor"); if (sslCtxCreator == null) { - LOG.warn("Using default SSL context creator..."); + LOG.warn("Using default SSL context creator"); sslCtxCreator = new DefaultMoquetteSslContextCreator(config); } - LOG.info("Binding server to the configured ports..."); + LOG.info("Binding server to the configured ports"); m_acceptor = new NettyAcceptor(); m_acceptor.initialize(processor, config, sslCtxCreator); m_processor = processor; - LOG.info("The Moquette server has been initialized successfully."); + LOG.info("Moquette server has been initialized successfully"); m_initialized = true; } private void configureCluster(IConfig config) throws FileNotFoundException { - LOG.info("Configuring embedded Hazelcast instance..."); + LOG.info("Configuring embedded Hazelcast instance"); String interceptHandlerClassname = config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME); if (interceptHandlerClassname == null || !HZ_INTERCEPT_HANDLER.equals(interceptHandlerClassname)) { LOG.info("There are no Hazelcast intercept handlers. The server won't start a Hazelcast instance."); @@ -219,17 +217,17 @@ public class Server { Config hzconfig = isHzConfigOnClasspath ? new ClasspathXmlConfig(hzConfigPath) : new FileSystemXmlConfig(hzConfigPath); - LOG.info("Starting Hazelcast instance. ConfigurationFile = {}.", hzconfig); + LOG.info("Starting Hazelcast instance. ConfigurationFile={}", hzconfig); hazelcastInstance = Hazelcast.newHazelcastInstance(hzconfig); } else { - LOG.info("Starting Hazelcast instance with default configuration."); + LOG.info("Starting Hazelcast instance with default configuration"); hazelcastInstance = Hazelcast.newHazelcastInstance(); } listenOnHazelCastMsg(); } private void listenOnHazelCastMsg() { - LOG.info("Subscribing to Hazelcast topic. TopicName = {}.", "moquette"); + LOG.info("Subscribing to Hazelcast topic. TopicName={}", "moquette"); HazelcastInstance hz = getHazelcastInstance(); ITopic topic = hz.getTopic("moquette"); topic.addMessageListener(new HazelcastListener(this)); @@ -253,34 +251,32 @@ public class Server { public void internalPublish(MqttPublishMessage msg, final String clientId) { final int messageID = msg.variableHeader().messageId(); if (!m_initialized) { - LOG.error( - "The server is not started. The message cannot be published. MqttClientId = {}, messageId = {}", - clientId, - messageID); + 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. MqttClientId = {}, messageId = {}", clientId, messageID); + LOG.debug("Publishing message. CId={}, messageId={}", clientId, messageID); m_processor.internalPublish(msg, clientId); } public void stopServer() { - LOG.info("Unbinding server from the configured ports..."); + LOG.info("Unbinding server from the configured ports"); m_acceptor.close(); - LOG.info("Stopping MQTT protocol processor..."); + LOG.trace("Stopping MQTT protocol processor"); m_processorBootstrapper.shutdown(); m_initialized = false; if (hazelcastInstance != null) { - LOG.info("Stopping embedded Hazelcast instance..."); + LOG.trace("Stopping embedded Hazelcast instance"); try { hazelcastInstance.shutdown(); } catch (HazelcastInstanceNotActiveException e) { - LOG.warn("The embedded Hazelcast instance is already shut down."); + LOG.warn("embedded Hazelcast instance is already shut down."); } } scheduler.shutdown(); - LOG.info("The Moquette server has been stopped."); + LOG.info("Moquette server has been stopped."); } /** @@ -304,12 +300,11 @@ public class Server { */ public void addInterceptHandler(InterceptHandler interceptHandler) { if (!m_initialized) { - LOG.error( - "The server is not started. The MQTT message interceptor cannot be added. InterceptorId = {}.", - interceptHandler.getID()); + 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()); + LOG.info("Adding MQTT message interceptor. InterceptorId={}", interceptHandler.getID()); m_processor.addInterceptHandler(interceptHandler); } @@ -321,12 +316,11 @@ public class Server { */ public void removeInterceptHandler(InterceptHandler interceptHandler) { if (!m_initialized) { - LOG.error( - "The server is not started. The MQTT message interceptor cannot be removed. InterceptorId = {}.", - interceptHandler.getID()); + 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()); + LOG.info("Removing MQTT message interceptor. InterceptorId={}", interceptHandler.getID()); m_processor.removeInterceptHandler(interceptHandler); } diff --git a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java index e6b2ceed..06d29d41 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -138,7 +138,7 @@ public class NettyAcceptor implements ServerAcceptor { } private void initFactory(String host, int port, String protocol, final PipelineInitializer pipeliner) { - LOG.info("Initializing server. Protocol = {}.", protocol); + LOG.info("Initializing server. Protocol={}", protocol); ServerBootstrap b = new ServerBootstrap(); b.group(m_bossGroup, m_workerGroup).channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer() { @@ -157,26 +157,24 @@ public class NettyAcceptor implements ServerAcceptor { .option(ChannelOption.TCP_NODELAY, nettyTcpNodelay) .childOption(ChannelOption.SO_KEEPALIVE, nettySoKeepalive); try { - LOG.info("Binding server. Protocol = {}, host = {}, port = {}.", host, port); + LOG.info("Binding server. host={}, port={}", host, port); // Bind and start to accept incoming connections. ChannelFuture f = b.bind(host, port); - LOG.info("The server has been binded. Protocol = {}, host = {}, port = {}", host, port); + LOG.info("Server has been bound. host={}, port={}", host, port); f.sync(); } catch (InterruptedException ex) { - LOG.error("An interruptedException was caught while initializing server. Protocol = {}.", protocol, ex); + LOG.error("An interruptedException was caught while initializing server. Protocol={}", protocol, ex); } } private void initializePlainTCPTransport(final NettyMQTTHandler handler, IConfig props) throws IOException { - LOG.info("Configuring TCP MQTT transport..."); + LOG.info("Configuring TCP MQTT transport"); final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler(); String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME); String tcpPortProp = props.getProperty(PORT_PROPERTY_NAME, DISABLED_PORT_BIND); if (DISABLED_PORT_BIND.equals(tcpPortProp)) { - LOG.info( - "The property {} has been setted to {}. TCP MQTT will be disabled.", - BrokerConstants.PORT_PROPERTY_NAME, - DISABLED_PORT_BIND); + LOG.info("Property {} has been set to {}. TCP MQTT will be disabled", BrokerConstants.PORT_PROPERTY_NAME, + DISABLED_PORT_BIND); return; } int port = Integer.parseInt(tcpPortProp); @@ -198,14 +196,12 @@ public class NettyAcceptor implements ServerAcceptor { } private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException { - LOG.info("Configuring Websocket MQTT transport..."); + LOG.info("Configuring Websocket MQTT transport"); String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); if (DISABLED_PORT_BIND.equals(webSocketPortProp)) { // Do nothing no WebSocket configured - LOG.info( - "The property {} has been setted to {}. Websocket MQTT will be disabled.", - BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, - DISABLED_PORT_BIND); + LOG.info("Property {} has been setted to {}. Websocket MQTT will be disabled", + BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); return; } int port = Integer.parseInt(webSocketPortProp); @@ -219,8 +215,7 @@ public class NettyAcceptor implements ServerAcceptor { void init(ChannelPipeline pipeline) { pipeline.addLast(new HttpServerCodec()); pipeline.addLast("aggregator", new HttpObjectAggregator(65536)); - pipeline.addLast( - "webSocketHandler", + pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST)); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder()); @@ -238,14 +233,12 @@ public class NettyAcceptor implements ServerAcceptor { private void initializeSSLTCPTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext) throws IOException { - LOG.info("Configuring SSL MQTT transport..."); + LOG.info("Configuring SSL MQTT transport"); String sslPortProp = props.getProperty(SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); if (DISABLED_PORT_BIND.equals(sslPortProp)) { // Do nothing no SSL configured - LOG.info( - "The property {} has been setted to {}. SSL MQTT will be disabled.", - BrokerConstants.SSL_PORT_PROPERTY_NAME, - DISABLED_PORT_BIND); + LOG.info("Property {} has been set to {}. SSL MQTT will be disabled", + BrokerConstants.SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); return; } @@ -276,14 +269,12 @@ public class NettyAcceptor implements ServerAcceptor { private void initializeWSSTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext) throws IOException { - LOG.info("Configuring secure websocket MQTT transport..."); + LOG.info("Configuring secure websocket MQTT transport"); String sslPortProp = props.getProperty(WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); if (DISABLED_PORT_BIND.equals(sslPortProp)) { // Do nothing no SSL configured - LOG.info( - "The property {} has been setted to {}. Secure websocket MQTT will be disabled.", - BrokerConstants.WSS_PORT_PROPERTY_NAME, - DISABLED_PORT_BIND); + LOG.info("Property {} has been set to {}. Secure websocket MQTT will be disabled", + BrokerConstants.WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND); return; } int sslPort = Integer.parseInt(sslPortProp); @@ -299,8 +290,7 @@ public class NettyAcceptor implements ServerAcceptor { pipeline.addLast("httpEncoder", new HttpResponseEncoder()); pipeline.addLast("httpDecoder", new HttpRequestDecoder()); pipeline.addLast("aggregator", new HttpObjectAggregator(65536)); - pipeline.addLast( - "webSocketHandler", + pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST)); pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder()); pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder()); @@ -319,7 +309,7 @@ public class NettyAcceptor implements ServerAcceptor { public void close() { LOG.info("Closing Netty acceptor..."); if (m_workerGroup == null || m_bossGroup == null) { - LOG.error("The Netty acceptor is not initialized."); + LOG.error("Netty acceptor is not initialized"); throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized"); } Future workerWaiter = m_workerGroup.shutdownGracefully(); @@ -349,17 +339,13 @@ public class NettyAcceptor implements ServerAcceptor { LOG.info("Collecting message metrics..."); MessageMetrics metrics = m_metricsCollector.computeMetrics(); - LOG.info( - "The metrics have been collected. Read messages = {}, written messages = {}.", - metrics.messagesRead(), - metrics.messagesWrote()); + LOG.info("Metrics have been collected. Read messages={}, written messages={}", metrics.messagesRead(), + metrics.messagesWrote()); LOG.info("Collecting bytes metrics..."); BytesMetrics bytesMetrics = m_bytesMetricsCollector.computeMetrics(); - LOG.info( - "The bytes metrics have been collected. Read bytes = {}, written bytes = {}.", - bytesMetrics.readBytes(), - bytesMetrics.wroteBytes()); + LOG.info("Bytes metrics have been collected. Read bytes={}, written bytes={}", bytesMetrics.readBytes(), + bytesMetrics.wroteBytes()); } private ChannelHandler createSslHandler(SSLContext sslContext, boolean needsClientAuth) { diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 663f6929..f817e7af 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -43,6 +43,39 @@ import java.util.concurrent.BlockingQueue; */ public class ClientSession { + class OutboundFlightZone { + + /** + * Save the binding messageID, clientID - guid + + * @param messageID the packet ID used in transmission + * @param guid the guid of the message being in flight. + */ + void waitingAck(int messageID, MessageGUID guid) { + if (LOG.isTraceEnabled()) { + LOG.trace("Adding to inflight {}, guid <{}>", messageID, guid); + } + m_sessionsStore.inFlight(ClientSession.this.clientID, messageID, guid); + } + + void acknowledged(int messageID) { + if (LOG.isTraceEnabled()) + LOG.trace("Acknowledging inflight, clientID <{}> messageID {}", ClientSession.this.clientID, messageID); + m_sessionsStore.inFlightAck(ClientSession.this.clientID, messageID); + } + } + + class InboundFlightZone { + + public IMessagesStore.StoredMessage lookup(int messageID) { + return m_sessionsStore.inboundInflight(clientID, messageID); + } + + public void waitingRel(int messageID, MessageGUID guid) { + m_sessionsStore.markAsInboundInflight(clientID, messageID, guid); + } + } + private final static Logger LOG = LoggerFactory.getLogger(ClientSession.class); public final String clientID; @@ -57,6 +90,8 @@ public class ClientSession { // private BlockingQueue m_queueToPublish = new // ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE); + private final OutboundFlightZone outboundFlightZone; + private final InboundFlightZone inboundFlightZone; public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore, boolean cleanSession) { @@ -64,6 +99,8 @@ public class ClientSession { this.messagesStore = messagesStore; this.m_sessionsStore = sessionsStore; this.cleanSession = cleanSession; + this.outboundFlightZone = new OutboundFlightZone(); + this.inboundFlightZone = new InboundFlightZone(); } /** @@ -84,17 +121,12 @@ public class ClientSession { } public boolean subscribe(Subscription newSubscription) { - LOG.info( - "Adding new subscription. ClientId={}, topics={}, qos={}", - newSubscription.getClientId(), - newSubscription.getTopicFilter(), - newSubscription.getRequestedQos()); + LOG.info("Adding new subscription. ClientId={}, topics={}, qos={}", newSubscription.getClientId(), + newSubscription.getTopicFilter(), newSubscription.getRequestedQos()); boolean validTopic = newSubscription.getTopicFilter().isValid(); if (!validTopic) { - LOG.warn( - "The topic filter is not valid. ClientId={}, topics={}", - newSubscription.getClientId(), - newSubscription.getTopicFilter()); + LOG.warn("The topic filter is not valid. ClientId={}, topics={}", newSubscription.getClientId(), + newSubscription.getTopicFilter()); // send SUBACK with 0x80 for this topic filter return false; } @@ -140,7 +172,10 @@ public class ClientSession { // remove also the messages stored of type QoS1/2 LOG.info("Removing stored messages with QoS 1 and 2. ClientId={}", this.clientID); - messagesStore.dropMessagesInSession(this.clientID); + messagesStore.dropInFlightMessagesInSession(m_sessionsStore.pendingAck(this.clientID)); + + //remove also the enqueued messages + this.m_sessionsStore.dropQueue(this.clientID); } public boolean isCleanSession() { @@ -157,15 +192,19 @@ public class ClientSession { } public void inFlightAcknowledged(int messageID) { - if (LOG.isTraceEnabled()) - LOG.trace("Acknowledging inflight, clientID <{}> messageID {}", this.clientID, messageID); - m_sessionsStore.inFlightAck(this.clientID, messageID); + outboundFlightZone.acknowledged(messageID); } - public void inFlightAckWaiting(MessageGUID guid, int messageID) { - if (LOG.isTraceEnabled()) - LOG.trace("Adding to inflight {}, guid <{}>", messageID, guid); - m_sessionsStore.inFlight(this.clientID, messageID, guid); + /** + * Mark the message identified by guid as publish in flight. + * + * @return the packetID for the message in flight. + * */ + public int inFlightAckWaiting(MessageGUID guid) { + LOG.debug("Adding message ot inflight zone. ClientId={}, guid={}", clientID, guid); + int messageId = ClientSession.this.nextPacketId(); + outboundFlightZone.waitingAck(messageId, guid); + return messageId; } public IMessagesStore.StoredMessage secondPhaseAcknowledged(int messageID) { @@ -183,9 +222,12 @@ public class ClientSession { this.m_sessionsStore.queue(this.clientID).add(message); } - public IMessagesStore.StoredMessage storedMessage(int messageID) { - final MessageGUID guid = messagesStore.mapToGuid(clientID, messageID); - return messagesStore.getMessageByGuid(guid); + public IMessagesStore.StoredMessage inboundInflight(int messageID) { + return inboundFlightZone.lookup(messageID); + } + + public void markAsInboundInflight(int messageID, MessageGUID guid) { + inboundFlightZone.waitingRel(messageID, guid); } public void moveInFlightToSecondPhaseAckWaiting(int messageID) { diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 202beaec..9b853a50 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -37,8 +37,6 @@ public interface IMessagesStore { final String m_topic; private boolean m_retained; private String m_clientID; - // Optional attribute, available only fo QoS 1 and 2 - private Integer m_msgID; private MessageGUID m_guid; public StoredMessage(byte[] message, MqttQoS qos, String topic) { @@ -75,14 +73,6 @@ public interface IMessagesStore { this.m_clientID = m_clientID; } - public void setMessageID(Integer messageID) { - this.m_msgID = messageID; - } - - public Integer getMessageID() { - return m_msgID; - } - public ByteBuf getMessage() { return Unpooled.copiedBuffer(m_payload); } @@ -97,7 +87,7 @@ public interface IMessagesStore { @Override public String toString() { - return "PublishEvent{" + "m_msgID=" + m_msgID + ", clientID='" + m_clientID + '\'' + ", m_retain=" + return "PublishEvent{clientID='" + m_clientID + '\'' + ", m_retain=" + m_retained + ", m_qos=" + m_qos + ", m_topic='" + m_topic + '\'' + '}'; } } @@ -135,13 +125,9 @@ public interface IMessagesStore { */ MessageGUID storePublishForFuture(StoredMessage storedMessage); - void dropMessagesInSession(String clientID); + void dropInFlightMessagesInSession(Collection pendingAckMessages); StoredMessage getMessageByGuid(MessageGUID guid); void cleanRetained(Topic topic); - - int getPendingPublishMessages(String clientID); - - MessageGUID mapToGuid(String clientID, int messageID); } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 6d795e9b..008daf28 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -16,13 +16,14 @@ package io.moquette.spi; -import java.util.Collection; -import java.util.List; -import java.util.concurrent.BlockingQueue; import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.BlockingQueue; + /** * Store used to handle the persistence of the subscriptions tree. * @@ -188,24 +189,33 @@ public interface ISessionsStore { /** * Returns the number of inflight messages for the given client ID * - * @param clientID - * @return + * @param clientID target client. + * @return count of pending in flight publish messages. */ int getInflightMessagesNo(String clientID); + /** + * @return the inflight inbound (PUBREL for Qos2) message. + * */ + IMessagesStore.StoredMessage inboundInflight(String clientID, int messageID); + + void markAsInboundInflight(String clientID, int messageID, MessageGUID guid); + /** * Returns the size of the session queue for the given client ID * - * @param clientID - * @return + * @param clientID target client. + * @return count of enqueued publish messages. */ int getPendingPublishMessagesNo(String clientID); /** * Returns the number of second-phase ACK pending messages for the given client ID * - * @param clientID - * @return + * @param clientID target client. + * @return count of pending in flight publish messages. */ int getSecondPhaseAckPendingMessages(String clientID); + + Collection pendingAck(String clientID); } 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 ac4f5ec8..e2a3d71e 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -47,7 +47,7 @@ final class BrokerInterceptor implements Interceptor { private final ExecutorService executor; private BrokerInterceptor(int poolSize, List handlers) { - LOG.info("Initializing broker interceptor. InterceptorIds = {}.", getInterceptorIds(handlers)); + LOG.info("Initializing broker interceptor. InterceptorIds={}", getInterceptorIds(handlers)); this.handlers = new HashMap<>(); for (Class messageType : InterceptHandler.ALL_MESSAGE_TYPES) { this.handlers.put(messageType, new CopyOnWriteArrayList()); @@ -94,10 +94,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientConnected(final MqttConnectMessage msg) { for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) { - LOG.debug( - "Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.", - msg.payload().clientIdentifier(), - handler.getID()); + LOG.debug("Sending MQTT CONNECT message to interceptor. CId={}, interceptorId={}", + msg.payload().clientIdentifier(), handler.getID()); executor.execute(new Runnable() { @Override @@ -111,12 +109,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientDisconnected(final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) { - LOG.debug( - "Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, " - + "interceptorId = {}.", - clientID, - username, - handler.getID()); + LOG.debug("Notifying MQTT client disconnection to interceptor. CId={}, username={}, interceptorId={}", + clientID, username, handler.getID()); executor.execute(new Runnable() { @Override @@ -130,12 +124,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientConnectionLost(final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) { - LOG.debug( - "Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, " - + "username = {}, interceptorId = {}.", - clientID, - username, - handler.getID()); + LOG.debug("Notifying unexpected MQTT client disconnection to interceptor CId={}, username={}, " + + "interceptorId={}", clientID, username, handler.getID()); executor.execute(new Runnable() { @Override @@ -151,13 +141,8 @@ final class BrokerInterceptor implements Interceptor { int messageId = msg.variableHeader().messageId(); String topic = msg.variableHeader().topicName(); for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) { - LOG.debug( - "Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " - + "interceptorId = {}.", - clientID, - messageId, - topic, - handler.getID()); + LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, interceptorId={}", + clientID, messageId, topic, handler.getID()); executor.execute(new Runnable() { @Override @@ -171,12 +156,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyTopicSubscribed(final Subscription sub, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) { - LOG.debug( - "Notifying MQTT SUBSCRIBE message to interceptor. MqttClientId = {}, topicFilter = {}, " - + "interceptorId = {}.", - sub.getClientId(), - sub.getTopicFilter(), - handler.getID()); + LOG.debug("Notifying MQTT SUBSCRIBE message to interceptor. CId={}, topicFilter={}, interceptorId={}", + sub.getClientId(), sub.getTopicFilter(), handler.getID()); executor.execute(new Runnable() { @Override @@ -190,12 +171,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyTopicUnsubscribed(final String topic, final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptUnsubscribeMessage.class)) { - LOG.debug( - "Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, " - + "interceptorId = {}.", - clientID, - topic, - handler.getID()); + LOG.debug("Notifying MQTT UNSUBSCRIBE message to interceptor. CId={}, topic={}, interceptorId={}", clientID, + topic, handler.getID()); executor.execute(new Runnable() { @Override @@ -209,13 +186,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg) { for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) { - LOG.debug( - "Notifying MQTT ACK message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " - + "interceptorId = {}.", - msg.getMsg().getClientID(), - msg.getMsg().getMessageID(), - msg.getTopic(), - handler.getID()); + LOG.debug("Notifying MQTT ACK message to interceptor. CId={}, messageId={}, topic={}, interceptorId={}", + msg.getMsg().getClientID(), msg.getPacketID(), msg.getTopic(), handler.getID()); executor.execute(new Runnable() { @Override @@ -229,10 +201,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void addInterceptHandler(InterceptHandler interceptHandler) { Class[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler); - LOG.info( - "Adding MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.", - interceptHandler.getID(), - interceptedMessageTypes); + LOG.info( "Adding MQTT message interceptor. InterceptorId={}, handledMessageTypes={}", + interceptHandler.getID(), interceptedMessageTypes); for (Class interceptMessageType : interceptedMessageTypes) { this.handlers.get(interceptMessageType).add(interceptHandler); } @@ -241,10 +211,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void removeInterceptHandler(InterceptHandler interceptHandler) { Class[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler); - LOG.info( - "Removing MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.", - interceptHandler.getID(), - interceptedMessageTypes); + LOG.info("Removing MQTT message interceptor. InterceptorId={}, handledMessageTypes={}", + interceptHandler.getID(), interceptedMessageTypes); for (Class interceptMessageType : interceptedMessageTypes) { this.handlers.get(interceptMessageType).remove(interceptHandler); } diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java index 5206c5ba..1f48cefe 100644 --- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -25,17 +25,16 @@ class InternalRepublisher { void publishRetained(ClientSession targetSession, Collection messages) { for (IMessagesStore.StoredMessage storedMsg : messages) { // fire as retained the message - Integer packetID = storedMsg.getQos() == MqttQoS.AT_MOST_ONCE ? null : targetSession.nextPacketId(); - if (packetID != null) { - LOG.debug("Adding message to inflight zone. ClientId={}, packetId={}, messageId={}, topic={}", - targetSession.clientID, packetID, storedMsg.getMessageID(), storedMsg.getTopic()); - targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID); - } MqttPublishMessage publishMsg = retainedPublish(storedMsg); - // set the PacketIdentifier only for QoS > 0 - if (publishMsg.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) { + if (storedMsg.getQos() != MqttQoS.AT_MOST_ONCE) { + LOG.debug("Adding message to inflight zone. ClientId={}, topic={}", targetSession.clientID, + storedMsg.getTopic()); + int packetID = targetSession.inFlightAckWaiting(storedMsg.getGuid()); + + // set the PacketIdentifier only for QoS > 0 publishMsg = retainedPublish(storedMsg, packetID); } + this.messageSender.sendPublish(targetSession, publishMsg); } } @@ -46,13 +45,13 @@ class InternalRepublisher { for (IMessagesStore.StoredMessage pubEvt : storedPublishes) { // put in flight zone - LOG.debug("Adding message ot inflight zone. ClientId={}, guid={}, messageId={}, topic={}", - clientSession.clientID, pubEvt.getGuid(), pubEvt.getMessageID(), pubEvt.getTopic()); - clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID()); + LOG.debug("Adding message ot inflight zone. ClientId={}, guid={}, topic={}", clientSession.clientID, + pubEvt.getGuid(), pubEvt.getTopic()); + int messageId = clientSession.inFlightAckWaiting(pubEvt.getGuid()); MqttPublishMessage publishMsg = notRetainedPublish(pubEvt); // set the PacketIdentifier only for QoS > 0 if (publishMsg.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) { - publishMsg = notRetainedPublish(pubEvt, pubEvt.getMessageID()); + publishMsg = notRetainedPublish(pubEvt, messageId); } this.messageSender.sendPublish(clientSession, publishMsg); } diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index e03a3e9d..486d4508 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -63,25 +63,20 @@ class MessagesPublisher { // refCnt = 0 ByteBuf message = origMessage.retainedDuplicate(); if (targetIsActive) { - LOG.debug( - "Sending PUBLISH message to active subscriber. MessageId={}, CId={}, topicFilter={}, qos={}", - pubMsg.getMessageID(), - sub.getClientId(), - sub.getTopicFilter(), - qos); + LOG.debug("Sending PUBLISH message to active subscriber. CId={}, topicFilter={}, qos={}", + sub.getClientId(), sub.getTopicFilter(), qos); MqttPublishMessage publishMsg = notRetainedPublish(topic, qos, message); if (qos != MqttQoS.AT_MOST_ONCE) { // QoS 1 or 2 - int messageId = targetSession.nextPacketId(); - targetSession.inFlightAckWaiting(guid, messageId); + int messageId = targetSession.inFlightAckWaiting(guid); // set the PacketIdentifier only for QoS > 0 publishMsg = notRetainedPublishWithMessageId(topic, qos, message, messageId); } this.messageSender.sendPublish(targetSession, publishMsg); } else { if (!targetSession.isCleanSession()) { - LOG.debug("Storing pending PUBLISH inactive message. MessageId={}, CId={}, topicFilter={}, qos={}", - pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos); + LOG.debug("Storing pending PUBLISH inactive message. CId={}, topicFilter={}, qos={}", + sub.getClientId(), sub.getTopicFilter(), qos); // store the message in targetSession queue to deliver targetSession.enqueue(pubMsg); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index c5b52510..b30e159e 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -474,7 +474,7 @@ public class ProtocolProcessor { targetSession.inFlightAcknowledged(messageID); String topic = inflightMsg.getTopic(); - m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username)); + m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID)); } public static IMessagesStore.StoredMessage asStoredMessage(MqttPublishMessage msg) { @@ -485,7 +485,6 @@ public class ProtocolProcessor { IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(payloadContent, msg.fixedHeader().qosLevel(), msg.variableHeader().topicName()); stored.setRetained(msg.fixedHeader().isRetain()); - stored.setMessageID(msg.variableHeader().messageId()); return stored; } @@ -540,7 +539,6 @@ public class ProtocolProcessor { } else { toStoreMsg.setClientID(clientId); } - toStoreMsg.setMessageID(1); if (qos == EXACTLY_ONCE) { // QoS2 guid = m_messagesStore.storePublishForFuture(toStoreMsg); } @@ -568,18 +566,12 @@ public class ProtocolProcessor { private void forwardPublishWill(WillMessage will, String clientID) { // it has just to publish the message downstream to the subscribers // NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1 - Integer messageId = null; - if (will.getQos() != AT_MOST_ONCE) { - messageId = m_sessionsStore.nextPacketID(clientID); - } - IMessagesStore.StoredMessage tobeStored = asStoredMessage(will); tobeStored.setClientID(clientID); - tobeStored.setMessageID(messageId); Topic topic = new Topic(tobeStored.getTopic()); List topicMatchingSubscriptions = subscriptions.matches(topic); - LOG.info("Publishing will message. CId={}, messageId={}, topic={}", clientID, messageId, topic); + LOG.info("Publishing will message. CId={}, topic={}", clientID, topic); this.messagesPublisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions); } @@ -626,7 +618,7 @@ public class ProtocolProcessor { StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID); String username = NettyUtils.userName(channel); String topic = inflightMsg.getTopic(); - m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username)); + m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID)); } public void processDisconnect(Channel channel) throws InterruptedException { @@ -759,8 +751,7 @@ public class ProtocolProcessor { if (!validTopic) { // close the connection, not valid topicFilter is a protocol violation channel.close(); - LOG.error( "Topic filter is not valid. CId={}, topics={}, badTopicFilter={}", clientID, topics, - topic); + LOG.error("Topic filter is not valid. CId={}, topics={}, badTopicFilter={}", clientID, topics, topic); return; } diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java index 063f66f2..4464447e 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -16,6 +16,8 @@ import io.netty.handler.codec.mqtt.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; + +import static io.moquette.spi.impl.DebugUtils.payload2Str; import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; import static io.moquette.spi.impl.Utils.messageId; import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; @@ -23,7 +25,7 @@ import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; class Qos2PublishHandler extends QosPublishHandler { - private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class); + private static final Logger LOG = LoggerFactory.getLogger(Qos2PublishHandler.class); private final SubscriptionsStore subscriptions; private final IMessagesStore m_messagesStore; @@ -57,18 +59,16 @@ class Qos2PublishHandler extends QosPublishHandler { String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); + LOG.info("Sending publish message to subscribers CId={}, topic={}, messageId={}", clientID, topic, messageID); if (LOG.isTraceEnabled()) { - LOG.trace( "Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " + - "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()), - subscriptions.dumpTree()); - } else { - LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic, - messageID); + LOG.trace("payload={}, subs Tree={}", payload2Str(toStoreMsg.getMessage()), subscriptions.dumpTree()); } // QoS2 MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); - // TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) { + ClientSession sourceSession = m_sessionsStore.sessionForClient(clientID); + sourceSession.markAsInboundInflight(messageID, guid); + sendPubRec(clientID, messageID); // Next the client will send us a pub rel @@ -92,18 +92,22 @@ class Qos2PublishHandler extends QosPublishHandler { void processPubRel(Channel channel, MqttMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); - LOG.info("Processing PUBREL message. ClientId={}, messageId={}", clientID, messageID); + LOG.info("Processing PUBREL message. CId={}, messageId={}", clientID, messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID); + IMessagesStore.StoredMessage evt = targetSession.inboundInflight(messageID); + if (evt == null) { + LOG.warn("Can't find inbound inflight message for CId={}, messageId={}", clientID, messageID); + throw new IllegalArgumentException("Can't find inbound inflight message"); + } final Topic topic = new Topic(evt.getTopic()); List topicMatchingSubscriptions = subscriptions.matches(topic); if (LOG.isDebugEnabled()) { - LOG.debug("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " + - "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(evt.getMessage()), + LOG.debug("Sending publish message to subscribers. CId={}, topic={}, messageId={}, payload={}, " + + "subscriptionTree={}", clientID, topic, messageID, payload2Str(evt.getMessage()), subscriptions.dumpTree()); } else { - LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic, + LOG.info("Sending publish message to subscribers. CId={}, topic={}, messageId={}", clientID, topic, messageID); } this.publisher.publish2Subscribers(evt, topicMatchingSubscriptions); @@ -120,14 +124,14 @@ class Qos2PublishHandler extends QosPublishHandler { } private void sendPubRec(String clientID, int messageID) { - LOG.debug("Sending PUBREC message. ClientId={}, messageId={}", clientID, messageID); + LOG.debug("Sending PUBREC message. CId={}, messageId={}", clientID, messageID); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, AT_MOST_ONCE, false, 0); MqttMessage pubRecMessage = new MqttMessage(fixedHeader, from(messageID)); connectionDescriptors.sendMessage(pubRecMessage, messageID, clientID); } private void sendPubComp(String clientID, int messageID) { - LOG.debug("Sending PUBCOMP message. ClientId={}, messageId={}", clientID, messageID); + LOG.debug("Sending PUBCOMP message. CId={}, messageId={}", clientID, messageID); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, AT_MOST_ONCE, false, 0); MqttMessage pubCompMessage = new MqttMessage(fixedHeader, from(messageID)); connectionDescriptors.sendMessage(pubCompMessage, messageID, clientID); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java index 9f8efbb7..f250b8f2 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationHazelcastHandlerInterceptorTest.java @@ -9,10 +9,7 @@ import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttClientPersistence; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; -import org.junit.After; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.Test; +import org.junit.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -114,14 +111,22 @@ public class ServerIntegrationHazelcastHandlerInterceptorTest { MqttMessage messageQos0 = m_messagesCollector.waitMessage(1); assertEquals("Hello world MQTT QoS0", messageQos0.toString()); assertEquals(0, messageQos0.getQos()); + } + @Test + public void checkPublishPassThroughCluster_Qos1() throws Exception { + LOG.info("*** checkPublishPassThroughCluster_Qos1 ***"); m_listener.subscribe("/topic", 1); m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(), 1, false); MqttMessage messageQos1 = m_messagesCollector.waitMessage(1); assertEquals("Hello world MQTT QoS1", messageQos1.toString()); assertEquals(1, messageQos1.getQos()); + } + @Test + public void checkPublishPassThroughCluster_Qos2() throws Exception { + LOG.info("*** checkPublishPassThroughCluster_Qos2 ***"); m_listener.subscribe("/topic", 2); m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 50e42f37..d64a2e62 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -364,7 +364,6 @@ public class ProtocolProcessorTest { StoredMessage retainedMessage = new StoredMessage("Hello".getBytes(), MqttQoS.EXACTLY_ONCE, "/topic"); retainedMessage.setRetained(true); - retainedMessage.setMessageID(120); retainedMessage.setClientID(FAKE_PUBLISHER_ID); m_messagesStore.storePublishForFuture(retainedMessage); diff --git a/broker/src/test/resources/log4j.properties b/broker/src/test/resources/log4j.properties index e923dd42..69c99512 100644 --- a/broker/src/test/resources/log4j.properties +++ b/broker/src/test/resources/log4j.properties @@ -2,20 +2,14 @@ log4j.rootLogger=ERROR, stdout, file log4j.logger.io.moquette=WARN -#log4j.logger.io.moquette=DEBUG -#log4j.logger.io.moquette.spi.impl.ProtocolProcessor=TRACE log4j.logger.io.moquette.spi.impl.ProtocolProcessor=WARN -#log4j.logger.io.moquette.spi.impl.SimpleMessaging=DEBUG -#Protocol parsing -#log4j.logger.io.moquette.server.netty.NettyMQTTHandler=DEBUG +#Protocol dispatching log4j.logger.io.moquette.server.netty.NettyMQTTHandler=WARN -#log4j.logger.io.moquette.parser.netty=DEBUG #Storage service #log4j.logger.io.moquette.spi.impl.subscriptions.SubscriptionsStore=DEBUG -#log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=DEBUG log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=WARN log4j.logger.io.moquette.spi.persistence.MapDBSessionsStore=WARN diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java index dad74093..b4ea2e90 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java @@ -48,14 +48,14 @@ class MapDBMessagesStore implements IMessagesStore { @Override public void initStore() { - LOG.info("Initializing store..."); m_retainedStore = m_db.getHashMap("retained"); m_persistentMessageStore = m_db.getHashMap("persistedMessages"); + LOG.info("Initialized store"); } @Override public void storeRetained(Topic topic, MessageGUID guid) { - LOG.debug("Storing retained messages. Topic = {}, guid = {}.", topic, guid); + LOG.debug("Storing retained messages. Topic={}, guid={}", topic, guid); m_retainedStore.put(topic, guid); } @@ -72,7 +72,7 @@ class MapDBMessagesStore implements IMessagesStore { } if (LOG.isTraceEnabled()) { - LOG.trace("The retained messages have been scanned. MatchingMessages = {}.", results); + LOG.trace("The retained messages have been scanned. MatchingMessages={}", results); } return results; @@ -80,75 +80,35 @@ class MapDBMessagesStore implements IMessagesStore { @Override public MessageGUID storePublishForFuture(StoredMessage storedMessage) { - assert storedMessage.getClientID() != null : "The message to be persisted must have a valid client ID"; + assert storedMessage.getClientID() != null : "Message to be persisted must have a valid client ID"; MessageGUID guid = new MessageGUID(UUID.randomUUID().toString()); storedMessage.setGuid(guid); - LOG.debug( - "Storing publish event. MqttClientId = {}, messageId = {}, guid = {}, topic = {}.", - storedMessage.getClientID(), - storedMessage.getMessageID(), - guid, - storedMessage.getTopic()); + LOG.debug("Storing publish event. CId={}, guid={}, topic={}", storedMessage.getClientID(), guid, + storedMessage.getTopic()); m_persistentMessageStore.put(guid, storedMessage); - ConcurrentMap messageIdToGuid = m_db - .getHashMap(messageId2GuidsMapName(storedMessage.getClientID())); - messageIdToGuid.put(storedMessage.getMessageID(), guid); return guid; } @Override - public void dropMessagesInSession(String clientID) { - LOG.debug("Dropping stored messages. ClientId = {}.", clientID); - ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); - for (MessageGUID guid : messageIdToGuid.values()) { - removeStoredMessage(guid); - } - messageIdToGuid.clear(); - } + public void dropInFlightMessagesInSession(Collection pendingAckMessages) { + //remove all guids from retained + Collection messagesToRemove = new HashSet<>(pendingAckMessages); + messagesToRemove.removeAll(m_retainedStore.values()); - void removeStoredMessage(MessageGUID guid) { - // remove only the not retained and no more referenced - StoredMessage storedMessage = m_persistentMessageStore.get(guid); - if (!storedMessage.isRetained()) { - LOG.debug( - "Dropping stored message. ClientId = {}, messageId = {}, guid = {}, topic = {}.", - storedMessage.getClientID(), - storedMessage.getMessageID(), - guid, - storedMessage.getTopic()); + for (MessageGUID guid : messagesToRemove) { m_persistentMessageStore.remove(guid); } } @Override public StoredMessage getMessageByGuid(MessageGUID guid) { - LOG.debug("Retrieving stored message. Guid = {}.", guid); + LOG.debug("Retrieving stored message. Guid={}", guid); return m_persistentMessageStore.get(guid); } @Override public void cleanRetained(Topic topic) { - LOG.debug("Cleaning retained messages. Topic = {}.", topic); + LOG.debug("Cleaning retained messages. Topic={}", topic); m_retainedStore.remove(topic); } - - @Override - public int getPendingPublishMessages(String clientID) { - ConcurrentMap messageIdToGuidMap = m_db - .getHashMap(messageId2GuidsMapName(clientID)); - return messageIdToGuidMap.size(); - } - - @Override - public MessageGUID mapToGuid(String clientID, int messageID) { - LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID); - ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); - MessageGUID result = messageIdToGuid.get(messageID); - LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result); - return result; - } - - static String messageId2GuidsMapName(String clientID) { - return "guidsMapping_" + clientID; - } } diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java index 70fe89b8..7e8f9e2b 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java @@ -28,6 +28,7 @@ import io.moquette.spi.impl.subscriptions.Topic; import org.mapdb.DB; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import java.util.*; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; @@ -310,9 +311,24 @@ class MapDBSessionsStore implements ISessionsStore { return m_inflightStore.get(clientID).size(); } + @Override + public StoredMessage inboundInflight(String clientID, int messageID) { + LOG.debug("Mapping inbound message ID to GUID CId={}, messageId={}", clientID, messageID); + ConcurrentMap messageIdToGuid = m_db.getHashMap(inboundMessageId2GuidsMapName(clientID)); + final MessageGUID guid = messageIdToGuid.get(messageID); + LOG.debug("Inbound message ID has been mapped CId={}, messageId={}, guid={}", clientID, messageID, guid); + return m_messagesStore.getMessageByGuid(guid); + } + + @Override + public void markAsInboundInflight(String clientID, int messageID, MessageGUID guid) { + ConcurrentMap messageIdToGuid = m_db.getHashMap(inboundMessageId2GuidsMapName(clientID)); + messageIdToGuid.put(messageID, guid); + } + @Override public int getPendingPublishMessagesNo(String clientID) { - return m_messagesStore.getPendingPublishMessages(clientID); + return queue(clientID).size(); } @Override @@ -322,4 +338,22 @@ class MapDBSessionsStore implements ISessionsStore { else return m_secondPhaseStore.get(clientID).size(); } + + @Override + public Collection pendingAck(String clientID) { + ConcurrentMap messageGUIDMap = m_db.getHashMap(messageId2GuidsMapName(clientID)); + if (messageGUIDMap == null || messageGUIDMap.isEmpty()) { + return Collections.emptyList(); + } + + return new ArrayList<>(messageGUIDMap.values()); + } + + static String messageId2GuidsMapName(String clientID) { + return "guidsMapping_" + clientID; + } + + static String inboundMessageId2GuidsMapName(String clientID) { + return "inboundInflightGuidsMapping_" + clientID; + } } diff --git a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java index ab83c8a7..11b6338a 100644 --- a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java +++ b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java @@ -16,9 +16,6 @@ package io.moquette.persistence.mapdb; -import java.io.File; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; import io.moquette.BrokerConstants; import io.moquette.server.config.IConfig; import io.moquette.server.config.MemoryConfig; @@ -29,15 +26,19 @@ import io.moquette.spi.ISessionsStore.ClientTopicCouple; import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; -import io.netty.handler.codec.mqtt.MqttQoS; +import static io.netty.handler.codec.mqtt.MqttQoS.*; import org.junit.After; import org.junit.Before; import org.junit.Test; + +import java.io.File; import java.util.List; import java.util.Properties; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME; +import static java.util.Collections.singletonList; import static org.junit.Assert.*; /** @@ -92,14 +93,11 @@ public class MapDBPersistentStoreTest { ClientSession session1 = m_sessionsStore.createNewSession("SESSION_ID_1", true); // Subscribe on /topic with QOSType.MOST_ONE - Subscription oldSubscription = new Subscription(session1.clientID, new Topic("/topic"), MqttQoS.AT_MOST_ONCE); + Subscription oldSubscription = new Subscription(session1.clientID, new Topic("/topic"), AT_MOST_ONCE); session1.subscribe(oldSubscription); // Subscribe on /topic again that overrides the previous subscription. - Subscription overridingSubscription = new Subscription( - session1.clientID, - new Topic("/topic"), - MqttQoS.EXACTLY_ONCE); + Subscription overridingSubscription = new Subscription(session1.clientID, new Topic("/topic"), EXACTLY_ONCE); session1.subscribe(overridingSubscription); // Verify @@ -156,18 +154,15 @@ public class MapDBPersistentStoreTest { @Test public void testDropMessagesInSessionCleanAllNotRetainedStoredMessages() { - m_sessionsStore.createNewSession("TestClient", true); - IMessagesStore.StoredMessage publishToStore = new IMessagesStore.StoredMessage( - "Hello".getBytes(), - MqttQoS.EXACTLY_ONCE, - "/topic"); + m_sessionsStore.createNewSession(TEST_CLIENT, true); + IMessagesStore.StoredMessage publishToStore = new IMessagesStore.StoredMessage("Hello".getBytes(), EXACTLY_ONCE, + "/topic"); publishToStore.setClientID(TEST_CLIENT); - publishToStore.setMessageID(1); publishToStore.setRetained(false); MessageGUID guid = m_messagesStore.storePublishForFuture(publishToStore); // Exercise - m_messagesStore.dropMessagesInSession("TestClient"); + m_messagesStore.dropInFlightMessagesInSession(singletonList(guid)); // Verify the message store for session is empty. IMessagesStore.StoredMessage storedPublish = m_messagesStore.getMessageByGuid(guid); @@ -176,18 +171,16 @@ public class MapDBPersistentStoreTest { @Test public void testDropMessagesInSessionDoesntCleanAnyRetainedStoredMessages() { - m_sessionsStore.createNewSession("TestClient", true); - IMessagesStore.StoredMessage publishToStore = new IMessagesStore.StoredMessage( - "Hello".getBytes(), - MqttQoS.EXACTLY_ONCE, - "/topic"); + m_sessionsStore.createNewSession(TEST_CLIENT, true); + IMessagesStore.StoredMessage publishToStore = new IMessagesStore.StoredMessage("Hello".getBytes(), EXACTLY_ONCE, + "/topic"); publishToStore.setClientID(TEST_CLIENT); - publishToStore.setMessageID(1); publishToStore.setRetained(true); MessageGUID guid = m_messagesStore.storePublishForFuture(publishToStore); + m_messagesStore.storeRetained(new Topic("/topic"), guid); // Exercise - m_messagesStore.dropMessagesInSession("TestClient"); + m_messagesStore.dropInFlightMessagesInSession(singletonList(guid)); // Verify the message store for session is empty. IMessagesStore.StoredMessage storedPublish = m_messagesStore.getMessageByGuid(guid);