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

Switched from Moquette's MQTT codec to Netty's MQTT codec ones

Browse Source
This commit is contained in:
andsel
2017-01-28 15:59:44 +01:00
parent 07093ab0cb
commit 1c64186a00
53 changed files with 1571 additions and 1048 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/interception/HazelcastInterceptHandler.java
+8 -1
View File
@@ -4,9 +4,12 @@ import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ITopic;
import io.moquette.interception.messages.InterceptPublishMessage;
import io.moquette.server.Server;
import io.netty.buffer.ByteBuf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static io.moquette.spi.impl.Utils.readBytesAndRewind;
/**
* Created by mackristof on 28/05/2016.
*/
@@ -26,7 +29,11 @@ public class HazelcastInterceptHandler extends AbstractInterceptHandler {
@Override
public void onPublish(InterceptPublishMessage msg) {
LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), new String(msg.getPayload().array()));
//TODO ugly, too much array copy
ByteBuf payload = msg.getPayload();
byte[] payloadContent = readBytesAndRewind(payload);
LOG.info("{} publish on {} message: {}", msg.getClientID(), msg.getTopicName(), new String(payloadContent));
ITopic<HazelcastMsg> topic = hz.getTopic("moquette");
HazelcastMsg hazelcastMsg = new HazelcastMsg(msg);
topic.publish(hazelcastMsg);
broker/src/main/java/io/moquette/interception/HazelcastMsg.java
+6 -4
View File
@@ -4,6 +4,8 @@ import io.moquette.interception.messages.InterceptPublishMessage;
import java.io.Serializable;
import static io.moquette.spi.impl.Utils.readBytesAndRewind;
/**
* Created by mackristof on 28/05/2016.
*/
@@ -11,22 +13,22 @@ public class HazelcastMsg implements Serializable {
private static final long serialVersionUID = -1431584750134928273L;
private final String clientId;
private final byte qos;
private final int qos;
private final byte[] payload;
private final String topic;
public HazelcastMsg(InterceptPublishMessage msg) {
this.clientId = msg.getClientID();
this.topic = msg.getTopicName();
this.qos = msg.getQos().byteValue();
this.payload = msg.getPayload().array();
this.qos = msg.getQos().value();
this.payload = readBytesAndRewind(msg.getPayload());
}
public String getClientId() {
return clientId;
}
public byte getQos() {
public int getQos() {
return qos;
}
broker/src/main/java/io/moquette/interception/InterceptHandler.java
+3 -9
View File
@@ -15,22 +15,16 @@
*/
package io.moquette.interception;
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
import io.moquette.interception.messages.InterceptConnectMessage;
import io.moquette.interception.messages.InterceptConnectionLostMessage;
import io.moquette.interception.messages.InterceptDisconnectMessage;
import io.moquette.interception.messages.InterceptPublishMessage;
import io.moquette.interception.messages.InterceptSubscribeMessage;
import io.moquette.interception.messages.InterceptUnsubscribeMessage;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.interception.messages.*;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.netty.handler.codec.mqtt.MqttMessage;
/**
* This interface is used to inject code for intercepting broker events.
* <p>
* The events can act only as observers.
* <p>
* Almost every method receives a subclass of {@link AbstractMessage}, except
* Almost every method receives a subclass of {@link MqttMessage}, except
* <code>onDisconnect</code> that receives the client id string and
* <code>onSubscribe</code> and <code>onUnsubscribe</code> that receive a
* {@link Subscription} object.
broker/src/main/java/io/moquette/interception/Interceptor.java
+4 -4
View File
@@ -16,9 +16,9 @@
package io.moquette.interception;
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
import io.moquette.parser.proto.messages.ConnectMessage;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
/**
* This interface is to be used internally by the broker components.
@@ -35,13 +35,13 @@ import io.moquette.spi.impl.subscriptions.Subscription;
*/
public interface Interceptor {
void notifyClientConnected(ConnectMessage msg);
void notifyClientConnected(MqttConnectMessage msg);
void notifyClientDisconnected(String clientID, String username);
void notifyClientConnectionLost(String clientID, String username);
void notifyTopicPublished(PublishMessage msg, String clientID, final String username);
void notifyTopicPublished(MqttPublishMessage msg, String clientID, final String username);
void notifyTopicSubscribed(Subscription sub, final String username);
broker/src/main/java/io/moquette/interception/messages/InterceptAbstractMessage.java
+8 -7
View File
@@ -15,27 +15,28 @@
*/
package io.moquette.interception.messages;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
/**
* @author Wagner Macedo
*/
public abstract class InterceptAbstractMessage implements InterceptMessage {
private final AbstractMessage msg;
private final MqttMessage msg;
InterceptAbstractMessage(AbstractMessage msg) {
InterceptAbstractMessage(MqttMessage msg) {
this.msg = msg;
}
public boolean isRetainFlag() {
return msg.isRetainFlag();
return msg.fixedHeader().isRetain();
}
public boolean isDupFlag() {
return msg.isDupFlag();
return msg.fixedHeader().isDup();
}
public AbstractMessage.QOSType getQos() {
return msg.getQos();
public MqttQoS getQos() {
return msg.fixedHeader().qosLevel();
}
}
broker/src/main/java/io/moquette/interception/messages/InterceptConnectMessage.java
+17 -17
View File
@@ -1,71 +1,71 @@
package io.moquette.interception.messages;
import io.moquette.parser.proto.messages.ConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
/**
* @author Wagner Macedo
*/
public class InterceptConnectMessage extends InterceptAbstractMessage {
private final ConnectMessage msg;
private final MqttConnectMessage msg;
public InterceptConnectMessage(ConnectMessage msg) {
public InterceptConnectMessage(MqttConnectMessage msg) {
super(msg);
this.msg = msg;
}
public String getClientID() {
return msg.getClientID();
return msg.payload().clientIdentifier();
}
public boolean isCleanSession() {
return msg.isCleanSession();
return msg.variableHeader().isCleanSession();
}
public int getKeepAlive() {
return msg.getKeepAlive();
return msg.variableHeader().keepAliveTimeSeconds();
}
public boolean isPasswordFlag() {
return msg.isPasswordFlag();
return msg.variableHeader().hasPassword();
}
public byte getProtocolVersion() {
return msg.getProtocolVersion();
return (byte) msg.variableHeader().version();
}
public String getProtocolName() {
return msg.getProtocolName();
return msg.variableHeader().name();
}
public boolean isUserFlag() {
return msg.isUserFlag();
return msg.variableHeader().hasUserName();
}
public boolean isWillFlag() {
return msg.isWillFlag();
return msg.variableHeader().isWillFlag();
}
public byte getWillQos() {
return msg.getWillQos();
return (byte) msg.variableHeader().willQos();
}
public boolean isWillRetain() {
return msg.isWillRetain();
return msg.variableHeader().isWillRetain();
}
public String getUsername() {
return msg.getUsername();
return msg.payload().userName();
}
public byte[] getPassword() {
return msg.getPassword();
return msg.payload().password().getBytes();
}
public String getWillTopic() {
return msg.getWillTopic();
return msg.payload().willTopic();
}
public byte[] getWillMessage() {
return msg.getWillMessage();
return msg.payload().willMessage().getBytes();
}
}
broker/src/main/java/io/moquette/interception/messages/InterceptPublishMessage.java
+7 -8
View File
@@ -1,18 +1,17 @@
package io.moquette.interception.messages;
import io.moquette.parser.proto.messages.PublishMessage;
import java.nio.ByteBuffer;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
/**
* @author Wagner Macedo
*/
public class InterceptPublishMessage extends InterceptAbstractMessage {
private final PublishMessage msg;
private final MqttPublishMessage msg;
private final String clientID;
private final String username;
public InterceptPublishMessage(PublishMessage msg, String clientID, String username) {
public InterceptPublishMessage(MqttPublishMessage msg, String clientID, String username) {
super(msg);
this.msg = msg;
this.clientID = clientID;
@@ -20,11 +19,11 @@ public class InterceptPublishMessage extends InterceptAbstractMessage {
}
public String getTopicName() {
return msg.getTopicName();
return msg.variableHeader().topicName();
}
public ByteBuffer getPayload() {
return msg.getPayload();
public ByteBuf getPayload() {
return msg.payload();
}
public String getClientID() {
broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java
+2 -2
View File
@@ -1,7 +1,7 @@
package io.moquette.interception.messages;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.netty.handler.codec.mqtt.MqttQoS;
/**
* @author Wagner Macedo
@@ -19,7 +19,7 @@ public class InterceptSubscribeMessage implements InterceptMessage {
return subscription.getClientId();
}
public AbstractMessage.QOSType getRequestedQos() {
public MqttQoS getRequestedQos() {
return subscription.getRequestedQos();
}
broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java
+25 -37
View File
@@ -1,23 +1,23 @@
package io.moquette.server;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.moquette.connections.IConnectionsManager;
import io.moquette.connections.MqttConnectionMetrics;
import io.moquette.connections.MqttSession;
import io.moquette.connections.MqttSubscription;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.server.netty.metrics.BytesMetrics;
import io.moquette.server.netty.metrics.MessageMetrics;
import io.moquette.spi.ClientSession;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class ConnectionDescriptorStore implements IConnectionsManager {
@@ -27,38 +27,29 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
private final ISessionsStore sessionsStore;
public ConnectionDescriptorStore(ISessionsStore sessionsStore) {
this.connectionDescriptors = new ConcurrentHashMap<String, ConnectionDescriptor>();
this.connectionDescriptors = new ConcurrentHashMap<>();
this.sessionsStore = sessionsStore;
}
public boolean sendMessage(AbstractMessage message, String clientId) {
return sendMessage(message, null, clientId);
}
public boolean sendMessage(AbstractMessage message, Integer messageID, String clientID) {
public boolean sendMessage(MqttMessage message, Integer messageID, String clientID) {
final MqttMessageType messageType = message.fixedHeader().messageType();
try {
// The connection descriptors map will never be null: it's a final
// attribute.
if (messageID != null) {
LOG.info("Sending {} message. MqttClientId = {}, messageId = {}.", message.getMessageType(), clientID,
messageID);
LOG.info("Sending {} message CId=<{}>, messageId={}", messageType, clientID, messageID);
} else {
LOG.debug("Sending {} message. MqttClientId = {}.", message.getMessageType(), clientID);
LOG.debug("Sending {} message CId=<{}>", messageType, clientID);
}
ConnectionDescriptor descriptor = connectionDescriptors.get(clientID);
if (descriptor == null) {
if (messageID != null) {
LOG.error(
"The client has just disconnected. The {} message could not be sent. MqttClientId = {}, messageId = {}.",
message.getMessageType(), clientID, messageID);
LOG.error("Client has just disconnected. The {} message could not be sent. CId=<{}>, messageId={}",
messageType, clientID, messageID);
} else {
LOG.error("The client has just disconnected. The {} could not be sent. MqttClientId = {}.",
message.getMessageType(), clientID);
LOG.error("Client has just disconnected. The {} could not be sent. CId=<{}>", messageType, clientID);
}
/*
* If the client hast just disconnected, its connection
* If the client has just disconnected, its connection
* descriptor will be null. We don't have to make the broker
* crash: we'll just discard the PUBACK message.
*/
@@ -67,14 +58,11 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
descriptor.writeAndFlush(message);
return true;
} catch (Throwable e) {
String errorMsg = "Unable to send " + messageType + " message. CId=<" + clientID + ">";
if (messageID != null) {
LOG.error(
"Unable to send {} message. MqttClientId = {}, messageId = {}, cause = {}, errorMessage = {}.",
message.getMessageType(), clientID, messageID, e.getCause(), e.getMessage());
} else {
LOG.error("Unable to send {} message. MqttClientId = {}, cause = {}, errorMessage = {}.",
message.getMessageType(), clientID, e.getCause(), e.getMessage());
errorMsg += ", messageId=" + messageID;
}
LOG.error(errorMsg, e);
return false;
}
}
@@ -111,7 +99,7 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
ConnectionDescriptor descriptor = connectionDescriptors.get(clientID);
if (descriptor == null) {
LOG.error(
"The connection descriptor doesn't exist. The MQTT connection cannot be closed. MqttClientId = {}, closeImmediately = {}.",
"The connection descriptor doesn't exist. The MQTT connection cannot be closed. CId=<{}>, closeImmediately = {}.",
clientID, closeImmediately);
return false;
}
@@ -125,10 +113,10 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
@Override
public MqttSession getSessionStatus(String clientID) {
LOG.info("Retrieving status of session. MqttClientId = {}.", clientID);
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. MqttClientId = {}.", clientID);
LOG.error("The given MQTT client ID doesn't have an associated session. CId=<{}>", clientID);
return null;
}
return buildMqttSession(session);
@@ -166,7 +154,7 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
result.setPendingPublishMessagesNo(session.getPendingPublishMessagesNo());
result.setSecondPhaseAckPendingMessages(session.getSecondPhaseAckPendingMessages());
result.setInflightMessages(session.getInflightMessagesNo());
LOG.info("The status of the session has been retrieved successfully. MqttClientId = {}.", session.clientID);
LOG.info("The status of the session has been retrieved successfully. CId=<{}>", session.clientID);
return result;
}
broker/src/main/java/io/moquette/server/HazelcastListener.java
+11 -11
View File
@@ -3,13 +3,12 @@ package io.moquette.server;
import com.hazelcast.core.Message;
import com.hazelcast.core.MessageListener;
import io.moquette.interception.HazelcastMsg;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.parser.proto.messages.PublishMessage;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
/**
* Created by mackristof on 28/05/2016.
*/
@@ -29,13 +28,14 @@ public class HazelcastListener implements MessageListener<HazelcastMsg> {
HazelcastMsg hzMsg = msg.getMessageObject();
LOG.info("{} received from hazelcast for topic {} message: {}", hzMsg.getClientId(),
hzMsg.getTopic(), hzMsg.getPayload());
PublishMessage publishMessage = new PublishMessage();
publishMessage.setTopicName(hzMsg.getTopic());
publishMessage.setQos(AbstractMessage.QOSType.valueOf(hzMsg.getQos()));
publishMessage.setPayload(ByteBuffer.wrap(hzMsg.getPayload()));
publishMessage.setLocal(false);
publishMessage.setClientId(hzMsg.getClientId());
server.internalPublish(publishMessage);
//TODO pass forward this information in somehow publishMessage.setLocal(false);
MqttQoS qos = MqttQoS.valueOf(hzMsg.getQos());
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, false, 0);
MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(hzMsg.getTopic(), 0);
ByteBuf payload = Unpooled.wrappedBuffer(hzMsg.getPayload());
MqttPublishMessage publishMessage = new MqttPublishMessage(fixedHeader, varHeader, payload);
server.internalPublish(publishMessage, hzMsg.getClientId());
}
} catch (Exception ex) {
LOG.error("error polling hazelcast msg queue", ex);
broker/src/main/java/io/moquette/server/Server.java
+10 -10
View File
@@ -27,15 +27,15 @@ import io.moquette.connections.IConnectionsManager;
import io.moquette.interception.HazelcastInterceptHandler;
import io.moquette.interception.HazelcastMsg;
import io.moquette.interception.InterceptHandler;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.config.*;
import io.moquette.spi.impl.ProtocolProcessorBootstrapper;
import io.moquette.server.netty.NettyAcceptor;
import io.moquette.spi.impl.ProtocolProcessor;
import io.moquette.spi.impl.ProtocolProcessorBootstrapper;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
import io.moquette.spi.security.ISslContextCreator;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -221,16 +221,18 @@ public class Server {
* It can be used only after the server is correctly started with startServer.
*
* @param msg the message to forward.
* @param clientId the id of the sending server.
* @throws IllegalStateException if the server is not yet started
* */
public void internalPublish(PublishMessage msg) {
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 = {}.",
msg.getClientId(), msg.getMessageID());
LOG.error("The server is not started. The message cannot be published. MqttClientId = {}, messageId = {}",
clientId, messageID);
throw new IllegalStateException("Can't publish on a server is not yet started");
}
LOG.debug("Publishing message. MqttClientId = {}, messageId = {}.", msg.getClientId(), msg.getMessageID());
m_processor.internalPublish(msg);
LOG.debug("Publishing message. MqttClientId = {}, messageId = {}", clientId, messageID);
m_processor.internalPublish(msg, clientId);
}
public void stopServer() {
@@ -266,7 +268,6 @@ public class Server {
/**
* SPI method used by Broker embedded applications to add intercept handlers.
* @param interceptHandler the handler to add.
* @return true id operation was successful.
* */
public void addInterceptHandler(InterceptHandler interceptHandler) {
if (!m_initialized) {
@@ -282,7 +283,6 @@ public class Server {
/**
* SPI method used by Broker embedded applications to remove intercept handlers.
* @param interceptHandler the handler to remove.
* @return true id operation was successful.
* */
public void removeInterceptHandler(InterceptHandler interceptHandler) {
if (!m_initialized) {
@@ -297,7 +297,7 @@ public class Server {
/**
* Returns the connections manager of this broker.
*
* @return
* @return IConnectionsManager the instance used bt the broker.
*/
public IConnectionsManager getConnectionsManager() {
return m_processorBootstrapper.getConnectionDescriptors();
broker/src/main/java/io/moquette/server/netty/MessageBuilder.java
+220
View File
@@ -0,0 +1,220 @@
/*
* Copyright (c) 2012-2017 The original author or authorsgetRockQuestions()
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.server.netty;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.*;
import java.util.ArrayList;
import java.util.List;
public class MessageBuilder {
public static class PublishBuilder {
private String topic = null;
private boolean retained = false;
private MqttQoS qos;
private byte[] payload;
private int messageId = 0;
public PublishBuilder topicName(String topic) {
this.topic = topic;
return PublishBuilder.this;
}
public PublishBuilder retained(boolean retained) {
this.retained = retained;
return this;
}
public PublishBuilder qos(MqttQoS qos) {
this.qos = qos;
return this;
}
public PublishBuilder payload(byte[] payload) {
this.payload = payload;
return this;
}
public PublishBuilder messageId(int messageId) {
this.messageId = messageId;
return this;
}
public MqttPublishMessage build() {
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retained, 0);
MqttPublishVariableHeader mqttVariableHeader = new MqttPublishVariableHeader(topic, messageId);
return new MqttPublishMessage(mqttFixedHeader, mqttVariableHeader, Unpooled.buffer().writeBytes(payload));
}
}
public static class ConnectBuilder {
private MqttVersion version = MqttVersion.MQTT_3_1_1;
private String clientId;
private boolean cleanSession = false;
private boolean hasUser = false;
private boolean hasPassword = false;
private int keepAliveSecs;
private boolean willFlag = false;
private MqttQoS willQos = MqttQoS.AT_MOST_ONCE;
private String willTopic = null;
private String willMessage = null;
private String username = null;
private String password = null;
public ConnectBuilder protocolVersion(MqttVersion version) {
this.version = version;
return this;
}
public ConnectBuilder clientId(String clientId) {
this.clientId = clientId;
return this;
}
public ConnectBuilder cleanSession(boolean cleanSession) {
this.cleanSession = cleanSession;
return this;
}
public ConnectBuilder keepAlive(int keepAliveSecs) {
this.keepAliveSecs = keepAliveSecs;
return this;
}
public ConnectBuilder willFlag(boolean willFlag) {
this.willFlag = willFlag;
return this;
}
public ConnectBuilder willQoS(MqttQoS willQos) {
this.willQos = willQos;
return this;
}
public ConnectBuilder willTopic(String willTopic) {
this.willTopic = willTopic;
return this;
}
public ConnectBuilder willMessage(String willMessage) {
this.willMessage = willMessage;
return this;
}
public ConnectBuilder hasUser() {
this.hasUser = true;
return this;
}
public ConnectBuilder hasPassword() {
this.hasPassword = true;
return this;
}
public ConnectBuilder username(String username) {
this.username = username;
return this;
}
public ConnectBuilder password(String password) {
this.password = password;
return this;
}
public MqttConnectMessage build() {
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttConnectVariableHeader mqttConnectVariableHeader =
new MqttConnectVariableHeader(
version.protocolName(),
version.protocolLevel(),
hasUser,
hasPassword,
false,
willQos.value(),
willFlag,
cleanSession,
keepAliveSecs);
MqttConnectPayload mqttConnectPayload = new MqttConnectPayload(clientId, willTopic, willMessage, username, password);
return new MqttConnectMessage(mqttFixedHeader, mqttConnectVariableHeader, mqttConnectPayload);
}
}
public static class SubscribeBuilder {
private List<MqttTopicSubscription> subscriptions = new ArrayList<>();
private int messageId;
public SubscribeBuilder addSubscription(MqttQoS qos, String topic) {
subscriptions.add(new MqttTopicSubscription(topic, qos));
return this;
}
public SubscribeBuilder messageId(int messageId) {
this.messageId = messageId;
return this;
}
public MqttSubscribeMessage build() {
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.SUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0);
MqttMessageIdVariableHeader mqttVariableHeader = MqttMessageIdVariableHeader.from(messageId);
MqttSubscribePayload mqttSubscribePayload = new MqttSubscribePayload(subscriptions);
return new MqttSubscribeMessage(mqttFixedHeader, mqttVariableHeader, mqttSubscribePayload);
}
}
public static class UnsubscribeBuilder {
private List<String> topicFilters = new ArrayList<>();
private int messageId = 0;
public UnsubscribeBuilder addTopicFilter(String topic) {
topicFilters.add(topic);
return this;
}
public UnsubscribeBuilder messageId(int messageId) {
this.messageId = messageId;
return this;
}
public MqttUnsubscribeMessage build() {
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBSCRIBE, false,
MqttQoS.AT_LEAST_ONCE, false, 0);
MqttMessageIdVariableHeader mqttVariableHeader = MqttMessageIdVariableHeader.from(messageId);
MqttUnsubscribePayload mqttSubscribePayload = new MqttUnsubscribePayload(topicFilters);
return new MqttUnsubscribeMessage(mqttFixedHeader, mqttVariableHeader, mqttSubscribePayload);
}
}
public static ConnectBuilder connect() {
return new ConnectBuilder();
}
public static PublishBuilder publish() {
return new PublishBuilder();
}
public static SubscribeBuilder subscribe() {
return new SubscribeBuilder();
}
public static UnsubscribeBuilder unsubscribe() {
return new UnsubscribeBuilder();
}
}
broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
+13 -12
View File
@@ -16,14 +16,11 @@
package io.moquette.server.netty;
import io.moquette.BrokerConstants;
import static io.moquette.BrokerConstants.*;
import io.moquette.parser.netty.MQTTDecoder;
import io.moquette.parser.netty.MQTTEncoder;
import io.moquette.spi.security.ISslContextCreator;
import io.moquette.server.ServerAcceptor;
import io.moquette.server.config.IConfig;
import io.moquette.server.netty.metrics.*;
import io.moquette.spi.impl.ProtocolProcessor;
import io.moquette.spi.security.ISslContextCreator;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
@@ -38,6 +35,8 @@ import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
@@ -50,6 +49,8 @@ import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static io.moquette.BrokerConstants.*;
/**
*
* @author andrea
@@ -179,8 +180,8 @@ public class NettyAcceptor implements ServerAcceptor {
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
// pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("decoder", new MqttDecoder());
pipeline.addLast("encoder", MqttEncoder.INSTANCE);
pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("messageLogger", new MQTTMessageLogger());
pipeline.addLast("handler", handler);
@@ -213,8 +214,8 @@ public class NettyAcceptor implements ServerAcceptor {
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds));
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("decoder", new MqttDecoder());
pipeline.addLast("encoder", MqttEncoder.INSTANCE);
pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("messageLogger", new MQTTMessageLogger());
pipeline.addLast("handler", handler);
@@ -247,8 +248,8 @@ public class NettyAcceptor implements ServerAcceptor {
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
//pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("decoder", new MqttDecoder());
pipeline.addLast("encoder", MqttEncoder.INSTANCE);
pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("messageLogger", new MQTTMessageLogger());
pipeline.addLast("handler", handler);
@@ -283,8 +284,8 @@ public class NettyAcceptor implements ServerAcceptor {
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds));
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("decoder", new MqttDecoder());
pipeline.addLast("encoder", MqttEncoder.INSTANCE);
pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("messageLogger", new MQTTMessageLogger());
pipeline.addLast("handler", handler);
broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java
+18 -21
View File
@@ -15,18 +15,16 @@
*/
package io.moquette.server.netty;
import io.moquette.parser.proto.Utils;
import io.moquette.parser.proto.messages.*;
import io.moquette.spi.impl.ProtocolProcessor;
import static io.moquette.parser.proto.messages.AbstractMessage.*;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
/**
*
@@ -44,47 +42,46 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
AbstractMessage msg = (AbstractMessage) message;
String messageType = Utils.msgType2String(msg.getMessageType());
LOG.debug("Processing MQTT message. MessageType = {}.", messageType);
MqttMessage msg = (MqttMessage) message;
MqttMessageType messageType = msg.fixedHeader().messageType();
LOG.debug("Processing MQTT message, type={}", messageType);
try {
switch (msg.getMessageType()) {
switch (messageType) {
case CONNECT:
m_processor.processConnect(ctx.channel(), (ConnectMessage) msg);
m_processor.processConnect(ctx.channel(), (MqttConnectMessage) msg);
break;
case SUBSCRIBE:
m_processor.processSubscribe(ctx.channel(), (SubscribeMessage) msg);
m_processor.processSubscribe(ctx.channel(), (MqttSubscribeMessage) msg);
break;
case UNSUBSCRIBE:
m_processor.processUnsubscribe(ctx.channel(), (UnsubscribeMessage) msg);
m_processor.processUnsubscribe(ctx.channel(), (MqttUnsubscribeMessage) msg);
break;
case PUBLISH:
m_processor.processPublish(ctx.channel(), (PublishMessage) msg);
m_processor.processPublish(ctx.channel(), (MqttPublishMessage) msg);
break;
case PUBREC:
m_processor.processPubRec(ctx.channel(), (PubRecMessage) msg);
m_processor.processPubRec(ctx.channel(), msg);
break;
case PUBCOMP:
m_processor.processPubComp(ctx.channel(), (PubCompMessage) msg);
m_processor.processPubComp(ctx.channel(), msg);
break;
case PUBREL:
m_processor.processPubRel(ctx.channel(), (PubRelMessage) msg);
m_processor.processPubRel(ctx.channel(), msg);
break;
case DISCONNECT:
m_processor.processDisconnect(ctx.channel());
break;
case PUBACK:
m_processor.processPubAck(ctx.channel(), (PubAckMessage) msg);
m_processor.processPubAck(ctx.channel(), (MqttPubAckMessage) msg);
break;
case PINGREQ:
PingRespMessage pingResp = new PingRespMessage();
MqttFixedHeader pingHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, AT_MOST_ONCE, false, 0);
MqttMessage pingResp = new MqttMessage(pingHeader);
ctx.writeAndFlush(pingResp);
break;
}
} catch (Throwable ex) {
LOG.error(
"An unexpected exception was caught while processing MQTT message. MessageType = {}, cause = {}, errorMessage = {}.",
messageType, ex.getCause(), ex.getMessage());
LOG.error("Exception was caught while processing MQTT message, " + ex.getCause(), ex);
ctx.fireExceptionCaught(ex);
}
}
broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java
+17 -41
View File
@@ -15,16 +15,16 @@
*/
package io.moquette.server.netty.metrics;
import io.moquette.parser.proto.messages.*;
import io.moquette.server.netty.NettyUtils;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static io.moquette.parser.proto.messages.AbstractMessage.*;
import static io.moquette.spi.impl.Utils.messageId;
/**
*
@@ -42,46 +42,42 @@ public class MQTTMessageLogger extends ChannelDuplexHandler {
}
private void logMQTTMessage(ChannelHandlerContext ctx, Object message, String direction) {
if (!(message instanceof AbstractMessage)) {
if (!(message instanceof MqttMessage)) {
return;
}
AbstractMessage msg = (AbstractMessage) message;
MqttMessage msg = (MqttMessage) message;
String clientID = NettyUtils.clientID(ctx.channel());
switch (msg.getMessageType()) {
MqttMessageType messageType = msg.fixedHeader().messageType();
switch (messageType) {
case CONNECT:
ConnectMessage connect = (ConnectMessage) msg;
LOG.info("{} CONNECT client <{}>", direction, connect.getClientID());
LOG.info("{} CONNECT client <{}>", direction, clientID);
break;
case SUBSCRIBE:
SubscribeMessage subscribe = (SubscribeMessage) msg;
LOG.info("{} SUBSCRIBE <{}> to topics {}", direction, clientID, subscribe.subscriptions());
MqttSubscribeMessage subscribe = (MqttSubscribeMessage) msg;
LOG.info("{} SUBSCRIBE <{}> to topics {}", direction, clientID, subscribe.payload().topicSubscriptions());
break;
case UNSUBSCRIBE:
UnsubscribeMessage unsubscribe = (UnsubscribeMessage) msg;
LOG.info("{} UNSUBSCRIBE <{}> to topics <{}>", direction, clientID, unsubscribe.topicFilters());
MqttUnsubscribeMessage unsubscribe = (MqttUnsubscribeMessage) msg;
LOG.info("{} UNSUBSCRIBE <{}> to topics <{}>", direction, clientID, unsubscribe.payload().topics());
break;
case PUBLISH:
PublishMessage publish = (PublishMessage) msg;
LOG.info("{} PUBLISH <{}> to topics <{}>", direction, clientID, publish.getTopicName());
MqttPublishMessage publish = (MqttPublishMessage) msg;
LOG.info("{} PUBLISH <{}> to topics <{}>", direction, clientID, publish.variableHeader().topicName());
break;
case PUBREC:
PubRecMessage pubrec = (PubRecMessage) msg;
LOG.info("{} PUBREC <{}> packetID <{}>", direction, clientID, pubrec.getMessageID());
LOG.info("{} PUBREC <{}> packetID <{}>", direction, clientID, messageId(msg));
break;
case PUBCOMP:
PubCompMessage pubCompleted = (PubCompMessage) msg;
LOG.info("{} PUBCOMP <{}> packetID <{}>", direction, clientID, pubCompleted.getMessageID());
LOG.info("{} PUBCOMP <{}> packetID <{}>", direction, clientID, messageId(msg));
break;
case PUBREL:
PubRelMessage pubRelease = (PubRelMessage) msg;
LOG.info("{} PUBREL <{}> packetID <{}>", direction, clientID, pubRelease.getMessageID());
LOG.info("{} PUBREL <{}> packetID <{}>", direction, clientID, messageId(msg));
break;
case DISCONNECT:
LOG.info("{} DISCONNECT <{}>", direction, clientID);
break;
case PUBACK:
PubAckMessage pubAck = (PubAckMessage) msg;
LOG.info("{} PUBACK <{}> packetID <{}>", direction, clientID, pubAck.getMessageID());
LOG.info("{} PUBACK <{}> packetID <{}>", direction, clientID, messageId(msg));
break;
}
}
@@ -100,24 +96,4 @@ public class MQTTMessageLogger extends ChannelDuplexHandler {
logMQTTMessage(ctx, msg, "C<-B");
ctx.write(msg, promise);
}
//
// @Override
// public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
// if (cause instanceof CorruptedFrameException) {
// //something goes bad with decoding
// LOG.warn("Error decoding a packet, probably a bad formatted packet, message: " + cause.getMessage());
// } else {
// LOG.error("Ugly error on networking", cause);
// }
// ctx.close();
// }
//
// @Override
// public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
// if (ctx.channel().isWritable()) {
// m_processor.notifyChannelWritable(ctx.channel());
// }
// ctx.fireChannelWritabilityChanged();
// }
}
broker/src/main/java/io/moquette/spi/ClientSession.java
+2 -3
View File
@@ -15,7 +15,6 @@
*/
package io.moquette.spi;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
@@ -85,7 +84,7 @@ public class ClientSession {
public boolean subscribe(Subscription newSubscription) {
LOG.info("Adding new subscription. MqttClientId = {}, topics = {}, qos = {}.", newSubscription.getClientId(),
newSubscription.getTopicFilter(), AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
newSubscription.getTopicFilter(), newSubscription.getRequestedQos());
boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter());
if (!validTopic) {
LOG.warn("The topic filter is not valid. MqttClientId = {}, topics = {}.", newSubscription.getClientId(),
@@ -96,7 +95,7 @@ public class ClientSession {
ClientTopicCouple matchingCouple = new ClientTopicCouple(this.clientID, newSubscription.getTopicFilter());
Subscription existingSub = m_sessionsStore.getSubscription(matchingCouple);
//update the selected subscriptions if not present or if has a greater qos
if (existingSub == null || existingSub.getRequestedQos().byteValue() < newSubscription.getRequestedQos().byteValue()) {
if (existingSub == null || existingSub.getRequestedQos().value() < newSubscription.getRequestedQos().value()) {
if (existingSub != null) {
LOG.info("The subscription already existed with a lower QoS value. It will be updated. MqttClientId = {}, topics = {}, existingQos = {}, newQos = {}.",
newSubscription.getClientId(), newSubscription.getTopicFilter(), existingSub.getRequestedQos(),
broker/src/main/java/io/moquette/spi/IMessagesStore.java
+9 -7
View File
@@ -15,10 +15,12 @@
*/
package io.moquette.spi;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.io.Serializable;
import java.nio.ByteBuffer;
import io.moquette.parser.proto.messages.AbstractMessage;
import java.util.Collection;
/**
@@ -28,7 +30,7 @@ public interface IMessagesStore {
class StoredMessage implements Serializable {
private static final long serialVersionUID = 1755296138639817304L;
final AbstractMessage.QOSType m_qos;
final MqttQoS m_qos;
final byte[] m_payload;
final String m_topic;
private boolean m_retained;
@@ -37,13 +39,13 @@ public interface IMessagesStore {
private Integer m_msgID;
private MessageGUID m_guid;
public StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) {
public StoredMessage(byte[] message, MqttQoS qos, String topic) {
m_qos = qos;
m_payload = message;
m_topic = topic;
}
public AbstractMessage.QOSType getQos() {
public MqttQoS getQos() {
return m_qos;
}
@@ -79,8 +81,8 @@ public interface IMessagesStore {
return m_msgID;
}
public ByteBuffer getMessage() {
return ByteBuffer.wrap(m_payload);
public ByteBuf getMessage() {
return Unpooled.copiedBuffer(m_payload);
}
public void setRetained(boolean retained) {
broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java
+17 -21
View File
@@ -15,6 +15,17 @@
*/
package io.moquette.spi.impl;
import io.moquette.BrokerConstants;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.Interceptor;
import io.moquette.interception.messages.*;
import io.moquette.server.config.IConfig;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -23,23 +34,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import io.moquette.BrokerConstants;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.Interceptor;
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
import io.moquette.interception.messages.InterceptConnectMessage;
import io.moquette.interception.messages.InterceptConnectionLostMessage;
import io.moquette.interception.messages.InterceptDisconnectMessage;
import io.moquette.interception.messages.InterceptPublishMessage;
import io.moquette.interception.messages.InterceptSubscribeMessage;
import io.moquette.interception.messages.InterceptUnsubscribeMessage;
import io.moquette.parser.proto.messages.ConnectMessage;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.config.IConfig;
import io.moquette.spi.impl.subscriptions.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static io.moquette.logging.LoggingUtils.getInterceptorIds;
/**
@@ -98,10 +92,10 @@ final class BrokerInterceptor implements Interceptor {
}
@Override
public void notifyClientConnected(final ConnectMessage msg) {
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.getClientID(), handler.getID());
msg.payload().clientIdentifier(), handler.getID());
executor.execute(new Runnable() {
@Override
public void run() {
@@ -140,10 +134,12 @@ final class BrokerInterceptor implements Interceptor {
}
@Override
public void notifyTopicPublished(final PublishMessage msg, final String clientID, final String username) {
public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) {
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, msg.getMessageID(), msg.getTopicName(), handler.getID());
"interceptorId = {}.", clientID, messageId, topic, handler.getID());
executor.execute(new Runnable() {
@Override
public void run() {
broker/src/main/java/io/moquette/spi/impl/DebugUtils.java
+9 -2
View File
@@ -15,6 +15,8 @@
*/
package io.moquette.spi.impl;
import io.netty.buffer.ByteBuf;
import java.nio.ByteBuffer;
/**
@@ -22,11 +24,16 @@ import java.nio.ByteBuffer;
* @author andrea
*/
class DebugUtils {
static String payload2Str(ByteBuffer content) {
static String payload2Str(ByteBuffer content) {
byte[] b = new byte[content.remaining()];
content.mark();
content.get(b);
content.reset();
return new String(b);
}
}
static String payload2Str(ByteBuf content) {
//byte[] b = new byte[content.readableBytes()];
return new String(content.copy().array());
}
}
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+27 -21
View File
@@ -1,14 +1,13 @@
package io.moquette.spi.impl;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
@@ -27,21 +26,23 @@ class InternalRepublisher {
void publishRetained(ClientSession targetSession, Collection<IMessagesStore.StoredMessage> messages) {
for (IMessagesStore.StoredMessage storedMsg : messages) {
//fire as retained the message
Integer packetID = storedMsg.getQos() == AbstractMessage.QOSType.MOST_ONE ? null : targetSession.nextPacketId();
Integer packetID = storedMsg.getQos() == MqttQoS.AT_MOST_ONCE ? null : targetSession.nextPacketId();
if (packetID != null) {
LOG.debug("Adding message to inflight zone. MqttClientId = {}, packetId = {}, messageId = {}, topic = {}.",
targetSession.clientID, packetID, storedMsg.getMessageID(), storedMsg.getTopic());
targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID);
}
PublishMessage publishMsg = retainedPublish(storedMsg);
MqttPublishMessage publishMsg = retainedPublish(storedMsg);
//set the PacketIdentifier only for QoS > 0
if (publishMsg.getQos() != AbstractMessage.QOSType.MOST_ONE) {
publishMsg.setMessageID(packetID);
if (publishMsg.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) {
publishMsg = retainedPublish(storedMsg, packetID);
}
this.messageSender.sendPublish(targetSession, publishMsg);
}
}
void publishStored(ClientSession clientSession, BlockingQueue<IMessagesStore.StoredMessage> publishedEvents) {
List<IMessagesStore.StoredMessage> storedPublishes = new ArrayList<>();
publishedEvents.drainTo(storedPublishes);
@@ -51,29 +52,34 @@ class InternalRepublisher {
LOG.debug("Adding message ot inflight zone. MqttClientId = {}, guid = {}, messageId = {}, topic = {}.",
clientSession.clientID, pubEvt.getGuid(), pubEvt.getMessageID(), pubEvt.getTopic());
clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID());
PublishMessage publishMsg = notRetainedPublish(pubEvt);
MqttPublishMessage publishMsg = notRetainedPublish(pubEvt);
//set the PacketIdentifier only for QoS > 0
if (publishMsg.getQos() != AbstractMessage.QOSType.MOST_ONE) {
publishMsg.setMessageID(pubEvt.getMessageID());
if (publishMsg.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) {
publishMsg = notRetainedPublish(pubEvt, pubEvt.getMessageID());
}
this.messageSender.sendPublish(clientSession, publishMsg);
}
}
private PublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false);
private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage, Integer messageID) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false, messageID);
}
private PublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true);
private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false, 0);
}
public static PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(retained);
pubMessage.setTopicName(topic);
pubMessage.setQos(qos);
pubMessage.setPayload(message);
return pubMessage;
private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true, 0);
}
private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage, Integer packetID) {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true, packetID);
}
public static MqttPublishMessage createPublishForQos(String topic, MqttQoS qos, ByteBuf message, boolean retained, int messageId) {
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retained, 0);
MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId);
return new MqttPublishMessage(fixedHeader, varHeader, message);
}
}
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+22 -22
View File
@@ -1,17 +1,16 @@
package io.moquette.spi.impl;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.List;
import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired;
@@ -32,50 +31,51 @@ class MessagesPublisher {
this.messageSender = messageSender;
}
static PublishMessage notRetainedPublish(String topic, AbstractMessage.QOSType qos, ByteBuffer message) {
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(false);
pubMessage.setTopicName(topic);
pubMessage.setQos(qos);
pubMessage.setPayload(message);
return pubMessage;
static MqttPublishMessage notRetainedPublish(String topic, MqttQoS qos, ByteBuf message) {
return notRetainedPublishWithMessageId(topic, qos, message, 0);
}
private static MqttPublishMessage notRetainedPublishWithMessageId(String topic, MqttQoS qos, ByteBuf message, int messageId) {
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, false, 0);
MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId);
return new MqttPublishMessage(fixedHeader, varHeader, message);
}
void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List<Subscription> topicMatchingSubscriptions) {
final String topic = pubMsg.getTopic();
final AbstractMessage.QOSType publishingQos = pubMsg.getQos();
final ByteBuffer origMessage = pubMsg.getMessage();
final MqttQoS publishingQos = pubMsg.getQos();
final ByteBuf origMessage = pubMsg.getMessage();
//if QoS 1 or 2 store the message
//TODO perhaps this block is not needed.
MessageGUID guid = null;
if (publishingQos != AbstractMessage.QOSType.MOST_ONE) {
if (publishingQos != MqttQoS.AT_MOST_ONCE) {
guid = m_messagesStore.storePublishForFuture(pubMsg);
}
for (final Subscription sub : topicMatchingSubscriptions) {
AbstractMessage.QOSType qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
MqttQoS qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
boolean targetIsActive = this.connectionDescriptors.isConnected(sub.getClientId());
ByteBuffer message = origMessage.duplicate();
//we need to retain because duplicate only copy r/w indexes and don't retain() causing refCnt = 0
ByteBuf message = origMessage.retainedDuplicate();
if (targetIsActive) {
LOG.debug("Sending PUBLISH message to active subscriber. MessageId = {}, mqttClientId = {}, topicFilter = {}, " +
"qos = {}.", pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos);
PublishMessage publishMsg = notRetainedPublish(topic, qos, message);
if (qos != AbstractMessage.QOSType.MOST_ONE) {
LOG.debug("Sending PUBLISH message to active subscriber. MessageId={}, CId={}, topicFilter={}, qos={}",
pubMsg.getMessageID(), 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);
//set the PacketIdentifier only for QoS > 0
publishMsg.setMessageID(messageId);
publishMsg = notRetainedPublishWithMessageId(topic, qos, message, messageId);
}
this.messageSender.sendPublish(targetSession, publishMsg);
} else {
if (!targetSession.isCleanSession()) {
LOG.debug("Storing pending PUBLISH inactive message. MessageId = {}, mqttClientId = {}, " +
"topicFilter = {}, qos = {}.",
LOG.debug("Storing pending PUBLISH inactive message. MessageId={}, CId={}, topicFilter={}, qos={}",
pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos);
//store the message in targetSession queue to deliver
targetSession.enqueue(pubMsg);
broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
+17 -16
View File
@@ -1,14 +1,14 @@
package io.moquette.spi.impl;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.spi.ClientSession;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
class PersistentQueueMessageSender {
@@ -19,28 +19,29 @@ class PersistentQueueMessageSender {
this.connectionDescriptorStore = connectionDescriptorStore;
}
void sendPublish(ClientSession clientsession, PublishMessage pubMessage) {
void sendPublish(ClientSession clientsession, MqttPublishMessage pubMessage) {
String clientId = clientsession.clientID;
final int messageId = pubMessage.variableHeader().messageId();
final String topicName = pubMessage.variableHeader().topicName();
if (LOG.isDebugEnabled()) {
LOG.debug("Sending PUBLISH message. MessageId = {}, mqttClientId = {}, topic = {}, qos = {}, payload = {}.",
pubMessage.getMessageID(),
clientId, pubMessage.getTopicName(), DebugUtils.payload2Str(pubMessage.getPayload()));
LOG.debug("Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}", messageId,
clientId, topicName, DebugUtils.payload2Str(pubMessage.payload()));
} else {
LOG.info("Sending PUBLISH message. MessageId = {}, mqttClientId = {}, topic = {}.", pubMessage.getMessageID(),
clientId, pubMessage.getTopicName());
LOG.info("Sending PUBLISH message. MessageId={}, CId={}, topic={}", messageId, clientId, topicName);
}
boolean messageDelivered = connectionDescriptorStore.sendMessage(pubMessage, pubMessage.getMessageID(), clientId);
if (!messageDelivered && pubMessage.getQos() != MOST_ONE && !clientsession.isCleanSession()) {
boolean messageDelivered = connectionDescriptorStore.sendMessage(pubMessage, messageId, clientId);
MqttQoS qos = pubMessage.fixedHeader().qosLevel();
if (!messageDelivered && qos != AT_MOST_ONCE && !clientsession.isCleanSession()) {
LOG.warn(
"The PUBLISH message could not be delivered. It will be stored. MessageId = {}, mqttClientId = {}, topic = {}, qos = {}, cleanSession = {}.",
pubMessage.getMessageID(), clientId, pubMessage.getTopicName(), pubMessage.getQos(), false);
"The PUBLISH message could not be delivered. It will be stored. MessageId = {}, CId={}, topic = {}, qos = {}, cleanSession = {}.",
messageId, clientId, topicName, qos, false);
clientsession.enqueue(asStoredMessage(pubMessage));
} else {
LOG.warn(
"The PUBLISH message could not be delivered. It will be discarded. MessageId = {}, mqttClientId = {}, topic = {}, qos = {}, cleanSession = {}.",
pubMessage.getMessageID(), clientId, pubMessage.getTopicName(), pubMessage.getQos(), true);
"The PUBLISH message could not be delivered. It will be discarded. MessageId = {}, CId={}, topic = {}, qos = {}, cleanSession = {}.",
messageId, clientId, topicName, qos, true);
}
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+227 -209
View File
@@ -15,35 +15,42 @@
*/
package io.moquette.spi.impl;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import io.moquette.interception.InterceptHandler;
import io.moquette.parser.proto.messages.*;
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.ConnectionDescriptor.ConnectionState;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.*;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Subscription;
import static io.moquette.parser.netty.Utils.VERSION_3_1;
import static io.moquette.parser.netty.Utils.VERSION_3_1_1;
import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos;
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
import io.moquette.parser.proto.messages.AbstractMessage.QOSType;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos;
import static io.moquette.spi.impl.Utils.messageId;
import static io.moquette.spi.impl.Utils.readBytesAndRewind;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
import static io.netty.handler.codec.mqtt.MqttQoS.*;
/**
* Class responsible to handle the logic of MQTT protocol it's the director of
* the protocol execution.
@@ -58,9 +65,9 @@ public class ProtocolProcessor {
private final String topic;
private final ByteBuffer payload;
private final boolean retained;
private final QOSType qos;
private final MqttQoS qos;
public WillMessage(String topic, ByteBuffer payload, boolean retained, QOSType qos) {
public WillMessage(String topic, ByteBuffer payload, boolean retained, MqttQoS qos) {
this.topic = topic;
this.payload = payload;
this.retained = retained;
@@ -79,7 +86,7 @@ public class ProtocolProcessor {
return retained;
}
public QOSType getQos() {
public MqttQoS getQos() {
return qos;
}
@@ -105,8 +112,7 @@ public class ProtocolProcessor {
RunningSubscription that = (RunningSubscription) o;
if (packetId != that.packetId) return false;
return clientID != null ? clientID.equals(that.clientID) : that.clientID == null;
return packetId == that.packetId && (clientID != null ? clientID.equals(that.clientID) : that.clientID == null);
}
@@ -217,65 +223,63 @@ public class ProtocolProcessor {
this.internalRepublisher = new InternalRepublisher(messageSender);
}
public void processConnect(Channel channel, ConnectMessage msg) {
LOG.info("Processing CONNECT message. MqttClientId = {}, username = {}, password = {}.", msg.getClientID(),
msg.getUsername(), msg.getPassword());
public void processConnect(Channel channel, MqttConnectMessage msg) {
MqttConnectPayload payload = msg.payload();
String clientId = payload.clientIdentifier();
LOG.info("Processing CONNECT message. CId = {}, username = {}", clientId, payload.userName());
if (msg.getProtocolVersion() != VERSION_3_1 && msg.getProtocolVersion() != VERSION_3_1_1) {
ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
LOG.error("The MQTT protocol version is not valid. MqttClientId = {}, protocolVersion = {}.",
msg.getProtocolVersion());
if (msg.variableHeader().version() != MqttVersion.MQTT_3_1.protocolLevel() &&
msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel()) {
MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
LOG.error("The MQTT protocol version is not valid. CId = {}", clientId);
channel.writeAndFlush(badProto);
channel.close();
return;
}
if (msg.getClientID() == null || msg.getClientID().length() == 0) {
if(!msg.isCleanSession() || !this.allowZeroByteClientId) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
channel.writeAndFlush(okResp);
if (clientId == null || clientId.length() == 0) {
if (!msg.variableHeader().isCleanSession() || !this.allowZeroByteClientId) {
MqttConnAckMessage badId = connAck(CONNECTION_REFUSED_IDENTIFIER_REJECTED);
channel.writeAndFlush(badId);
channel.close();
LOG.error("The MQTT client ID cannot be empty., Username = {}, password = {}.", msg.getUsername(),
msg.getPassword());
LOG.error("The MQTT client ID cannot be empty. Username = {}", payload.userName());
return;
}
// Generating client id.
String randomIdentifier = UUID.randomUUID().toString().replace("-", "");
msg.setClientID(randomIdentifier);
LOG.info("The client has connected with a server generated identifier. MqttClientId = {}, username = {}, password = {}.",
randomIdentifier, msg.getUsername(), msg.getPassword());
clientId = UUID.randomUUID().toString().replace("-", "");
LOG.info("The client has connected with a server generated identifier. CId = {}, username = {}",
clientId, payload.userName());
}
if (!login(channel, msg)) {
if (!login(channel, msg, clientId)) {
channel.close();
return;
}
final String clientID = msg.getClientID();
ConnectionDescriptor descriptor = new ConnectionDescriptor(clientID, channel, msg.isCleanSession());
ConnectionDescriptor descriptor = new ConnectionDescriptor(clientId, channel, msg.variableHeader().isCleanSession());
ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor);
if (existing != null) {
LOG.info("The client ID is being used in an existing connection. It will be closed. MqttClientId = {}.",
msg.getClientID());
LOG.info("The client ID is being used in an existing connection. It will be closed. CId = {}",
clientId);
existing.abort();
return;
}
initializeKeepAliveTimeout(channel, msg);
initializeKeepAliveTimeout(channel, msg, clientId);
storeWillMessage(msg);
storeWillMessage(msg, clientId);
if (!sendAck(descriptor, msg)) {
if (!sendAck(descriptor, msg, clientId)) {
channel.close();
return;
}
m_interceptor.notifyClientConnected(msg);
final ClientSession clientSession = createOrLoadClientSession(descriptor, msg);
final ClientSession clientSession = createOrLoadClientSession(descriptor, msg, clientId);
if (clientSession == null) {
channel.close();
return;
@@ -290,117 +294,133 @@ public class ProtocolProcessor {
channel.close();
}
LOG.info("The CONNECT message has been processed. MqttClientId = {}, username = {}, password = {}.",
msg.getClientID(), msg.getUsername(), msg.getPassword());
LOG.info("The CONNECT message has been processed. CId = {}, username = {}.", clientId, payload.userName());
}
private boolean login(Channel channel, ConnectMessage msg) {
private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) {
return connAck(returnCode, false);
}
private MqttConnAckMessage connAckWithSessionPresent(MqttConnectReturnCode returnCode) {
return connAck(returnCode, true);
}
private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode, boolean sessionPresent) {
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent);
return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}
private boolean login(Channel channel, MqttConnectMessage msg, final String clientId) {
//handle user authentication
if (msg.isUserFlag()) {
if (msg.variableHeader().hasUserName()) {
byte[] pwd = null;
if (msg.isPasswordFlag()) {
pwd = msg.getPassword();
if (msg.variableHeader().hasPassword()) {
pwd = msg.payload().password().getBytes();
} else if (!this.allowAnonymous) {
LOG.error("The client didn't supply any password and MQTT anonymous mode is disabled. MqttClientId = {}.",
msg.getClientID());
LOG.error("The client didn't supply any password and MQTT anonymous mode is disabled. CId = {}.",
clientId);
failedCredentials(channel);
return false;
}
if (!m_authenticator.checkValid(msg.getClientID(), msg.getUsername(), pwd)) {
LOG.error("The authenticator has rejected the MQTT credentials. MqttClientId = {}, username = {}, password = {}.",
msg.getClientID(), msg.getUsername(), pwd);
if (!m_authenticator.checkValid(clientId, msg.payload().userName(), pwd)) {
LOG.error("The authenticator has rejected the MQTT credentials. CId = {}, username = {}, password = {}.",
clientId, msg.payload().userName(), pwd);
failedCredentials(channel);
return false;
}
NettyUtils.userName(channel, msg.getUsername());
NettyUtils.userName(channel, msg.payload().userName());
} else if (!this.allowAnonymous) {
LOG.error("The client didn't supply any credentials and MQTT anonymous mode is disabled. MqttClientId = {}.",
msg.getClientID());
LOG.error("The client didn't supply any credentials and MQTT anonymous mode is disabled. CId = {}.",
clientId);
failedCredentials(channel);
return false;
}
return true;
}
private boolean sendAck(ConnectionDescriptor descriptor, ConnectMessage msg) {
LOG.info("Sending connect ACK. MqttClientId = {}.", msg.getClientID());
private boolean sendAck(ConnectionDescriptor descriptor, MqttConnectMessage msg, final String clientId) {
LOG.info("Sending connect ACK. CId = {}.", clientId);
final boolean success = descriptor.assignState(ConnectionState.DISCONNECTED, ConnectionState.SENDACK);
if (!success) {
return false;
}
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
ClientSession clientSession = m_sessionsStore.sessionForClient(msg.getClientID());
MqttConnAckMessage okResp;
ClientSession clientSession = m_sessionsStore.sessionForClient(clientId);
boolean isSessionAlreadyStored = clientSession != null;
if (!msg.isCleanSession() && isSessionAlreadyStored) {
okResp.setSessionPresent(true);
if (!msg.variableHeader().isCleanSession() && isSessionAlreadyStored) {
okResp = connAckWithSessionPresent(CONNECTION_ACCEPTED);
} else {
okResp = connAck(CONNECTION_ACCEPTED);
}
if (isSessionAlreadyStored) {
LOG.info("Cleaning session. MqttClientId = {}.", msg.getClientID());
clientSession.cleanSession(msg.isCleanSession());
LOG.info("Cleaning session. CId = {}.", clientId);
clientSession.cleanSession(msg.variableHeader().isCleanSession());
}
descriptor.writeAndFlush(okResp);
LOG.info("The connect ACK has been sent. MqttClientId = {}.", msg.getClientID());
LOG.info("The connect ACK has been sent. CId = {}.", clientId);
return true;
}
private void initializeKeepAliveTimeout(Channel channel, ConnectMessage msg) {
int keepAlive = msg.getKeepAlive();
LOG.info("Configuring connection. MqttClientId = {}.", msg.getClientID());
private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, final String clientId) {
int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
LOG.info("Configuring connection. CId = {}.", clientId);
NettyUtils.keepAlive(channel, keepAlive);
//session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession());
NettyUtils.cleanSession(channel, msg.isCleanSession());
//session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.variableHeader().isCleanSession());
NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession());
//used to track the client in the subscription and publishing phases.
//session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID());
NettyUtils.clientID(channel, msg.getClientID());
NettyUtils.clientID(channel, clientId);
int idleTime = Math.round(keepAlive * 1.5f);
setIdleTime(channel.pipeline(), idleTime);
LOG.info("The connection has been configured. MqttClientId = {}, keepAlive = {}, cleanSession = {}, idleTime = {}.",
msg.getClientID(), keepAlive, msg.isCleanSession(), idleTime);
LOG.info("The connection has been configured. CId = {}, keepAlive = {}, cleanSession = {}, idleTime = {}.",
clientId, keepAlive, msg.variableHeader().isCleanSession(), idleTime);
}
private void storeWillMessage(ConnectMessage msg) {
private void storeWillMessage(MqttConnectMessage msg, final String clientId) {
//Handle will flag
if (msg.isWillFlag()) {
AbstractMessage.QOSType willQos = AbstractMessage.QOSType.valueOf(msg.getWillQos());
LOG.info("Configuring MQTT last will and testament. MqttClientId = {}, willTopic = {}, willQos = {}, willRetain = {}.",
msg.getClientID(), willQos, msg.getWillTopic(), msg.isWillRetain());
byte[] willPayload = msg.getWillMessage();
if (msg.variableHeader().isWillFlag()) {
MqttQoS willQos = MqttQoS.valueOf(msg.variableHeader().willQos());
LOG.info("Configuring MQTT last will and testament. CId = {}, willQos = {}, willTopic = {}, willRetain = {}.",
clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain());
byte[] willPayload = msg.payload().willMessage().getBytes();
ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip();
//save the will testament in the clientID store
WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos);
m_willStore.put(msg.getClientID(), will);
LOG.info("MQTT last will and testament has been configured. MqttClientId = {}.", msg.getClientID());
WillMessage will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(), willQos);
m_willStore.put(clientId, will);
LOG.info("MQTT last will and testament has been configured. CId = {}.", clientId);
}
}
private ClientSession createOrLoadClientSession(ConnectionDescriptor descriptor, ConnectMessage msg) {
private ClientSession createOrLoadClientSession(ConnectionDescriptor descriptor, MqttConnectMessage msg,
String clientId) {
final boolean success = descriptor.assignState(ConnectionState.SENDACK, ConnectionState.SESSION_CREATED);
if (!success) {
return null;
}
ClientSession clientSession = m_sessionsStore.sessionForClient(msg.getClientID());
ClientSession clientSession = m_sessionsStore.sessionForClient(clientId);
boolean isSessionAlreadyStored = clientSession != null;
if (!isSessionAlreadyStored) {
clientSession = m_sessionsStore.createNewSession(msg.getClientID(), msg.isCleanSession());
clientSession = m_sessionsStore.createNewSession(clientId, msg.variableHeader().isCleanSession());
}
if (msg.isCleanSession()) {
LOG.info("Cleaning session. MqttClientId = {}.", msg.getClientID());
if (msg.variableHeader().isCleanSession()) {
LOG.info("Cleaning session. CId = {}.", clientId);
clientSession.cleanSession();
}
return clientSession;
}
private boolean republish(ConnectionDescriptor descriptor, ConnectMessage msg, ClientSession clientSession) {
private boolean republish(ConnectionDescriptor descriptor, MqttConnectMessage msg, ClientSession clientSession) {
final boolean success = descriptor.assignState(ConnectionState.SESSION_CREATED, ConnectionState.MESSAGES_REPUBLISHED);
if (!success) {
return false;
}
if (!msg.isCleanSession()) {
if (!msg.variableHeader().isCleanSession()) {
//force the republish of stored QoS1 and QoS2
republishStoredInSession(clientSession);
}
@@ -410,9 +430,7 @@ public class ProtocolProcessor {
}
private void failedCredentials(Channel session) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
session.writeAndFlush(okResp);
session.writeAndFlush(connAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD));
LOG.info("Client {} failed to connect with bad username or password.", session);
}
@@ -427,7 +445,7 @@ public class ProtocolProcessor {
* Republish QoS1 and QoS2 messages stored into the session for the clientID.
* */
private void republishStoredInSession(ClientSession clientSession) {
LOG.info("Republishing stored publish events. MqttClientId = {}.", clientSession.clientID);
LOG.info("Republishing stored publish events. CId = {}.", clientSession.clientID);
BlockingQueue<StoredMessage> publishedEvents = clientSession.queue();
if (publishedEvents.isEmpty()) {
LOG.info("There are no stored publish events. ClientId = {}.", clientSession.clientID);
@@ -437,9 +455,9 @@ public class ProtocolProcessor {
this.internalRepublisher.publishStored(clientSession, publishedEvents);
}
public void processPubAck(Channel channel, PubAckMessage msg) {
public void processPubAck(Channel channel, MqttPubAckMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
int messageID = msg.variableHeader().messageId();
String username = NettyUtils.userName(channel);
LOG.trace("retrieving inflight for messageID <{}>", messageID);
@@ -448,18 +466,18 @@ public class ProtocolProcessor {
targetSession.inFlightAcknowledged(messageID);
String topic = inflightMsg.getTopic();
// MessageGUID guid = inflightMsg.getGuid();
//Remove the message from message store
// m_messagesStore.decUsageCounter(guid);
m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username));
}
public static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) {
IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(msg.getPayload().array(), msg.getQos(), msg.getTopicName());
stored.setRetained(msg.isRetainFlag());
stored.setMessageID(msg.getMessageID());
public static IMessagesStore.StoredMessage asStoredMessage(MqttPublishMessage msg) {
//TODO ugly, too much array copy
ByteBuf payload = msg.payload();
byte[] payloadContent = readBytesAndRewind(payload);
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;
}
@@ -469,15 +487,16 @@ public class ProtocolProcessor {
return pub;
}
public void processPublish(Channel channel, PublishMessage msg) {
LOG.info("Processing PUBLISH message. MqttClientId = {}, topic = {}, messageId = {}, qos = {}.",
msg.getClientId(), msg.getTopicName(), msg.getMessageID(), msg.getQos());
final AbstractMessage.QOSType qos = msg.getQos();
public void processPublish(Channel channel, MqttPublishMessage msg) {
final MqttQoS qos = msg.fixedHeader().qosLevel();
final String clientId = NettyUtils.clientID(channel);
LOG.info("Processing PUBLISH message. CId = {}, topic = {}, messageId = {}, qos = {}.",
clientId, msg.variableHeader().topicName(), msg.variableHeader().messageId(), qos);
switch (qos) {
case MOST_ONE:
case AT_MOST_ONCE:
this.qos0PublishHandler.receivedPublishQos0(channel, msg);
break;
case LEAST_ONE:
case AT_LEAST_ONCE:
this.qos1PublishHandler.receivedPublishQos1(channel, msg);
break;
case EXACTLY_ONCE:
@@ -497,30 +516,31 @@ public class ProtocolProcessor {
* it's publishing.
*
* @param msg the message to publish.
* @param clientId the clientID
* */
public void internalPublish(PublishMessage msg) {
final AbstractMessage.QOSType qos = msg.getQos();
final String topic = msg.getTopicName();
public void internalPublish(MqttPublishMessage msg, final String clientId) {
final MqttQoS qos = msg.fixedHeader().qosLevel();
final String topic = msg.variableHeader().topicName();
LOG.info("Sending PUBLISH message. Topic = {}, qos = {}.", topic, qos);
MessageGUID guid = null;
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
if (msg.getClientId() == null || msg.getClientId().isEmpty()) {
if (clientId == null || clientId.isEmpty()) {
toStoreMsg.setClientID("BROKER_SELF");
} else {
toStoreMsg.setClientID(msg.getClientId());
toStoreMsg.setClientID(clientId);
}
toStoreMsg.setMessageID(1);
if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2
if (qos == EXACTLY_ONCE) { //QoS2
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.messagesPublisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
if (!msg.isRetainFlag()) {
if (!msg.fixedHeader().isRetain()) {
return;
}
if (qos == AbstractMessage.QOSType.MOST_ONE || !msg.getPayload().hasRemaining()) {
if (qos == AT_MOST_ONCE || msg.payload().readableBytes() == 0) {
//QoS == 0 && retain => clean old retained
m_messagesStore.cleanRetained(topic);
return;
@@ -539,7 +559,7 @@ public class ProtocolProcessor {
//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() != AbstractMessage.QOSType.MOST_ONE) {
if (will.getQos() != AT_MOST_ONCE) {
messageId = m_sessionsStore.nextPacketID(clientID);
}
@@ -549,12 +569,12 @@ public class ProtocolProcessor {
String topic = tobeStored.getTopic();
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
LOG.info("Publishing will message. MqttClientId = {}, messageId = {}, topic = {}.", clientID, messageId, topic);
LOG.info("Publishing will message. CId = {}, messageId = {}, topic = {}.", clientID, messageId, topic);
this.messagesPublisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions);
}
static QOSType lowerQosToTheSubscriptionDesired(Subscription sub, QOSType qos) {
if (qos.byteValue() > sub.getRequestedQos().byteValue()) {
static MqttQoS lowerQosToTheSubscriptionDesired(Subscription sub, MqttQoS qos) {
if (qos.value() > sub.getRequestedQos().value()) {
qos = sub.getRequestedQos();
}
return qos;
@@ -566,29 +586,28 @@ public class ProtocolProcessor {
* @param channel the channel of the incoming message.
* @param msg the decoded pubrel message.
* */
public void processPubRel(Channel channel, PubRelMessage msg) {
public void processPubRel(Channel channel, MqttMessage msg) {
this.qos2PublishHandler.processPubRel(channel, msg);
}
public void processPubRec(Channel channel, PubRecMessage msg) {
public void processPubRec(Channel channel, MqttMessage msg) {
String clientID = NettyUtils.clientID(channel);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
//remove from the inflight and move to the QoS2 second phase queue
int messageID = msg.getMessageID();
int messageID = messageId(msg);
targetSession.moveInFlightToSecondPhaseAckWaiting(messageID);
//once received a PUBREC reply with a PUBREL(messageID)
LOG.debug("Processing PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID);
PubRelMessage pubRelMessage = new PubRelMessage();
pubRelMessage.setMessageID(messageID);
pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE);
LOG.debug("Processing PUBREC message. CId = {}, messageId = {}.", clientID, messageID);
MqttFixedHeader pubRelHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0);
MqttMessage pubRelMessage = new MqttMessage(pubRelHeader, from(messageID));
channel.writeAndFlush(pubRelMessage);
}
public void processPubComp(Channel channel, PubCompMessage msg) {
public void processPubComp(Channel channel, MqttMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
LOG.debug("Processing PUBCOMP message. MqttClientId = {}, messageId = {}.", clientID, messageID);
int messageID = messageId(msg);
LOG.debug("Processing PUBCOMP message. CId = {}, messageId = {}.", clientID, messageID);
//once received the PUBCOMP then remove the message from the temp memory
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID);
@@ -599,7 +618,7 @@ public class ProtocolProcessor {
public void processDisconnect(Channel channel) throws InterruptedException {
final String clientID = NettyUtils.clientID(channel);
LOG.info("Processing DISCONNECT message. MqttClientId = {}.", clientID);
LOG.info("Processing DISCONNECT message. CId = {}.", clientID);
channel.flush();
final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID);
if (existingDescriptor == null) {
@@ -610,43 +629,43 @@ public class ProtocolProcessor {
if (existingDescriptor.doesNotUseChannel(channel)) {
//another client saved it's descriptor, exit
LOG.warn("Another client is using the connection descriptor. Closing connection. MqttClientId = {}.",
LOG.warn("Another client is using the connection descriptor. Closing connection. CId = {}.",
clientID);
existingDescriptor.abort();
return;
}
if (!removeSubscriptions(existingDescriptor, clientID)) {
LOG.warn("Unable to remove subscriptions. Closing connection. MqttClientId = {}.", clientID);
LOG.warn("Unable to remove subscriptions. Closing connection. CId = {}.", clientID);
existingDescriptor.abort();
return;
}
if (!dropStoredMessages(existingDescriptor, clientID)) {
LOG.warn("Unable to drop stored messages. Closing connection. MqttClientId = {}.", clientID);
LOG.warn("Unable to drop stored messages. Closing connection. CId = {}.", clientID);
existingDescriptor.abort();
return;
}
if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) {
LOG.warn("Unable to drop will message. Closing connection. MqttClientId = {}.", clientID);
LOG.warn("Unable to drop will message. Closing connection. CId = {}.", clientID);
existingDescriptor.abort();
return;
}
if (!existingDescriptor.close()) {
LOG.info("The connection has been closed. MqttClientId = {}.", clientID);
LOG.info("The connection has been closed. CId = {}.", clientID);
return;
}
boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor);
if (!stillPresent) {
//another descriptor was inserted
LOG.warn("Another descriptor has been inserted. MqttClientId = {}.", clientID);
LOG.warn("Another descriptor has been inserted. CId = {}.", clientID);
return;
}
LOG.info("The DISCONNECT message has been processed. MqttClientId = {}.", clientID);
LOG.info("The DISCONNECT message has been processed. CId = {}.", clientID);
}
private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) {
@@ -656,9 +675,9 @@ public class ProtocolProcessor {
}
if (descriptor.cleanSession) {
LOG.info("Removing saved subscriptions. MqttClientId = {}.", descriptor.clientID);
LOG.info("Removing saved subscriptions. CId = {}.", descriptor.clientID);
m_sessionsStore.wipeSubscriptions(clientID);
LOG.info("The saved subscriptions have been removed. MqttClientId = {}.", descriptor.clientID);
LOG.info("The saved subscriptions have been removed. CId = {}.", descriptor.clientID);
}
return true;
}
@@ -670,9 +689,9 @@ public class ProtocolProcessor {
}
if (descriptor.cleanSession) {
LOG.debug("Removing messages of session. MqttClientId = {}.", descriptor.clientID);
LOG.debug("Removing messages of session. CId = {}.", descriptor.clientID);
this.m_sessionsStore.dropQueue(clientID);
LOG.debug("The messages of the session have been removed. MqttClientId = {}.", descriptor.clientID);
LOG.debug("The messages of the session have been removed. CId = {}.", descriptor.clientID);
}
return true;
}
@@ -692,7 +711,7 @@ public class ProtocolProcessor {
}
public void processConnectionLost(String clientID, Channel channel) {
LOG.info("Processing connection lost event. MqttClientId = {}.", clientID);
LOG.info("Processing connection lost event. CId = {}.", clientID);
ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true);
connectionDescriptors.removeConnection(oldConnDescr);
//publish the Will message (if any) for the clientID
@@ -712,11 +731,11 @@ public class ProtocolProcessor {
* @param channel the channel of the incoming message.
* @param msg the decoded unsubscribe message.
*/
public void processUnsubscribe(Channel channel, UnsubscribeMessage msg) {
List<String> topics = msg.topicFilters();
public void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg) {
List<String> topics = msg.payload().topics();
String clientID = NettyUtils.clientID(channel);
LOG.info("Processing UNSUBSCRIBE message. MqttClientId = {}, topics = {}.", clientID, topics);
LOG.info("Processing UNSUBSCRIBE message. CId = {}, topics = {}.", clientID, topics);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
for (String topic : topics) {
@@ -724,12 +743,12 @@ public class ProtocolProcessor {
if (!validTopic) {
//close the connection, not valid topicFilter is a protocol violation
channel.close();
LOG.error("The topic filter is not valid. MqttClientId = {}, topics = {}, badTopicFilter = {}.",
LOG.error("The topic filter is not valid. CId = {}, topics = {}, badTopicFilter = {}.",
clientID, topics, topic);
return;
}
LOG.debug("Removing subscription. MqttClientId = {}, topic = {}.", clientID, topic);
LOG.debug("Removing subscription. CId = {}, topic = {}.", clientID, topic);
subscriptions.removeSubscription(topic, clientID);
clientSession.unsubscribeFrom(topic);
String username = NettyUtils.userName(channel);
@@ -737,40 +756,37 @@ public class ProtocolProcessor {
}
//ack the client
int messageID = msg.getMessageID();
UnsubAckMessage ackMessage = new UnsubAckMessage();
ackMessage.setMessageID(messageID);
int messageID = msg.variableHeader().messageId();
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0);
MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID));
LOG.info("Sending UNSUBACK message. MqttClientId = {}, topics = {}, messageId = {}.", clientID, topics,
messageID);
LOG.info("Sending UNSUBACK message. CId = {}, topics = {}, messageId = {}.", clientID, topics, messageID);
channel.writeAndFlush(ackMessage);
}
public void processSubscribe(Channel channel, SubscribeMessage msg) {
public void processSubscribe(Channel channel, MqttSubscribeMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
LOG.info("Processing SUBSCRIBE message. MqttClientId = {}, messageId = {}.", clientID, msg.getMessageID());
int messageID = messageId(msg);
LOG.info("Processing SUBSCRIBE message. CId = {}, messageId = {}.", clientID, messageID);
RunningSubscription executionKey = new RunningSubscription(clientID, messageID);
SubscriptionState currentStatus = this.subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED);
if (currentStatus != null) {
LOG.warn("The client sent another SUBSCRIBE message while this one was being processed. MqttClientId = {}, messageId = {}.",
clientID, msg.getMessageID());
LOG.warn("The client sent another SUBSCRIBE message while this one was being processed. " +
"CId = {}, messageId = {}.", clientID, messageID);
return;
}
String username = NettyUtils.userName(channel);
List<SubscribeMessage.Couple> ackTopics = doVerify(clientID, username, msg);
SubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics);
List<MqttTopicSubscription> ackTopics = doVerify(clientID, username, msg);
MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID);
if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) {
LOG.warn("The client sent another SUBSCRIBE message while the topic filters were being verified. MqttClientId = {}, messageId = {}.",
clientID, msg.getMessageID());
LOG.warn("The client sent another SUBSCRIBE message while the topic filters were being verified. " +
"CId = {}, messageId = {}.", clientID, messageID);
return;
}
LOG.info("Creating and storing subscriptions. MqttClientId = {}, messageId = {}, topics = {}.", clientID,
msg.getMessageID(), ackTopics);
LOG.info("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics);
ackMessage.setMessageID(messageID);
List<Subscription> newSubscriptions = doStoreSubscription(ackTopics, clientID);
//save session, persist subscriptions from session
@@ -779,7 +795,7 @@ public class ProtocolProcessor {
subscriptions.add(subscription.asClientTopicCouple());
}
LOG.info("Sending SUBACK response. MqttClientId = {}, messageId = {}.", clientID, msg.getMessageID());
LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID);
channel.writeAndFlush(ackMessage);
//fire the persisted messages in session
@@ -789,22 +805,20 @@ public class ProtocolProcessor {
boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED);
if (!success) {
LOG.warn("Unable to perform the final subscription state update. MqttClientId = {}, messageId = {}.",
clientID, msg.getMessageID());
LOG.warn("Unable to perform the final subscription state update CId={}, messageId={}", clientID, messageID);
}
}
private List<Subscription> doStoreSubscription(List<SubscribeMessage.Couple> ackTopics, String clientID) {
private List<Subscription> doStoreSubscription(List<MqttTopicSubscription> ackTopics, String clientID) {
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
List<Subscription> newSubscriptions = new ArrayList<>();
for (SubscribeMessage.Couple req : ackTopics) {
for (MqttTopicSubscription req : ackTopics) {
//TODO this is SUPER UGLY
if (req.qos == AbstractMessage.QOSType.FAILURE.byteValue()) {
if (req.qualityOfService() == FAILURE) {
continue;
}
AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.qos);
Subscription newSubscription = new Subscription(clientID, req.topicFilter, qos);
Subscription newSubscription = new Subscription(clientID, req.topicName(), req.qualityOfService());
clientSession.subscribe(newSubscription);
newSubscriptions.add(newSubscription);
}
@@ -812,32 +826,34 @@ public class ProtocolProcessor {
}
/**
* @return the list of verified topics
* @param clientID
* @param username
* @param clientID the clientID
* @param username the username
* @param msg the subscribe message to verify
* @return the list of verified topics for the given subscribe message.
* */
private List<SubscribeMessage.Couple> doVerify(String clientID, String username, SubscribeMessage msg) {
private List<MqttTopicSubscription> doVerify(String clientID, String username, MqttSubscribeMessage msg) {
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
List<SubscribeMessage.Couple> ackTopics = new ArrayList<>();
List<MqttTopicSubscription> ackTopics = new ArrayList<>();
for (SubscribeMessage.Couple req : msg.subscriptions()) {
if (!m_authorizator.canRead(req.topicFilter, username, clientSession.clientID)) {
final int messageId = messageId(msg);
for (MqttTopicSubscription req : msg.payload().topicSubscriptions()) {
if (!m_authorizator.canRead(req.topicName(), username, clientSession.clientID)) {
//send SUBACK with 0x80, the user hasn't credentials to read the topic
LOG.error("The client does not have read permissions on the topic. MqttClientId = {}, username = {}, messageId = {}, topic = {}.",
clientID, username, msg.getMessageID(), req.topicFilter);
ackTopics.add(new SubscribeMessage.Couple(AbstractMessage.QOSType.FAILURE.byteValue(), req.topicFilter));
LOG.error("The client does not have read permissions on the topic. CId = {}, username = {}, messageId = {}, topic = {}.",
clientID, username, messageId, req.topicName());
ackTopics.add(new MqttTopicSubscription(req.topicName(), FAILURE));
} else {
AbstractMessage.QOSType qos = null;
if (SubscriptionsStore.validate(req.topicFilter)) {
qos = AbstractMessage.QOSType.valueOf(req.qos);
LOG.info("The client will be subscribed to the topic. MqttClientId = {}, username = {}, messageId = {}, topic = {}.",
clientID, username, msg.getMessageID(), req.topicFilter);
MqttQoS qos;
if (SubscriptionsStore.validate(req.topicName())) {
LOG.info("The client will be subscribed to the topic. CId = {}, username = {}, messageId = {}, topic = {}.",
clientID, username, messageId, req.topicName());
qos = req.qualityOfService();
} else {
LOG.error("The topic filter is not valid. MqttClientId = {}, username = {}, messageId = {}, topic = {}.",
clientID, username, msg.getMessageID(), req.topicFilter);
qos = AbstractMessage.QOSType.FAILURE;
LOG.error("The topic filter is not valid. CId = {}, username = {}, messageId = {}, topic = {}.",
clientID, username, messageId, req.topicName());
qos = FAILURE;
}
ackTopics.add(new SubscribeMessage.Couple(qos.byteValue(), req.topicFilter));
ackTopics.add(new MqttTopicSubscription(req.topicName(), qos));
}
}
return ackTopics;
@@ -846,17 +862,19 @@ public class ProtocolProcessor {
/**
* Create the SUBACK response from a list of topicFilters
* */
private SubAckMessage doAckMessageFromValidateFilters(List<SubscribeMessage.Couple> topicFilters) {
//ack the client
SubAckMessage ackMessage = new SubAckMessage();
for (SubscribeMessage.Couple req : topicFilters) {
ackMessage.addType(AbstractMessage.QOSType.valueOf(req.qos));
private MqttSubAckMessage doAckMessageFromValidateFilters(List<MqttTopicSubscription> topicFilters, int messageId) {
List<Integer> grantedQoSLevels = new ArrayList<>();
for (MqttTopicSubscription req : topicFilters) {
grantedQoSLevels.add(req.qualityOfService().value());
}
return ackMessage;
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, AT_LEAST_ONCE, false, 0);
MqttSubAckPayload payload = new MqttSubAckPayload(grantedQoSLevels);
return new MqttSubAckMessage(fixedHeader, from(messageId), payload);
}
private void publishRetainedMessagesInSession(final Subscription newSubscription, String username) {
LOG.info("Retrieving retained messages. MqttClientId = {}, topics = {}.", newSubscription.getClientId(),
LOG.info("Retrieving retained messages. CId = {}, topics = {}.", newSubscription.getClientId(),
newSubscription.getTopicFilter());
//scans retained messages to be published to the new subscription
@@ -869,7 +887,7 @@ public class ProtocolProcessor {
});
if (!messages.isEmpty()) {
LOG.info("Publishing retained messages. MqttClientId = {}, topics = {}, messagesNo = {}.",
LOG.info("Publishing retained messages. CId = {}, topics = {}, messagesNo = {}.",
newSubscription.getClientId(), newSubscription.getTopicFilter(), messages.size());
}
ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId());
@@ -890,7 +908,7 @@ public class ProtocolProcessor {
} else {
//recreate a publish from stored publish in queue
boolean retained = m_messagesStore.getMessageByGuid(msg.getGuid()) != null;
PublishMessage pubMsg = createPublishForQos(msg.getTopic(), msg.getQos(), msg.getMessage(), retained);
MqttPublishMessage pubMsg = createPublishForQos(msg.getTopic(), msg.getQos(), msg.getMessage(), retained, 0);
channel.write(pubMsg);
}
}
broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java
+8 -6
View File
@@ -1,12 +1,12 @@
package io.moquette.spi.impl;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,9 +33,9 @@ class Qos0PublishHandler extends QosPublishHandler {
this.publisher = messagesPublisher;
}
void receivedPublishQos0(Channel channel, PublishMessage msg) {
void receivedPublishQos0(Channel channel, MqttPublishMessage msg) {
//verify if topic can be write
final String topic = msg.getTopicName();
final String topic = msg.variableHeader().topicName();
if (checkWriteOnTopic(topic, channel)) {
return;
}
@@ -45,19 +45,21 @@ class Qos0PublishHandler extends QosPublishHandler {
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
final int messageID = msg.variableHeader().messageId();
if (LOG.isTraceEnabled()) {
LOG.trace("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()),
clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID,
topic, msg.getMessageID());
topic, messageID);
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
if (msg.isRetainFlag()) {
if (msg.fixedHeader().isRetain()) {
//QoS == 0 && retain => clean old retained
m_messagesStore.cleanRetained(topic);
}
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+18 -13
View File
@@ -1,7 +1,5 @@
package io.moquette.spi.impl;
import io.moquette.parser.proto.messages.PubAckMessage;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.IMessagesStore;
@@ -10,12 +8,18 @@ import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
class Qos1PublishHandler extends QosPublishHandler {
private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class);
@@ -38,13 +42,15 @@ class Qos1PublishHandler extends QosPublishHandler {
this.publisher = messagesPublisher;
}
void receivedPublishQos1(Channel channel, PublishMessage msg) {
void receivedPublishQos1(Channel channel, MqttPublishMessage msg) {
//verify if topic can be write
final String topic = msg.getTopicName();
final String topic = msg.variableHeader().topicName();
if (checkWriteOnTopic(topic, channel)) {
return;
}
final int messageID = msg.variableHeader().messageId();
//route message to subscribers
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
String clientID = NettyUtils.clientID(channel);
@@ -52,24 +58,23 @@ class Qos1PublishHandler extends QosPublishHandler {
if (LOG.isTraceEnabled()) {
LOG.trace("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()),
clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID,
topic, msg.getMessageID());
topic, messageID);
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
//send PUBACK
final Integer messageID = msg.getMessageID();
if (msg.isLocal()) {
//TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) {
sendPubAck(clientID, messageID);
}
// }
if (msg.isRetainFlag()) {
if (!msg.getPayload().hasRemaining()) {
if (msg.fixedHeader().isRetain()) {
if (!msg.payload().isReadable()) {
m_messagesStore.cleanRetained(topic);
} else {
//before wasn't stored
@@ -84,8 +89,8 @@ class Qos1PublishHandler extends QosPublishHandler {
private void sendPubAck(String clientId, int messageID) {
LOG.trace("sendPubAck invoked");
PubAckMessage pubAckMessage = new PubAckMessage();
pubAckMessage.setMessageID(messageID);
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, AT_MOST_ONCE, false, 0);
MqttPubAckMessage pubAckMessage = new MqttPubAckMessage(fixedHeader, from(messageID));
try {
if (connectionDescriptors == null) {
broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java
+26 -23
View File
@@ -1,6 +1,5 @@
package io.moquette.spi.impl;
import io.moquette.parser.proto.messages.*;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.ClientSession;
@@ -11,13 +10,16 @@ import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
import java.util.List;
import static io.moquette.spi.impl.Utils.messageId;
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
class Qos2PublishHandler extends QosPublishHandler {
@@ -43,37 +45,38 @@ class Qos2PublishHandler extends QosPublishHandler {
this.publisher = messagesPublisher;
}
void receivedPublishQos2(Channel channel, PublishMessage msg) {
final String topic = msg.getTopicName();
void receivedPublishQos2(Channel channel, MqttPublishMessage msg) {
final String topic = msg.variableHeader().topicName();
//check if the topic can be wrote
if (checkWriteOnTopic(topic, channel)) {
return;
}
final Integer messageID = msg.getMessageID();
final int messageID = msg.variableHeader().messageId();
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
if (LOG.isTraceEnabled()) {
LOG.trace("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()),
LOG.trace("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, " +
"payload = {}, subscriptionTree = {}.",
clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID,
topic, msg.getMessageID());
LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.",
clientID, topic, messageID);
}
//QoS2
MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg);
if (msg.isLocal()) {
//TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) {
sendPubRec(clientID, messageID);
}
//}
//Next the client will send us a pub rel
//NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
if (msg.isRetainFlag()) {
if (!msg.getPayload().hasRemaining()) {
if (msg.fixedHeader().isRetain()) {
if (msg.payload().readableBytes() == 0) {
m_messagesStore.cleanRetained(topic);
} else {
m_messagesStore.storeRetained(topic, guid);
@@ -87,9 +90,9 @@ class Qos2PublishHandler extends QosPublishHandler {
* Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish
* to all interested subscribers.
* */
void processPubRel(Channel channel, PubRelMessage msg) {
void processPubRel(Channel channel, MqttMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
int messageID = messageId(msg);
LOG.info("Processing PUBREL message. MqttClientId = {}, messageId = {}.", clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID);
@@ -98,16 +101,16 @@ class Qos2PublishHandler extends QosPublishHandler {
if (LOG.isDebugEnabled()) {
LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(evt.getMessage()),
clientID, topic, messageID, DebugUtils.payload2Str(evt.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID,
topic, msg.getMessageID());
topic, messageID);
}
this.publisher.publish2Subscribers(evt, topicMatchingSubscriptions);
if (evt.isRetained()) {
if (!evt.getMessage().hasRemaining()) {
if (evt.getMessage().readableBytes() == 0) {
m_messagesStore.cleanRetained(topic);
} else {
m_messagesStore.storeRetained(topic, evt.getGuid());
@@ -120,15 +123,15 @@ class Qos2PublishHandler extends QosPublishHandler {
private void sendPubRec(String clientID, int messageID) {
LOG.debug("Sending PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID);
PubRecMessage pubRecMessage = new PubRecMessage();
pubRecMessage.setMessageID(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. MqttClientId = {}, messageId = {}.", clientID, messageID);
PubCompMessage pubCompMessage = new PubCompMessage();
pubCompMessage.setMessageID(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);
}
}
broker/src/main/java/io/moquette/spi/impl/Utils.java
+17
View File
@@ -1,11 +1,16 @@
package io.moquette.spi.impl;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import java.util.Map;
/**
* Utility static methods, like Map get with default value, or elvis operator.
*/
public class Utils {
public static <T, K> T defaultGet(Map<K, T> map, K key, T defaultValue) {
T value = map.get(key);
if (value != null) {
@@ -13,4 +18,16 @@ public class Utils {
}
return defaultValue;
}
public static int messageId(MqttMessage msg) {
return ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
}
public static byte[] readBytesAndRewind(ByteBuf payload) {
byte[] payloadContent = new byte[payload.readableBytes()];
int mark = payload.readerIndex();
payload.readBytes(payloadContent);
payload.readerIndex(mark);
return payloadContent;
}
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java
+8 -6
View File
@@ -15,9 +15,10 @@
*/
package io.moquette.spi.impl.subscriptions;
import java.io.Serializable;
import io.moquette.parser.proto.messages.AbstractMessage.QOSType;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.io.Serializable;
/**
* Maintain the information about which Topic a certain ClientID is subscribed
@@ -29,12 +30,12 @@ import io.moquette.spi.ISessionsStore.ClientTopicCouple;
public final class Subscription implements Serializable {
private static final long serialVersionUID = -3383457629635732794L;
final QOSType requestedQos; //max QoS acceptable
final MqttQoS requestedQos; //max QoS acceptable
final String clientId;
final String topicFilter;
final boolean active;
public Subscription(String clientId, String topicFilter, QOSType requestedQos) {
public Subscription(String clientId, String topicFilter, MqttQoS requestedQos) {
this.requestedQos = requestedQos;
this.clientId = clientId;
this.topicFilter = topicFilter;
@@ -52,7 +53,7 @@ public final class Subscription implements Serializable {
return clientId;
}
public QOSType getRequestedQos() {
public MqttQoS getRequestedQos() {
return requestedQos;
}
@@ -85,7 +86,8 @@ public final class Subscription implements Serializable {
@Override
public String toString() {
return String.format("[filter:%s, cliID: %s, qos: %s, active: %s]", this.topicFilter, this.clientId, this.requestedQos, this.active);
return String.format("[filter:%s, cliID: %s, qos: %s, active: %s]", this.topicFilter, this.clientId,
this.requestedQos, this.active);
}
@Override
broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java
+6 -6
View File
@@ -15,16 +15,16 @@
*/
package io.moquette.spi.impl.subscriptions;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicReference;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicReference;
/**
* Represents a tree of topics subscriptions.
*
@@ -235,7 +235,7 @@ public class SubscriptionsStore {
continue;
}
//update the selected subscriptions if not present or if has a greater qos
if (existingSub == null || existingSub.getRequestedQos().byteValue() < sub.getRequestedQos().byteValue()) {
if (existingSub == null || existingSub.getRequestedQos().value() < sub.getRequestedQos().value()) {
subsForClient.put(matchingCouple.clientID, sub);
}
}
broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java
+2 -3
View File
@@ -19,7 +19,6 @@ package io.moquette.spi.persistence;
import io.moquette.server.config.IConfig;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.parser.proto.MQTTException;
import org.mapdb.DB;
import org.mapdb.DBMaker;
import org.slf4j.Logger;
@@ -32,8 +31,8 @@ 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 io.moquette.BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME;
import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME;
/**
* MapDB main persistence implementation
@@ -94,7 +93,7 @@ public class MapDBPersistentStore {
LOG.warn("Using {} MapDB store file. Path = {}.", fileNewlyCreated ? "fresh" : "existing", m_storePath);
} catch (IOException ex) {
LOG.error("Unable to open MapDB store file. Path = {}, cause = {}, errorMessage = {}.", m_storePath, ex.getCause(), ex.getMessage());
throw new MQTTException("Can't create temp file for subscriptions storage [" + m_storePath + "]", ex);
throw new RuntimeException("Can't create temp file for subscriptions storage [" + m_storePath + "]", ex);
}
m_db = DBMaker.newFileDB(tmpFile).make();
}
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+27 -30
View File
@@ -66,13 +66,13 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public void addNewSubscription(Subscription newSubscription) {
LOG.info("Adding new subscription. MqttClientId = {}, topics = {}.", newSubscription.getClientId(),
LOG.info("Adding new subscription. CId= {}, topics = {}.", newSubscription.getClientId(),
newSubscription.getTopicFilter());
final String clientID = newSubscription.getClientId();
m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription);
if (LOG.isTraceEnabled()) {
LOG.trace("The subscription has been added. MqttClientId = {}, topics = {}, clientSubscriptions = {}.",
LOG.trace("The subscription has been added. CId= {}, topics = {}, clientSubscriptions = {}.",
newSubscription.getClientId(), newSubscription.getTopicFilter(),
m_db.getHashMap("subscriptions_" + clientID));
}
@@ -80,23 +80,23 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public void removeSubscription(String topicFilter, String clientID) {
LOG.info("Removing subscription. MqttClientId = {}, topics = {}.", clientID, topicFilter);
LOG.info("Removing subscription. CId= {}, topics = {}.", clientID, topicFilter);
if (!m_db.exists("subscriptions_" + clientID)) {
return;
}
m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter);
if (LOG.isDebugEnabled()) {
LOG.debug("The subscription has been removed. MqttClientId = {}, topics = {}, clientSubscriptions = {}.",
LOG.debug("The subscription has been removed. CId= {}, topics = {}, clientSubscriptions = {}.",
clientID, topicFilter, m_db.getHashMap("subscriptions_" + clientID));
}
}
@Override
public void wipeSubscriptions(String clientID) {
LOG.info("Wiping subscriptions. MqttClientId = {}.", clientID);
LOG.info("Wiping subscriptions. CId= {}.", clientID);
m_db.delete("subscriptions_" + clientID);
if (LOG.isDebugEnabled()) {
LOG.debug("The subscriptions have been removed. MqttClientId = {}, clientSubscriptions = {}.", clientID,
LOG.debug("The subscriptions have been removed. CId= {}, clientSubscriptions = {}.", clientID,
m_db.getHashMap("subscriptions_" + clientID));
}
}
@@ -120,7 +120,7 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public Subscription getSubscription(ClientTopicCouple couple) {
ConcurrentMap<String, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID);
LOG.info("Retrieving subscriptions. MqttClientId = {}, subscriptions = {}.", couple.clientID,
LOG.info("Retrieving subscriptions. CId= {}, subscriptions = {}.", couple.clientID,
clientSubscriptions);
return clientSubscriptions.get(couple.topicFilter);
}
@@ -146,20 +146,20 @@ class MapDBSessionsStore implements ISessionsStore {
public ClientSession createNewSession(String clientID, boolean cleanSession) {
if (m_persistentSessions.containsKey(clientID)) {
LOG.error(
"Unable to create a new session: the client ID is already in use. MqttClientId = {}, cleanSession = {}.",
"Unable to create a new session: the client ID is already in use. CId= {}, cleanSession = {}.",
clientID, cleanSession);
throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
}
LOG.info("Creating new session. MqttClientId = {}, cleanSession = {}.", clientID, cleanSession);
LOG.info("Creating new session. CId= {}, cleanSession = {}.", clientID, cleanSession);
m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession));
return new ClientSession(clientID, m_messagesStore, this, cleanSession);
}
@Override
public ClientSession sessionForClient(String clientID) {
LOG.info("Retrieving session. MqttClientId = {}.", clientID);
LOG.info("Retrieving session. CId= {}.", clientID);
if (!m_persistentSessions.containsKey(clientID)) {
LOG.warn("The session does not exist. MqttClientId = {}.", clientID);
LOG.warn("The session does not exist. CId= {}.", clientID);
return null;
}
@@ -178,7 +178,7 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public void updateCleanStatus(String clientID, boolean cleanSession) {
LOG.info("Updating cleanSession flag. MqttClientId = {}, cleanSession = {}.", clientID, cleanSession);
LOG.info("Updating cleanSession flag. CId= {}, cleanSession = {}.", clientID, cleanSession);
m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
}
@@ -188,7 +188,7 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public int nextPacketID(String clientID) {
if (LOG.isDebugEnabled()) {
LOG.debug("Generating next packet ID. MqttClientId = {}.", clientID);
LOG.debug("Generating next packet ID. CId= {}.", clientID);
}
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient == null) {
@@ -204,17 +204,17 @@ class MapDBSessionsStore implements ISessionsStore {
int nextPacketId = (maxId + 1) % 0xFFFF;
inFlightForClient.add(nextPacketId);
if (LOG.isDebugEnabled()) {
LOG.debug("The next packet ID has been generated. MqttClientId = {}, result = {}.", clientID, nextPacketId);
LOG.debug("The next packet ID has been generated. CId= {}, result = {}.", clientID, nextPacketId);
}
return nextPacketId;
}
@Override
public void inFlightAck(String clientID, int messageID) {
LOG.debug("Acknowledging inflight message. MqttClientId = {}, messageId = {}.", clientID, messageID);
LOG.debug("Acknowledging inflight message. CId= {}, messageId = {}.", clientID, messageID);
Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
if (m == null) {
LOG.warn("Unable to retrieve inflight message record. MqttClientId = {}, messageId = {}.", clientID, messageID);
LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID);
return;
}
m.remove(messageID);
@@ -228,7 +228,7 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public void inFlight(String clientID, int messageID, MessageGUID guid) {
LOG.debug("Storing inflight message. MqttClientId = {}, messageId = {}, guid = {}.", clientID, messageID, guid);
LOG.debug("Storing inflight message. CId= {}, messageId = {}, guid = {}.", clientID, messageID, guid);
ConcurrentMap<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
if (m == null) {
m = new ConcurrentHashMap<>();
@@ -240,26 +240,24 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public BlockingQueue<StoredMessage> queue(String clientID) {
LOG.info("Queuing pending message. MqttClientId = {}, guid = {}.", clientID);
LOG.info("Queuing pending message. CId= {}, guid = {}.", clientID);
return this.m_db.getQueue(clientID);
}
@Override
public void dropQueue(String clientID) {
LOG.info("Removing pending messages. MqttClientId = {}.", clientID);
LOG.info("Removing pending messages. CId= {}.", clientID);
this.m_db.delete(clientID);
}
@Override
public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) {
if (LOG.isDebugEnabled()) {
LOG.debug("Moving inflight message to second phase ack state. MqttClientId = {}, messageID = {}.", clientID,
messageID);
LOG.debug("Moving inflight message to 2nd phase ack state. CId= {}, messageID = {}.", clientID, messageID);
}
Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
if (m == null) {
LOG.warn("Unable to retrieve inflight message record. MqttClientId = {}, messageId = {}.", clientID,
messageID);
LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID);
return;
}
MessageGUID guid = m.remove(messageID);
@@ -278,7 +276,7 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) {
if (LOG.isDebugEnabled()) {
LOG.debug("Processing second phase ACK. MqttClientId = {}, messageId = {}.", clientID, messageID);
LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID);
}
Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
MessageGUID guid = messageIDs.remove(messageID);
@@ -288,11 +286,10 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public MessageGUID mapToGuid(String clientID, int messageID) {
LOG.debug("Mapping message ID to GUID. MqttClientId = {}, messageId = {}.", clientID, messageID);
LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID);
ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
MessageGUID result = messageIdToGuid.get(messageID);
LOG.debug("The message ID has been mapped to a GUID. MqttClientId = {}, messageId = {}, guid = {}.",
clientID, messageID, result);
LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result);
return result;
}
@@ -302,15 +299,15 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public StoredMessage getInflightMessage(String clientID, int messageID) {
LOG.info("Retrieving inflight message. MqttClientId = {}, messageId = {}.", clientID, messageID);
LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID);
Map<Integer, MessageGUID> clientEntries = m_inflightStore.get(clientID);
if (clientEntries == null) {
LOG.warn("The client has no inflight messages. MqttClientId = {}, messageId = {}.", clientID, messageID);
LOG.warn("The client has no inflight messages CId={}, messageId={}", clientID, messageID);
return null;
}
MessageGUID guid = clientEntries.get(messageID);
if (guid == null) {
LOG.warn("The message ID does not have an associated GUID. MqttClientId = {}, messageId = {}.", clientID,
LOG.warn("The message ID does not have an associated GUID. CId= {}, messageId = {}.", clientID,
messageID);
return null;
}