anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Introduced the FlightZone concept (inbound and outbound) and separated logic of storage system.

Browse Source
This commit is contained in:
andsel
2017-04-09 19:22:50 +02:00
parent 5c6889fbd5
commit 6dc762f47f
21 changed files with 356 additions and 396 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java
+35 -29
View File
@@ -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;
}
}
broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
+2 -27
View File
@@ -35,7 +35,6 @@ public class MemoryMessagesStore implements IMessagesStore {
private Map<Topic, MessageGUID> m_retainedStore = new HashMap<>();
private Map<MessageGUID, StoredMessage> m_persistentMessageStore = new HashMap<>();
private Map<String, Map<Integer, MessageGUID>> 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<Integer, MessageGUID> emptyGuids = new HashMap<>();
Map<Integer, MessageGUID> 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<Integer, MessageGUID> messageGUIDMap = m_messageToGuids.get(clientID);
if (messageGUIDMap == null || messageGUIDMap.isEmpty()) {
return;
}
public void dropInFlightMessagesInSession(Collection<MessageGUID> pendingAckMessages) {
//remove all guids from retained
Collection<MessageGUID> messagesToRemove = new HashSet<>(messageGUIDMap.values());
Collection<MessageGUID> 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<Integer, MessageGUID> messageToGuids = m_messageToGuids.get(clientID);
if (messageToGuids == null)
return 0;
else
return messageToGuids.size();
}
@Override
public MessageGUID mapToGuid(String clientID, int messageID) {
final HashMap<Integer, MessageGUID> emptyGuids = new HashMap<>();
Map<Integer, MessageGUID> guids = Utils.defaultGet(m_messageToGuids, clientID, emptyGuids);
return guids.get(messageID);
}
}
broker/src/main/java/io/moquette/persistence/MemorySessionStore.java
+37 -2
View File
@@ -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<String, Map<Integer, MessageGUID>> m_inflightStore = new HashMap<>();
// private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
// maps clientID->BlockingQueue
private Map<String, BlockingQueue<StoredMessage>> queues = new HashMap<>();
// maps clientID->[MessageId -> guid]
private Map<String, Map<Integer, MessageGUID>> m_secondPhaseStore = new HashMap<>();
private Map<String, Map<Integer, MessageGUID>> outboundFlightMessageToGuid = new HashMap<>();
private Map<String, Map<Integer, MessageGUID>> 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<Integer, MessageGUID> emptyGuids = new HashMap<>();
Map<Integer, MessageGUID> 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<Integer, MessageGUID> emptyGuids = new HashMap<>();
Map<Integer, MessageGUID> 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<Integer, MessageGUID> emptyGuids = new HashMap<>();
Map<Integer, MessageGUID> 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<MessageGUID> pendingAck(String clientID) {
Map<Integer, MessageGUID> messageGUIDMap = outboundFlightMessageToGuid.get(clientID);
if (messageGUIDMap == null || messageGUIDMap.isEmpty()) {
return Collections.emptyList();
}
return new ArrayList<>(messageGUIDMap.values());
}
}
broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java
+10 -28
View File
@@ -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<MqttSubscription> 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;
}
broker/src/main/java/io/moquette/server/Server.java
+31 -37
View File
@@ -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<? extends InterceptHandler> 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<HazelcastMsg> 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);
}
broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
+23 -37
View File
@@ -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<SocketChannel>() {
@@ -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) {
broker/src/main/java/io/moquette/spi/ClientSession.java
+62 -20
View File
@@ -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<AbstractMessage> 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) {
broker/src/main/java/io/moquette/spi/IMessagesStore.java
+2 -16
View File
@@ -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<MessageGUID> pendingAckMessages);
StoredMessage getMessageByGuid(MessageGUID guid);
void cleanRetained(Topic topic);
int getPendingPublishMessages(String clientID);
MessageGUID mapToGuid(String clientID, int messageID);
}
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+19 -9
View File
@@ -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<MessageGUID> pendingAck(String clientID);
}
broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java
+19 -51
View File
@@ -47,7 +47,7 @@ final class BrokerInterceptor implements Interceptor {
private final ExecutorService executor;
private BrokerInterceptor(int poolSize, List<InterceptHandler> 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<InterceptHandler>());
@@ -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);
}
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+11 -12
View File
@@ -25,17 +25,16 @@ class InternalRepublisher {
void publishRetained(ClientSession targetSession, Collection<IMessagesStore.StoredMessage> 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);
}
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+5 -10
View File
@@ -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);
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+4 -13
View File
@@ -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<Subscription> 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;
}
broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java
+19 -15
View File
@@ -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<Subscription> 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);