Merge branch Sofia2 with lot of improvements

This commit is contained in:
andsel
2017-01-17 21:40:59 +01:00
62 changed files with 1608 additions and 502 deletions
@@ -22,6 +22,7 @@ import java.io.File;
*/
public class BrokerConstants {
public static final String INTERCEPT_HANDLER_PROPERTY_NAME ="intercept.handler";
public static final String BROKER_INTERCEPTOR_THREAD_POOL_SIZE = "intercept.thread_pool.size";
public static final String PERSISTENT_STORE_PROPERTY_NAME = "persistent_store";
public static final String AUTOSAVE_INTERVAL_PROPERTY_NAME = "autosave_interval";
public static final String PASSWORD_FILE_PROPERTY_NAME = "password_file";
@@ -50,4 +51,9 @@ public class BrokerConstants {
public static final String HOST = "0.0.0.0";
public static final String NEED_CLIENT_AUTH = "need_client_auth";
public static final String HAZELCAST_CONFIGURATION = "hazelcast.configuration";
public static final String NETTY_SO_BACKLOG_PROPERTY_NAME = "netty.so_backlog";
public static final String NETTY_SO_REUSEADDR_PROPERTY_NAME = "netty.so_reuseaddr";
public static final String NETTY_TCP_NODELAY_PROPERTY_NAME = "netty.tcp_nodelay";
public static final String NETTY_SO_KEEPALIVE_PROPERTY_NAME = "netty.so_keepalive";
public static final String NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME = "netty.channel_timeout.seconds";
}
@@ -0,0 +1,58 @@
package io.moquette.connections;
import java.util.Collection;
/**
* This interface will be used by an external codebase to retrieve and close
* physical connections.
*
* @author lbarrios
*
*/
public interface IConnectionsManager {
/**
* Returns the number of physical connections
*
* @return
*/
public int getActiveConnectionsNo();
/**
* Determines wether a MQTT client is connected to the broker.
*
* @param clientID
* @return
*/
public boolean isConnected(String clientID);
/**
* Returns the identifiers of the MQTT clients that are connected to the
* broker.
*
* @return
*/
public Collection<String> getConnectedClientIds();
/**
* Closes a physical connection.
*
* @param clientID
* @param closeImmediately
* If false, the connection will be flushed before it is closed.
* @return
*/
public boolean closeConnection(String clientID, boolean closeImmediately);
/**
* Returns the state of the session of a given client.
* @param clientID
* @return
*/
public MqttSession getSessionStatus(String clientID);
/**
* Returns the state of all the sessions
* @return
*/
public Collection<MqttSession> getSessions();
}
@@ -0,0 +1,38 @@
package io.moquette.connections;
/**
* A class that represents the metrics of a given MQTT connection.
*
* @author lbarrios
*
*/
public class MqttConnectionMetrics {
private long readKb;
private long writtenKb;
private long readMessages;
private long writtenMessages;
public MqttConnectionMetrics(long readBytes, long writtenBytes, long readMessages, long writtenMessages) {
this.readKb = readBytes / 1024;
this.writtenKb = writtenBytes / 1024;
this.readMessages = readMessages;
this.writtenMessages = writtenMessages;
}
public long getReadKb() {
return readKb;
}
public long getWrittenKb() {
return writtenKb;
}
public long getReadMessages() {
return readMessages;
}
public long getWrittenMessages() {
return writtenMessages;
}
}
@@ -0,0 +1,79 @@
package io.moquette.connections;
import java.util.Collection;
/**
* A class that represents the overall connection status of a MQTT session. Its
* instances will be used by an external codebase when the broker is configured
* in embedded mode.
*
* @author lbarrios
*
*/
public class MqttSession {
private boolean connectionEstablished;
private boolean cleanSession;
private int inflightMessages;
private int pendingPublishMessagesNo;
private int secondPhaseAckPendingMessages;
private Collection<MqttSubscription> activeSubscriptions;
private MqttConnectionMetrics connectionMetrics;
public boolean isConnectionEstablished() {
return connectionEstablished;
}
public void setConnectionEstablished(boolean connectionEstablished) {
this.connectionEstablished = connectionEstablished;
}
public boolean isCleanSession() {
return cleanSession;
}
public void setCleanSession(boolean cleanSession) {
this.cleanSession = cleanSession;
}
public int getPendingPublishMessagesNo() {
return pendingPublishMessagesNo;
}
public void setPendingPublishMessagesNo(int pendingPublishMessagesNo) {
this.pendingPublishMessagesNo = pendingPublishMessagesNo;
}
public int getInflightMessages() {
return inflightMessages;
}
public void setInflightMessages(int inflightMessages) {
this.inflightMessages = inflightMessages;
}
public int getSecondPhaseAckPendingMessages() {
return secondPhaseAckPendingMessages;
}
public void setSecondPhaseAckPendingMessages(int secondPhaseAckPendingMessages) {
this.secondPhaseAckPendingMessages = secondPhaseAckPendingMessages;
}
public Collection<MqttSubscription> getActiveSubscriptions() {
return activeSubscriptions;
}
public void setActiveSubscriptions(Collection<MqttSubscription> activeSubscriptions) {
this.activeSubscriptions = activeSubscriptions;
}
public MqttConnectionMetrics getConnectionMetrics() {
return connectionMetrics;
}
public void setConnectionMetrics(MqttConnectionMetrics connectionMetrics) {
this.connectionMetrics = connectionMetrics;
}
}
@@ -0,0 +1,38 @@
package io.moquette.connections;
/**
* A class that represents a MQTT subscription.
*
* @author lbarrios
*
*/
public class MqttSubscription {
private final String requestedQos;
private final String clientId;
private final String topicFilter;
private final boolean active;
public MqttSubscription(String requestedQos, String clientId, String topicFilter, boolean active) {
this.requestedQos = requestedQos;
this.clientId = clientId;
this.topicFilter = topicFilter;
this.active = active;
}
public String getRequestedQos() {
return requestedQos;
}
public String getClientId() {
return clientId;
}
public String getTopicFilter() {
return topicFilter;
}
public boolean isActive() {
return active;
}
}
@@ -15,7 +15,13 @@
*/
package io.moquette.interception;
import io.moquette.interception.messages.*;
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;
/**
* Basic abstract class usefull to avoid empty methods creation in subclasses.
@@ -23,7 +29,12 @@ import io.moquette.interception.messages.*;
* Created by andrea on 08/12/15.
*/
public abstract class AbstractInterceptHandler implements InterceptHandler {
@Override
public Class<?>[] getInterceptedMessageTypes() {
return InterceptHandler.ALL_MESSAGE_TYPES;
}
@Override
public void onConnect(InterceptConnectMessage msg) {}
@@ -18,6 +18,11 @@ public class HazelcastInterceptHandler extends AbstractInterceptHandler {
public HazelcastInterceptHandler(Server server){
this.hz = server.getHazelcastInstance();
}
@Override
public String getID() {
return HazelcastInterceptHandler.class.getName() + "@" + hz.getName();
}
@Override
public void onPublish(InterceptPublishMessage msg) {
@@ -9,7 +9,8 @@ import java.io.Serializable;
*/
public class HazelcastMsg implements Serializable {
private final String clientId;
private static final long serialVersionUID = 1L;
private final String clientId;
private final byte qos;
private final byte[] payload;
private final String topic;
@@ -15,7 +15,13 @@
*/
package io.moquette.interception;
import io.moquette.interception.messages.*;
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.spi.impl.subscriptions.Subscription;
@@ -32,7 +38,24 @@ import io.moquette.spi.impl.subscriptions.Subscription;
* @author Wagner Macedo
*/
public interface InterceptHandler {
public static final Class<?> ALL_MESSAGE_TYPES[] = { InterceptConnectMessage.class,
InterceptDisconnectMessage.class, InterceptConnectionLostMessage.class, InterceptPublishMessage.class,
InterceptSubscribeMessage.class, InterceptUnsubscribeMessage.class, InterceptAcknowledgedMessage.class };
/**
* Returns the identifier of this intercept handler.
* @return
*/
String getID();
/**
* Returns the InterceptMessage subtypes that this handler can process. If the result is null or equal to ALL_MESSAGE_TYPES,
* all the message types will be processed.
* @return
*/
Class<?>[] getInterceptedMessageTypes();
void onConnect(InterceptConnectMessage msg);
void onDisconnect(InterceptDisconnectMessage msg);
@@ -49,7 +49,7 @@ public interface Interceptor {
void notifyMessageAcknowledged(InterceptAcknowledgedMessage msg);
boolean addInterceptHandler(InterceptHandler interceptHandler);
void addInterceptHandler(InterceptHandler interceptHandler);
boolean removeInterceptHandler(InterceptHandler interceptHandler);
void removeInterceptHandler(InterceptHandler interceptHandler);
}
@@ -20,7 +20,7 @@ import io.moquette.parser.proto.messages.AbstractMessage;
/**
* @author Wagner Macedo
*/
public abstract class InterceptAbstractMessage {
public abstract class InterceptAbstractMessage implements InterceptMessage {
private final AbstractMessage msg;
InterceptAbstractMessage(AbstractMessage msg) {
@@ -2,7 +2,7 @@ package io.moquette.interception.messages;
import static io.moquette.spi.IMessagesStore.StoredMessage;
public class InterceptAcknowledgedMessage {
public class InterceptAcknowledgedMessage implements InterceptMessage {
final private StoredMessage msg;
private final String username;
private final String topic;
@@ -3,7 +3,7 @@ package io.moquette.interception.messages;
/**
* @author Wagner Macedo
*/
public class InterceptConnectionLostMessage {
public class InterceptConnectionLostMessage implements InterceptMessage {
private final String clientID;
private final String username;
@@ -3,7 +3,7 @@ package io.moquette.interception.messages;
/**
* @author Wagner Macedo
*/
public class InterceptDisconnectMessage {
public class InterceptDisconnectMessage implements InterceptMessage {
private final String clientID;
private final String username;
@@ -0,0 +1,10 @@
package io.moquette.interception.messages;
/**
* An interface that sets the root of the interceptor messages type hierarchy.
* @author lbarrios
*
*/
public interface InterceptMessage {
}
@@ -6,7 +6,7 @@ import io.moquette.spi.impl.subscriptions.Subscription;
/**
* @author Wagner Macedo
*/
public class InterceptSubscribeMessage {
public class InterceptSubscribeMessage implements InterceptMessage {
private final Subscription subscription;
private final String username;
@@ -3,7 +3,7 @@ package io.moquette.interception.messages;
/**
* @author Wagner Macedo
*/
public class InterceptUnsubscribeMessage {
public class InterceptUnsubscribeMessage implements InterceptMessage {
private final String topicFilter;
private final String clientID;
private final String username;
@@ -0,0 +1,16 @@
package io.moquette.logging;
import java.util.ArrayList;
import java.util.Collection;
import io.moquette.interception.InterceptHandler;
public class LoggingUtils {
public static <T extends InterceptHandler> Collection<String> getInterceptorIds(Collection<T> handlers) {
Collection<String> result = new ArrayList<>(handlers.size());
for (T handler : handlers) {
result.add(handler.getID());
}
return result;
}
}
@@ -15,11 +15,21 @@
*/
package io.moquette.server;
import io.netty.channel.Channel;
import java.util.NoSuchElementException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicReference;
import io.moquette.server.netty.AutoFlushHandler;
import io.moquette.server.netty.NettyUtils;
import io.moquette.server.netty.metrics.BytesMetrics;
import io.moquette.server.netty.metrics.BytesMetricsHandler;
import io.moquette.server.netty.metrics.MessageMetrics;
import io.moquette.server.netty.metrics.MessageMetricsHandler;
import io.netty.channel.Channel;
/**
* Value object to maintain the information of single connection, like ClientID, Channel,
@@ -40,7 +50,7 @@ public class ConnectionDescriptor {
}
public final String clientID;
public final Channel channel;
private final Channel channel;
public final boolean cleanSession;
private final AtomicReference<ConnectionState> channelState = new AtomicReference<>(ConnectionState.DISCONNECTED);
@@ -49,20 +59,59 @@ public class ConnectionDescriptor {
this.channel = session;
this.cleanSession = cleanSession;
}
public void writeAndFlush(Object payload) {
this.channel.writeAndFlush(payload);
}
public void setupAutoFlusher(int flushIntervalMs) {
try {
this.channel.pipeline().addAfter("idleEventHandler", "autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
} catch (NoSuchElementException nseex) {
//the idleEventHandler is not present on the pipeline
this.channel.pipeline().addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
}
}
public boolean doesNotUseChannel(Channel channel) {
return !(this.channel.equals(channel));
}
public boolean close() {
LOG.info("Closing connection descriptor. MqttClientId = {}.", clientID);
final boolean success = assignState(ConnectionState.INTERCEPTORS_NOTIFIED, ConnectionState.DISCONNECTED);
if (!success) {
return false;
}
this.channel.close();
return true;
}
public String getUsername() {
return NettyUtils.userName(this.channel);
}
public void abort() {
LOG.info("closing the channel");
LOG.info("Closing connection descriptor. MqttClientId = {}.", clientID);
// try {
//this.channel.disconnect().sync();
this.channel.close();//.sync();
this.channel.close();//.sync();
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
}
public boolean assignState(ConnectionState expected, ConnectionState newState) {
return channelState.compareAndSet(expected, newState);
}
LOG.debug("Updating state of connection descriptor. MqttClientId = {}, expectedState = {}, newState = {}.",
clientID, expected, newState);
boolean retval = channelState.compareAndSet(expected, newState);
if (!retval) {
LOG.error(
"Unable to update state of connection descriptor. MqttclientId = {}, expectedState = {}, newState = {}.",
clientID, expected, newState);
}
return retval;
}
@Override
public String toString() {
@@ -84,6 +133,14 @@ public class ConnectionDescriptor {
}
public BytesMetrics getBytesMetrics() {
return BytesMetricsHandler.getBytesMetrics(channel);
}
public MessageMetrics getMessageMetrics() {
return MessageMetricsHandler.getMessageMetrics(channel);
}
@Override
public int hashCode() {
int result = clientID != null ? clientID.hashCode() : 0;
@@ -0,0 +1,173 @@
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;
public class ConnectionDescriptorStore implements IConnectionsManager {
private static final Logger LOG = LoggerFactory.getLogger(ConnectionDescriptorStore.class);
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
private final ISessionsStore sessionsStore;
public ConnectionDescriptorStore(ISessionsStore sessionsStore) {
this.connectionDescriptors = new ConcurrentHashMap<String, ConnectionDescriptor>();
this.sessionsStore = sessionsStore;
}
public boolean sendMessage(AbstractMessage message, String clientId) {
return sendMessage(message, null, clientId);
}
public boolean sendMessage(AbstractMessage message, Integer messageID, String clientID) {
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);
} else {
LOG.debug("Sending {} message. MqttClientId = {}.", message.getMessageType(), 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);
} else {
LOG.error("The client has just disconnected. The {} could not be sent. MqttClientId = {}.",
message.getMessageType(), clientID);
}
/*
* If the client hast just disconnected, its connection
* descriptor will be null. We don't have to make the broker
* crash: we'll just discard the PUBACK message.
*/
return false;
}
descriptor.writeAndFlush(message);
return true;
} catch (Throwable e) {
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());
}
return false;
}
}
public ConnectionDescriptor addConnection(ConnectionDescriptor descriptor) {
return connectionDescriptors.putIfAbsent(descriptor.clientID, descriptor);
}
public boolean removeConnection(ConnectionDescriptor descriptor) {
return connectionDescriptors.remove(descriptor.clientID, descriptor);
}
public ConnectionDescriptor getConnection(String clientID) {
return connectionDescriptors.get(clientID);
}
@Override
public boolean isConnected(String clientID) {
return connectionDescriptors.containsKey(clientID);
}
@Override
public int getActiveConnectionsNo() {
return connectionDescriptors.size();
}
@Override
public Collection<String> getConnectedClientIds() {
return connectionDescriptors.keySet();
}
@Override
public boolean closeConnection(String clientID, boolean closeImmediately) {
ConnectionDescriptor descriptor = connectionDescriptors.get(clientID);
if (descriptor == null) {
LOG.error(
"The connection descriptor doesn't exist. The MQTT connection cannot be closed. MqttClientId = {}, closeImmediately = {}.",
clientID, closeImmediately);
return false;
}
if (closeImmediately) {
descriptor.abort();
return true;
} else {
return descriptor.close();
}
}
@Override
public MqttSession getSessionStatus(String clientID) {
LOG.info("Retrieving status of session. MqttClientId = {}.", clientID);
ClientSession session = sessionsStore.sessionForClient(clientID);
if (session == null) {
LOG.error("The given MQTT client ID doesn't have an associated session. MqttClientId = {}.", clientID);
return null;
}
return buildMqttSession(session);
}
@Override
public Collection<MqttSession> getSessions() {
LOG.info("Retrieving status of all sessions.");
Collection<MqttSession> result = new ArrayList<MqttSession>();
for (ClientSession session : sessionsStore.getAllSessions()) {
result.add(buildMqttSession(session));
}
return result;
}
private MqttSession buildMqttSession(ClientSession session) {
MqttSession result = new MqttSession();
Collection<MqttSubscription> mqttSubscriptions = new ArrayList<>();
for (Subscription subscription : session.getSubscriptions()) {
mqttSubscriptions.add(new MqttSubscription(subscription.getRequestedQos().toString(),
subscription.getClientId(), subscription.getTopicFilter(), subscription.isActive()));
}
result.setActiveSubscriptions(mqttSubscriptions);
result.setCleanSession(session.isCleanSession());
ConnectionDescriptor descriptor = this.getConnection(session.clientID);
if (descriptor != null) {
result.setConnectionEstablished(true);
BytesMetrics bytesMetrics = descriptor.getBytesMetrics();
MessageMetrics messageMetrics = descriptor.getMessageMetrics();
result.setConnectionMetrics(new MqttConnectionMetrics(bytesMetrics.readBytes(), bytesMetrics.wroteBytes(),
messageMetrics.messagesRead(), messageMetrics.messagesWrote()));
} else {
result.setConnectionEstablished(false);
}
result.setPendingPublishMessagesNo(session.getPendingPublishMessagesNo());
result.setSecondPhaseAckPendingMessages(session.getSecondPhaseAckPendingMessages());
result.setInflightMessages(session.getInflightMessagesNo());
LOG.info("The status of the session has been retrieved successfully. MqttClientId = {}.", session.clientID);
return result;
}
}
@@ -49,11 +49,12 @@ class DefaultMoquetteSslContextCreator implements ISslContextCreator {
@Override
public SSLContext initSSLContext() {
LOG.info("Checking SSL configuration properties...");
final String jksPath = props.getProperty(BrokerConstants.JKS_PATH_PROPERTY_NAME);
LOG.info("Starting SSL using keystore at {}", jksPath);
LOG.info("Initializing SSL context. KeystorePath = {}.", jksPath);
if (jksPath == null || jksPath.isEmpty()) {
//key_store_password or key_manager_password are empty
LOG.warn("You have configured the SSL port but not the jks_path, SSL not started");
LOG.warn("The keystore path is null or empty. The SSL context won't be initialized.");
return null;
}
@@ -63,12 +64,12 @@ class DefaultMoquetteSslContextCreator implements ISslContextCreator {
final String keyManagerPassword = props.getProperty(BrokerConstants.KEY_MANAGER_PASSWORD_PROPERTY_NAME);
if (keyStorePassword == null || keyStorePassword.isEmpty()) {
//key_store_password or key_manager_password are empty
LOG.warn("You have configured the SSL port but not the key_store_password, SSL not started");
LOG.warn("The keystore password is null or empty. The SSL context won't be initialized.");
return null;
}
if (keyManagerPassword == null || keyManagerPassword.isEmpty()) {
//key_manager_password or key_manager_password are empty
LOG.warn("You have configured the SSL port but not the key_manager_password, SSL not started");
LOG.warn("The key manager password is null or empty. The SSL context won't be initialized.");
return null;
}
@@ -78,26 +79,33 @@ class DefaultMoquetteSslContextCreator implements ISslContextCreator {
boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
try {
LOG.info("Loading keystore. KeystorePath = {}.", jksPath);
InputStream jksInputStream = jksDatastore(jksPath);
SSLContext serverContext = SSLContext.getInstance("TLS");
final KeyStore ks = KeyStore.getInstance("JKS");
ks.load(jksInputStream, keyStorePassword.toCharArray());
LOG.info("Initializing key manager...");
final KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
kmf.init(ks, keyManagerPassword.toCharArray());
TrustManager[] trustManagers = null;
if (needsClientAuth) {
LOG.warn(
"Client authentication is enabled. The keystore will be used as a truststore. KeystorePath = {}.",
jksPath);
// use keystore as truststore, as server needs to trust certificates signed by the server certificates
TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(ks);
trustManagers = tmf.getTrustManagers();
}
// init sslContext
LOG.info("Initializing SSL context...");
serverContext.init(kmf.getKeyManagers(), trustManagers, null);
LOG.info("The SSL context has been initialized successfully.");
return serverContext;
} catch (NoSuchAlgorithmException | UnrecoverableKeyException | CertificateException | KeyStoreException
| KeyManagementException | IOException ex) {
LOG.error("Can't start SSL layer!", ex);
LOG.error("Unable to initialize SSL context. Cause = {}, errorMessage = {}.", ex.getCause(), ex.getMessage());
return null;
}
}
@@ -108,13 +116,13 @@ class DefaultMoquetteSslContextCreator implements ISslContextCreator {
LOG.info("Starting with jks at {}, jks normal {}", jksUrl.toExternalForm(), jksUrl);
return getClass().getClassLoader().getResourceAsStream(jksPath);
}
LOG.info("jks not found in bundled resources, try on the filesystem");
LOG.warn("No keystore has been found in the bundled resources. Scanning filesystem...");
File jksFile = new File(jksPath);
if (jksFile.exists()) {
LOG.info("Using {} ", jksFile.getAbsolutePath());
LOG.info("Loading external keystore. Url = {}.", jksFile.getAbsolutePath());
return new FileInputStream(jksFile);
}
LOG.warn("File {} doesn't exists", jksFile.getAbsolutePath());
LOG.warn("The keystore file does not exist. Url = {}.", jksFile.getAbsolutePath());
return null;
}
}
@@ -23,6 +23,7 @@ import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.HazelcastInstanceNotActiveException;
import com.hazelcast.core.ITopic;
import io.moquette.BrokerConstants;
import io.moquette.connections.IConnectionsManager;
import io.moquette.interception.HazelcastInterceptHandler;
import io.moquette.interception.HazelcastMsg;
import io.moquette.interception.InterceptHandler;
@@ -45,6 +46,8 @@ import java.util.Collections;
import java.util.List;
import java.util.Properties;
import static io.moquette.logging.LoggingUtils.getInterceptorIds;
/**
* Launch a configured version of the server.
* @author andrea
@@ -84,7 +87,9 @@ public class Server {
* @throws IOException in case of any IO error.
*/
public void startServer() throws IOException {
IResourceLoader filesystemLoader = new FileResourceLoader(defaultConfigFile());
File defaultConfigurationFile = defaultConfigFile();
LOG.info("Starting Moquette server. Configuration file path = {}.", defaultConfigurationFile.getAbsolutePath());
IResourceLoader filesystemLoader = new FileResourceLoader(defaultConfigurationFile);
final IConfig config = new ResourceLoaderConfig(filesystemLoader);
startServer(config);
}
@@ -100,7 +105,7 @@ public class Server {
* @throws IOException in case of any IO Error.
*/
public void startServer(File configFile) throws IOException {
LOG.info("Using m_config file: " + configFile.getAbsolutePath());
LOG.info("Starting Moquette server. Configuration file path = {}.", configFile.getAbsolutePath());
IResourceLoader filesystemLoader = new FileResourceLoader(configFile);
final IConfig config = new ResourceLoaderConfig(filesystemLoader);
startServer(config);
@@ -119,6 +124,7 @@ public class Server {
* @throws IOException in case of any IO Error.
*/
public void startServer(Properties configProps) throws IOException {
LOG.info("Starting Moquette server using properties object...");
final IConfig config = new MemoryConfig(configProps);
startServer(config);
}
@@ -129,6 +135,7 @@ public class Server {
* @throws IOException in case of any IO Error.
*/
public void startServer(IConfig config) throws IOException {
LOG.info("Starting Moquette server using IConfig instance...");
startServer(config, null);
}
@@ -140,6 +147,7 @@ public class Server {
* @throws IOException in case of any IO Error.
* */
public void startServer(IConfig config, List<? extends InterceptHandler> handlers) throws IOException {
LOG.info("Starting moquette server using IConfig instance and intercept handlers...");
startServer(config, handlers, null, null, null);
}
@@ -149,29 +157,37 @@ public class Server {
if (handlers == null) {
handlers = Collections.emptyList();
}
LOG.info("Starting Moquette Server. MQTT message interceptors = {}.", getInterceptorIds(handlers));
final String handlerProp = System.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
if (handlerProp != null) {
config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp);
}
configureCluster(config);
LOG.info("Persistent store file: {}", config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME));
LOG.info("Configuring Using persistent store file. Path = {}.", config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME));
m_processorBootstrapper = new ProtocolProcessorBootstrapper();
LOG.info("Initializing MQTT protocol processor...");
final ProtocolProcessor processor = m_processorBootstrapper.init(config, handlers, authenticator, authorizator, this);
if (sslCtxCreator == null) {
LOG.warn("Using default SSL context creator...");
sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
}
LOG.info("Binding server to the configured ports...");
m_acceptor = new NettyAcceptor();
m_acceptor.initialize(processor, config, sslCtxCreator);
m_processor = processor;
LOG.info("The Moquette server has been initialized successfully.");
m_initialized = true;
}
private void configureCluster(IConfig config) throws FileNotFoundException {
LOG.info("Configuring embedded Hazelcast instance...");
String interceptHandlerClassname = config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
if (interceptHandlerClassname == null || !HZ_INTERCEPT_HANDLER.equals(interceptHandlerClassname)) {
LOG.info("There are no Hazelcast intercept handlers. The server won't start a Hazelcast instance.");
return;
}
String hzConfigPath = config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION);
@@ -180,16 +196,17 @@ public class Server {
Config hzconfig = isHzConfigOnClasspath ?
new ClasspathXmlConfig(hzConfigPath) :
new FileSystemXmlConfig(hzConfigPath);
LOG.info(String.format("starting server with hazelcast configuration file : %s",hzconfig));
LOG.info("Starting Hazelcast instance. ConfigurationFile = {}.",hzconfig);
hazelcastInstance = Hazelcast.newHazelcastInstance(hzconfig);
} else {
LOG.info("starting server with hazelcast default file");
LOG.info("Starting Hazelcast instance with default configuration.");
hazelcastInstance = Hazelcast.newHazelcastInstance();
}
listenOnHazelCastMsg();
}
private void listenOnHazelCastMsg() {
LOG.info("Subscribing to Hazelcast topic. TopicName = {}.", "moquette");
HazelcastInstance hz = getHazelcastInstance();
ITopic<HazelcastMsg> topic = hz.getTopic("moquette");
topic.addMessageListener(new HazelcastListener(this));
@@ -208,24 +225,31 @@ public class Server {
* */
public void internalPublish(PublishMessage msg) {
if (!m_initialized) {
throw new IllegalStateException("Can't publish on a server is not yet started");
LOG.error("The server is not started. The message cannot be published. MqttClientId = {}, messageId = {}.",
msg.getClientId(), msg.getMessageID());
throw new IllegalStateException("Can't publish on a server is not yet started");
}
if (LOG.isDebugEnabled()) {
LOG.debug("Publishing message. MqttClientId = {}, messageId = {}.", msg.getClientId(), msg.getMessageID());
}
m_processor.internalPublish(msg);
}
public void stopServer() {
LOG.info("Server stopping...");
LOG.info("Unbinding server from the configured ports...");
m_acceptor.close();
LOG.info("Stopping MQTT protocol processor...");
m_processorBootstrapper.shutdown();
m_initialized = false;
if (hazelcastInstance != null) {
LOG.info("Stopping embedded Hazelcast instance...");
try {
hazelcastInstance.shutdown();
} catch (HazelcastInstanceNotActiveException e) {
LOG.info("hazelcast already shutdown");
LOG.warn("The embedded Hazelcast instance is already shut down.");
}
}
LOG.info("Server stopped");
LOG.info("The Moquette server has been stopped.");
}
/**
@@ -246,11 +270,15 @@ public class Server {
* @param interceptHandler the handler to add.
* @return true id operation was successful.
* */
public boolean addInterceptHandler(InterceptHandler interceptHandler) {
public void addInterceptHandler(InterceptHandler interceptHandler) {
if (!m_initialized) {
LOG.error("The server is not started. The MQTT message interceptor cannot be added. InterceptorId = {}.",
interceptHandler.getID());
throw new IllegalStateException("Can't register interceptors on a server that is not yet started");
}
return m_processor.addInterceptHandler(interceptHandler);
LOG.info("Adding MQTT message interceptor. InterceptorId = {}.", interceptHandler.getID());
m_processor.addInterceptHandler(interceptHandler);
}
/**
@@ -258,11 +286,23 @@ public class Server {
* @param interceptHandler the handler to remove.
* @return true id operation was successful.
* */
public boolean removeInterceptHandler(InterceptHandler interceptHandler) {
public void removeInterceptHandler(InterceptHandler interceptHandler) {
if (!m_initialized) {
LOG.error("The server is not started. The MQTT message interceptor cannot be removed. InterceptorId = {}.",
interceptHandler.getID());
throw new IllegalStateException("Can't deregister interceptors from a server that is not yet started");
}
return m_processor.removeInterceptHandler(interceptHandler);
LOG.info("Removing MQTT message interceptor. InterceptorId = {}.", interceptHandler.getID());
m_processor.removeInterceptHandler(interceptHandler);
}
/**
* Returns the connections manager of this broker.
*
* @return
*/
public IConnectionsManager getConnectionsManager() {
return m_processorBootstrapper.getConnectionDescriptors();
}
}
@@ -4,8 +4,13 @@ import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ClasspathResourceLoader implements IResourceLoader {
private static final Logger LOG = LoggerFactory.getLogger(ClasspathResourceLoader.class);
private final String defaultResource;
private final ClassLoader classLoader;
@@ -29,6 +34,7 @@ public class ClasspathResourceLoader implements IResourceLoader {
@Override
public Reader loadResource(String relativePath) {
LOG.info("Loading resource. RelativePath = {}.", relativePath);
InputStream is = this.classLoader.getResourceAsStream(relativePath);
return is != null ? new InputStreamReader(is) : null;
}
@@ -5,8 +5,13 @@ import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.Reader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class FileResourceLoader implements IResourceLoader {
private static final Logger LOG = LoggerFactory.getLogger(FileResourceLoader.class);
private final File defaultFile;
private final String parentPath;
@@ -42,12 +47,15 @@ public class FileResourceLoader implements IResourceLoader {
}
public Reader loadResource(File f) {
LOG.info("Loading file. Path = {}.", f.getAbsolutePath());
if (f.isDirectory()) {
LOG.error("The given file is a directory. Path = {}.", f.getAbsolutePath());
throw new ResourceIsDirectoryException("File \"" + f + "\" is a directory!");
}
try {
return new FileReader(f);
} catch (FileNotFoundException e) {
LOG.error("The file does not exist. Path = {}.", f.getAbsolutePath());
return null;
}
}
@@ -11,7 +11,9 @@ public interface IResourceLoader {
String getName();
class ResourceIsDirectoryException extends RuntimeException {
public ResourceIsDirectoryException(String message) {
private static final long serialVersionUID = 1L;
public ResourceIsDirectoryException(String message) {
super(message);
}
}
@@ -37,20 +37,40 @@ public class ResourceLoaderConfig extends IConfig {
}
public ResourceLoaderConfig(IResourceLoader resourceLoader, String configName) {
LOG.info("Loading configuration. ResourceLoader = {}, configName = {}.", resourceLoader.getName(), configName);
this.resourceLoader = resourceLoader;
Reader configReader = configName != null ? resourceLoader.loadResource(configName) : resourceLoader.loadDefaultResource();
if (configReader == null) {
throw new IllegalArgumentException(
"Can't locate " + resourceLoader.getName() + " \"" + configName + "\"");
/*
* If we use a conditional operator, the loadResource() and the
* loadDefaultResource() methods will be always called. This makes the
* log traces confusing.
*/
Reader configReader;
if (configName != null) {
configReader = resourceLoader.loadResource(configName);
} else {
configReader = resourceLoader.loadDefaultResource();
}
if (configReader == null) {
LOG.error("The resource loader returned no configuration reader. ResourceLoader = {}, configName = {}.",
resourceLoader.getName(), configName);
throw new IllegalArgumentException("Can't locate " + resourceLoader.getName() + " \"" + configName + "\"");
}
LOG.info("Parsing configuration properties. ResourceLoader = {}, configName = {}.", resourceLoader.getName(),
configName);
ConfigurationParser confParser = new ConfigurationParser();
m_properties = confParser.getProperties();
assignDefaults();
try {
confParser.parse(configReader);
} catch (ParseException pex) {
LOG.warn("An error occurred in parsing configuration, fallback on default configuration", pex);
}
try {
confParser.parse(configReader);
} catch (ParseException pex) {
LOG.warn(
"Unable to parse configuration properties. Using default configuration. ResourceLoader = {}, configName = {}, cause = {}, errorMessage = {}.",
resourceLoader.getName(), configName, pex.getCause(), pex.getMessage());
}
}
@Override
@@ -17,9 +17,11 @@ package io.moquette.server.netty;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.EventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -31,6 +33,7 @@ import java.util.concurrent.TimeUnit;
*/
public class AutoFlushHandler extends ChannelDuplexHandler {
private static final Logger LOG = LoggerFactory.getLogger(AutoFlushHandler.class);
private static final long MIN_TIMEOUT_NANOS = TimeUnit.MILLISECONDS.toNanos(1);
private final long writerIdleTimeNanos;
@@ -109,6 +112,9 @@ public class AutoFlushHandler extends ChannelDuplexHandler {
private void initialize(ChannelHandlerContext ctx) {
// Avoid the case where destroy() is called before scheduling timeouts.
// See: https://github.com/netty/netty/issues/143
if (LOG.isDebugEnabled()) {
LOG.debug("Initializing autoflush handler. MqttClientId = {}.", NettyUtils.clientID(ctx.channel()));
}
switch (state) {
case 1:
case 2:
@@ -142,6 +148,9 @@ public class AutoFlushHandler extends ChannelDuplexHandler {
*/
protected void channelIdle(ChannelHandlerContext ctx/*, IdleStateEvent evt*/) throws Exception {
// ctx.fireUserEventTriggered(evt);
if (LOG.isDebugEnabled()) {
LOG.debug("Flushing idle Netty channel. MqttClientId = {}.", NettyUtils.clientID(ctx.channel()));
}
ctx.channel().flush();
}
@@ -15,6 +15,9 @@
*/
package io.moquette.server.netty;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
@@ -23,16 +26,24 @@ import io.netty.handler.timeout.IdleStateEvent;
@Sharable
class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler {
private static final Logger LOG = LoggerFactory.getLogger(MoquetteIdleTimeoutHandler.class);
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleState e = ((IdleStateEvent) evt).state();
if (e == IdleState.ALL_IDLE) {
LOG.info("Firing channel inactive event. MqttClientId = {}.", NettyUtils.clientID(ctx.channel()));
//fire a channelInactive to trigger publish of Will
ctx.fireChannelInactive();
ctx.close();
}
} else {
if (LOG.isDebugEnabled()) {
LOG.debug("Firing Netty event. MqttClientId = {}, eventClass = {}.", NettyUtils.clientID(ctx.channel()),
evt.getClass().getName());
}
super.userEventTriggered(ctx, evt);
}
}
@@ -17,7 +17,6 @@ package io.moquette.server.netty;
import io.moquette.BrokerConstants;
import static io.moquette.BrokerConstants.*;
import io.moquette.parser.commons.Constants;
import io.moquette.parser.netty.MQTTDecoder;
import io.moquette.parser.netty.MQTTEncoder;
import io.moquette.spi.security.ISslContextCreator;
@@ -38,7 +37,6 @@ import io.netty.handler.codec.http.HttpRequestDecoder;
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.Utf8FrameValidator;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
@@ -50,6 +48,7 @@ import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
*
@@ -95,8 +94,23 @@ public class NettyAcceptor implements ServerAcceptor {
BytesMetricsCollector m_bytesMetricsCollector = new BytesMetricsCollector();
MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector();
private int nettySoBacklog;
private boolean nettySoReuseaddr;
private boolean nettyTcpNodelay;
private boolean nettySoKeepalive;
private int nettyChannelTimeoutSeconds;
@Override
public void initialize(ProtocolProcessor processor, IConfig props, ISslContextCreator sslCtxCreator) throws IOException {
LOG.info("Initializing Netty acceptor...");
nettySoBacklog = Integer.parseInt(props.getProperty(BrokerConstants.NETTY_SO_BACKLOG_PROPERTY_NAME, "128"));
nettySoReuseaddr = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_SO_REUSEADDR_PROPERTY_NAME, "true"));
nettyTcpNodelay = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_TCP_NODELAY_PROPERTY_NAME, "true"));
nettySoKeepalive = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_SO_KEEPALIVE_PROPERTY_NAME, "true"));
nettyChannelTimeoutSeconds = Integer
.parseInt(props.getProperty(BrokerConstants.NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME, "10"));
m_bossGroup = new NioEventLoopGroup();
m_workerGroup = new NioEventLoopGroup();
final NettyMQTTHandler handler = new NettyMQTTHandler(processor);
@@ -116,7 +130,8 @@ public class NettyAcceptor implements ServerAcceptor {
}
}
private void initFactory(String host, int port, final PipelineInitializer pipeliner) {
private void initFactory(String host, int port, String protocol, final PipelineInitializer pipeliner) {
LOG.info("Initializing server. Protocol = {}.", protocol);
ServerBootstrap b = new ServerBootstrap();
b.group(m_bossGroup, m_workerGroup)
.channel(NioServerSocketChannel.class)
@@ -132,33 +147,35 @@ public class NettyAcceptor implements ServerAcceptor {
}
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.option(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.TCP_NODELAY, true)
.childOption(ChannelOption.SO_KEEPALIVE, true);
.option(ChannelOption.SO_BACKLOG, nettySoBacklog)
.option(ChannelOption.SO_REUSEADDR, nettySoReuseaddr)
.option(ChannelOption.TCP_NODELAY, nettyTcpNodelay)
.childOption(ChannelOption.SO_KEEPALIVE, nettySoKeepalive);
try {
LOG.info("Binding server. Protocol = {}, host = {}, port = {}.", host, port);
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(host, port);
LOG.info("Server binded host: {}, port: {}", host, port);
LOG.info("The server has been binded. Protocol = {}, host = {}, port = {}", host, port);
f.sync();
} catch (InterruptedException ex) {
LOG.error(null, ex);
LOG.error("An interruptedException was caught while initializing server. Protocol = {}.", protocol, ex);
}
}
private void initializePlainTCPTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
LOG.info("Configuring TCP MQTT transport...");
final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
String tcpPortProp = props.getProperty(PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (DISABLED_PORT_BIND.equals(tcpPortProp)) {
LOG.info("tcp MQTT is disabled because the value for the property with key {}", BrokerConstants.PORT_PROPERTY_NAME);
LOG.info("The property {} has been setted to {}. TCP MQTT will be disabled.", BrokerConstants.PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
return;
}
int port = Integer.parseInt(tcpPortProp);
initFactory(host, port, new PipelineInitializer() {
initFactory(host, port, "TCP MQTT", new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) {
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds));
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
// pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
@@ -172,10 +189,12 @@ public class NettyAcceptor implements ServerAcceptor {
}
private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
LOG.info("Configuring Websocket MQTT transport...");
String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (DISABLED_PORT_BIND.equals(webSocketPortProp)) {
//Do nothing no WebSocket configured
LOG.info("WebSocket is disabled");
LOG.info("The property {} has been setted to {}. Websocket MQTT will be disabled.",
BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
return;
}
int port = Integer.parseInt(webSocketPortProp);
@@ -183,7 +202,7 @@ public class NettyAcceptor implements ServerAcceptor {
final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
initFactory(host, port, new PipelineInitializer() {
initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) {
pipeline.addLast(new HttpServerCodec());
@@ -191,7 +210,7 @@ public class NettyAcceptor implements ServerAcceptor {
pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
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());
@@ -204,10 +223,12 @@ public class NettyAcceptor implements ServerAcceptor {
}
private void initializeSSLTCPTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext) throws IOException {
String sslPortProp = props.getProperty(SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
LOG.info("Configuring SSL MQTT transport...");
String sslPortProp = props.getProperty(SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (DISABLED_PORT_BIND.equals(sslPortProp)) {
//Do nothing no SSL configured
LOG.info("SSL MQTT is disabled because there is no value in properties for key {}", BrokerConstants.SSL_PORT_PROPERTY_NAME);
LOG.info("The property {} has been setted to {}. SSL MQTT will be disabled.",
BrokerConstants.SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
return;
}
@@ -218,11 +239,11 @@ public class NettyAcceptor implements ServerAcceptor {
String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
initFactory(host, sslPort, new PipelineInitializer() {
initFactory(host, sslPort, "SSL MQTT", new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) throws Exception {
pipeline.addLast("ssl", createSslHandler(sslContext, needsClientAuth));
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, nettyChannelTimeoutSeconds));
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
//pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
@@ -236,10 +257,12 @@ public class NettyAcceptor implements ServerAcceptor {
}
private void initializeWSSTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext) throws IOException {
String sslPortProp = props.getProperty(WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
LOG.info("Configuring secure websocket MQTT transport...");
String sslPortProp = props.getProperty(WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
if (DISABLED_PORT_BIND.equals(sslPortProp)) {
//Do nothing no SSL configured
LOG.info("SSL websocket is disabled because there is no value in properties for key {}", BrokerConstants.WSS_PORT_PROPERTY_NAME);
LOG.info("The property {} has been setted to {}. Secure websocket MQTT will be disabled.",
BrokerConstants.WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
return;
}
int sslPort = Integer.parseInt(sslPortProp);
@@ -247,7 +270,7 @@ public class NettyAcceptor implements ServerAcceptor {
String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
initFactory(host, sslPort, new PipelineInitializer() {
initFactory(host, sslPort, "Secure websocket", new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) throws Exception {
pipeline.addLast("ssl", createSslHandler(sslContext, needsClientAuth));
@@ -257,7 +280,7 @@ public class NettyAcceptor implements ServerAcceptor {
pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
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());
@@ -270,32 +293,45 @@ public class NettyAcceptor implements ServerAcceptor {
}
public void close() {
if (m_workerGroup == null) {
throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
}
if (m_bossGroup == null) {
throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
}
Future workerWaiter = m_workerGroup.shutdownGracefully();
Future bossWaiter = m_bossGroup.shutdownGracefully();
LOG.info("Closing Netty acceptor...");
if (m_workerGroup == null || m_bossGroup == null) {
LOG.error("The Netty acceptor is not initialized.");
throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
}
Future<?> workerWaiter = m_workerGroup.shutdownGracefully();
Future<?> bossWaiter = m_bossGroup.shutdownGracefully();
/*
* We shouldn't raise an IllegalStateException if we are interrupted. If
* we did so, the broker is not shut down properly.
*/
LOG.info("Waiting for worker and boss event loop groups to terminate...");
try {
workerWaiter.await(100);
workerWaiter.await(10, TimeUnit.SECONDS);
bossWaiter.await(10, TimeUnit.SECONDS);
} catch (InterruptedException iex) {
throw new IllegalStateException(iex);
LOG.warn("An InterruptedException was caught while waiting for event loops to terminate...");
}
try {
bossWaiter.await(100);
} catch (InterruptedException iex) {
throw new IllegalStateException(iex);
if (!m_workerGroup.isTerminated()) {
LOG.warn("Forcing shutdown of worker event loop...");
m_workerGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS);
}
if (!m_bossGroup.isTerminated()) {
LOG.warn("Forcing shutdown of boss event loop...");
m_bossGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS);
}
LOG.info("Collecting message metrics...");
MessageMetrics metrics = m_metricsCollector.computeMetrics();
LOG.info("Msg read: {}, msg wrote: {}", metrics.messagesRead(), metrics.messagesWrote());
LOG.info("The metrics have been collected. Read messages = {}, written messages = {}.",
metrics.messagesRead(), metrics.messagesWrote());
LOG.info("Collecting bytes metrics...");
BytesMetrics bytesMetrics = m_bytesMetricsCollector.computeMetrics();
LOG.info(String.format("Bytes read: %d, bytes wrote: %d", bytesMetrics.readBytes(), bytesMetrics.wroteBytes()));
LOG.info("The bytes metrics have been collected. Read bytes = {}, written bytes = {}.", bytesMetrics.readBytes(),
bytesMetrics.wroteBytes());
}
private ChannelHandler createSslHandler(SSLContext sslContext, boolean needsClientAuth) {
@@ -23,12 +23,10 @@ import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.CorruptedFrameException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
/**
*
@@ -47,7 +45,10 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
AbstractMessage msg = (AbstractMessage) message;
LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType()));
String messageType = Utils.msgType2String(msg.getMessageType());
if (LOG.isDebugEnabled()) {
LOG.debug("Processing MQTT message. MessageType = {}.", messageType);
}
try {
switch (msg.getMessageType()) {
case CONNECT:
@@ -82,9 +83,11 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
ctx.writeAndFlush(pingResp);
break;
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
ctx.fireExceptionCaught(ex);
} catch (Throwable ex) {
LOG.error(
"An unexpected exception was caught while processing MQTT message. MessageType = {}, cause = {}, errorMessage = {}.",
messageType, ex.getCause(), ex.getMessage());
ctx.fireExceptionCaught(ex);
}
}
@@ -92,6 +95,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String clientID = NettyUtils.clientID(ctx.channel());
if (clientID != null && !clientID.isEmpty()) {
LOG.info("Notifying connection lost event. MqttClientId = {}.", clientID);
m_processor.processConnectionLost(clientID, ctx.channel());
}
ctx.close();
@@ -99,14 +103,9 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
@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 if (cause instanceof IOException && "Connection reset by peer".equals(cause.getMessage())) {
LOG.warn("Network connection closed abruptly");
} else {
LOG.error("Ugly error on networking", cause);
}
LOG.error(
"An unexpected exception was caught while processing MQTT message. Closing Netty channel. MqttClientId = {}, cause = {}, errorMessage = {}.",
NettyUtils.clientID(ctx.channel()), cause.getCause(), cause.getMessage());
ctx.close();
}
@@ -16,7 +16,6 @@
package io.moquette.server.netty;
import io.moquette.server.Constants;
import io.moquette.spi.impl.ProtocolProcessor;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.Attribute;
@@ -16,6 +16,7 @@
package io.moquette.server.netty.metrics;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
@@ -34,7 +35,7 @@ public class BytesMetricsHandler extends ChannelDuplexHandler {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Attribute<BytesMetrics> attr = ctx.attr(ATTR_KEY_METRICS);
Attribute<BytesMetrics> attr = ctx.channel().attr(ATTR_KEY_METRICS);
attr.set(new BytesMetrics());
super.channelActive(ctx);
@@ -42,14 +43,14 @@ public class BytesMetricsHandler extends ChannelDuplexHandler {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
BytesMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
metrics.incrementRead(((ByteBuf)msg).readableBytes());
ctx.fireChannelRead(msg);
}
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
BytesMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
metrics.incrementWrote(((ByteBuf)msg).writableBytes());
ctx.write(msg, promise);
}
@@ -58,9 +59,13 @@ public class BytesMetricsHandler extends ChannelDuplexHandler {
@Override
public void close(ChannelHandlerContext ctx,
ChannelPromise promise) throws Exception {
BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
BytesMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
m_collector.sumReadBytes(metrics.readBytes());
m_collector.sumWroteBytes(metrics.wroteBytes());
super.close(ctx, promise);
}
public static BytesMetrics getBytesMetrics(Channel channel) {
return channel.attr(ATTR_KEY_METRICS).get();
}
}
@@ -15,6 +15,7 @@
*/
package io.moquette.server.netty.metrics;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
@@ -33,7 +34,7 @@ public class MessageMetricsHandler extends ChannelDuplexHandler {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Attribute<MessageMetrics> attr = ctx.attr(ATTR_KEY_METRICS);
Attribute<MessageMetrics> attr = ctx.channel().attr(ATTR_KEY_METRICS);
attr.set(new MessageMetrics());
super.channelActive(ctx);
@@ -41,14 +42,14 @@ public class MessageMetricsHandler extends ChannelDuplexHandler {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
MessageMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
metrics.incrementRead(1);
ctx.fireChannelRead(msg);
}
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
MessageMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
metrics.incrementWrote(1);
ctx.write(msg, promise);
}
@@ -57,9 +58,13 @@ public class MessageMetricsHandler extends ChannelDuplexHandler {
@Override
public void close(ChannelHandlerContext ctx,
ChannelPromise promise) throws Exception {
MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
MessageMetrics metrics = ctx.channel().attr(ATTR_KEY_METRICS).get();
m_collector.sumReadMessages(metrics.messagesRead());
m_collector.sumWroteMessages(metrics.messagesWrote());
super.close(ctx, promise);
}
public static MessageMetrics getMessageMetrics(Channel channel) {
return channel.attr(ATTR_KEY_METRICS).get();
}
}
@@ -74,6 +74,7 @@ public class ClientSession {
* @return the list of messages to be delivered for client related to the session.
* */
public BlockingQueue<IMessagesStore.StoredMessage> queue() {
LOG.info("Retrieving stored messages. MqttClientId = {}.", clientID);
return this.m_sessionsStore.queue(clientID);
}
@@ -83,11 +84,12 @@ public class ClientSession {
}
public boolean subscribe(Subscription newSubscription) {
LOG.info("<{}> subscribed to the topic filter <{}> with QoS {}",
newSubscription.getClientId(), newSubscription.getTopicFilter(),
AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
LOG.info("Adding new subscription. MqttClientId = {}, topics = {}, qos = {}.", newSubscription.getClientId(),
newSubscription.getTopicFilter(), AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter());
if (!validTopic) {
LOG.error("The topic filter is not valid. MqttClientId = {}, topics = {}.", newSubscription.getClientId(),
newSubscription.getTopicFilter());
//send SUBACK with 0x80 for this topic filter
return false;
}
@@ -96,6 +98,9 @@ public class ClientSession {
//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) {
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(),
newSubscription.getRequestedQos());
subscriptions.remove(newSubscription);
}
subscriptions.add(newSubscription);
@@ -105,6 +110,7 @@ public class ClientSession {
}
public void unsubscribeFrom(String topicFilter) {
LOG.info("Removing subscription. MqttClientID = {}, topics = {}.", clientID, topicFilter);
m_sessionsStore.removeSubscription(topicFilter, clientID);
Set<Subscription> subscriptionsToRemove = new HashSet<>();
for (Subscription sub : this.subscriptions) {
@@ -117,6 +123,7 @@ public class ClientSession {
public void disconnect() {
if (this.cleanSession) {
LOG.info("The client has disconnected. Removing its subscriptions. MqttClientId = {}.", clientID);
//cleanup topic subscriptions
cleanSession();
}
@@ -124,13 +131,12 @@ public class ClientSession {
}
public void cleanSession() {
LOG.info("cleaning old saved subscriptions for client <{}>", this.clientID);
LOG.info("Wiping existing subscriptions. MqttClientId = {}.", this.clientID);
m_sessionsStore.wipeSubscriptions(this.clientID);
LOG.debug("Wiped subscriptions for client <{}>", this.clientID);
//remove also the messages stored of type QoS1/2
LOG.info("Removing stored messages with QoS 1 and 2. MqttClientId = {}.", this.clientID);
messagesStore.dropMessagesInSession(this.clientID);
LOG.debug("Removed messages in session for client <{}>", this.clientID);
}
public boolean isCleanSession() {
@@ -147,12 +153,14 @@ public class ClientSession {
}
public void inFlightAcknowledged(int messageID) {
LOG.trace("Acknowledging inflight, clientID <{}> messageID {}", this.clientID, messageID);
if (LOG.isTraceEnabled())
LOG.trace("Acknowledging inflight, clientID <{}> messageID {}", this.clientID, messageID);
m_sessionsStore.inFlightAck(this.clientID, messageID);
}
public void inFlightAckWaiting(MessageGUID guid, int messageID) {
LOG.trace("Adding to inflight {}, guid <{}>", messageID, guid);
if (LOG.isTraceEnabled())
LOG.trace("Adding to inflight {}, guid <{}>", messageID, guid);
m_sessionsStore.inFlight(this.clientID, messageID, guid);
}
@@ -181,4 +189,21 @@ public class ClientSession {
public IMessagesStore.StoredMessage getInflightMessage(int messageID) {
return m_sessionsStore.getInflightMessage(clientID, messageID);
}
public Set<Subscription> getSubscriptions() {
return subscriptions;
}
public int getPendingPublishMessagesNo() {
return m_sessionsStore.getPendingPublishMessagesNo(clientID);
}
public int getSecondPhaseAckPendingMessages() {
return m_sessionsStore.getSecondPhaseAckPendingMessages(clientID);
}
public int getInflightMessagesNo() {
return m_sessionsStore.getInflightMessagesNo(clientID);
}
}
@@ -20,8 +20,6 @@ import java.nio.ByteBuffer;
import io.moquette.parser.proto.messages.AbstractMessage;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Defines the SPI to be implemented by a StorageService that handle persistence of messages
@@ -29,7 +27,8 @@ import java.util.concurrent.atomic.AtomicInteger;
public interface IMessagesStore {
class StoredMessage implements Serializable {
final AbstractMessage.QOSType m_qos;
private static final long serialVersionUID = 1L;
final AbstractMessage.QOSType m_qos;
final byte[] m_payload;
final String m_topic;
private boolean m_retained;
@@ -136,4 +135,6 @@ public interface IMessagesStore {
StoredMessage getMessageByGuid(MessageGUID guid);
void cleanRetained(String topic);
int getPendingPublishMessages(String clientID);
}
@@ -15,6 +15,7 @@
*/
package io.moquette.spi;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
@@ -117,6 +118,12 @@ public interface ISessionsStore {
* @return the session for the given clientID, null if not found.
* */
ClientSession sessionForClient(String clientID);
/**
* Returns all the sessions
* @return
*/
Collection<ClientSession> getAllSessions();
void inFlightAck(String clientID, int messageID);
@@ -156,4 +163,29 @@ public interface ISessionsStore {
MessageGUID mapToGuid(String clientID, int messageID);
StoredMessage getInflightMessage(String clientID, int messageID);
/**
* Returns the number of inflight messages for the given client ID
*
* @param clientID
* @return
*/
int getInflightMessagesNo(String clientID);
/**
* Returns the size of the session queue for the given client ID
*
* @param clientID
* @return
*/
int getPendingPublishMessagesNo(String clientID);
/**
* Returns the number of second-phase ACK pending messages for the given
* client ID
*
* @param clientID
* @return
*/
int getSecondPhaseAckPendingMessages(String clientID);
}
@@ -21,7 +21,8 @@ import java.io.Serializable;
* Value object for GUIDs of messages.
* */
public class MessageGUID implements Serializable {
private final String guid;
private static final long serialVersionUID = 1L;
private final String guid;
public MessageGUID(String guid) {
this.guid = guid;
@@ -15,17 +15,32 @@
*/
package io.moquette.spi.impl;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.Interceptor;
import io.moquette.interception.messages.*;
import io.moquette.parser.proto.messages.ConnectMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.parser.proto.messages.PublishMessage;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
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;
/**
* An interceptor that execute the interception tasks asynchronously.
@@ -33,24 +48,62 @@ import java.util.concurrent.Executors;
* @author Wagner Macedo
*/
final class BrokerInterceptor implements Interceptor {
private final List<InterceptHandler> handlers;
private static final Logger LOG = LoggerFactory.getLogger(BrokerInterceptor.class);
private final Map<Class<?>, List<InterceptHandler>> handlers;
private final ExecutorService executor;
BrokerInterceptor(List<InterceptHandler> handlers) {
this.handlers = new CopyOnWriteArrayList<>(handlers);
executor = Executors.newFixedThreadPool(1);
}
private BrokerInterceptor(int poolSize, List<InterceptHandler> handlers) {
LOG.info("Initializing broker interceptor. InterceptorIds = {}.", getInterceptorIds(handlers));
this.handlers = new HashMap<>();
for (Class<?> messageType : InterceptHandler.ALL_MESSAGE_TYPES) {
this.handlers.put(messageType, new CopyOnWriteArrayList<InterceptHandler>());
}
for (InterceptHandler handler : handlers) {
this.addInterceptHandler(handler);
}
executor = Executors.newFixedThreadPool(poolSize);
}
/**
* Configures a broker interceptor, with a thread pool of one thread.
* @param handlers
*/
public BrokerInterceptor(List<InterceptHandler> handlers) {
this(1, handlers);
}
/**
* Configures a broker interceptor using the pool size specified in the
* IConfig argument.
*/
public BrokerInterceptor(IConfig props, List<InterceptHandler> handlers) {
this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers);
}
/**
* Shutdown graciously the executor service
*/
void stop() {
executor.shutdown();
}
void stop() {
LOG.info("Shutting down interceptor thread pool...");
executor.shutdown();
try {
LOG.info("Waiting for thread pool tasks to terminate...");
executor.awaitTermination(10L, TimeUnit.SECONDS);
} catch (InterruptedException e) {}
if (!executor.isTerminated()) {
LOG.warn("Forcing shutdown of interceptor thread pool...");
executor.shutdownNow();
}
}
@Override
public void notifyClientConnected(final ConnectMessage msg) {
for (final InterceptHandler handler : this.handlers) {
for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) {
if (LOG.isDebugEnabled()) {
LOG.debug("Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.",
msg.getClientID(), handler.getID());
}
executor.execute(new Runnable() {
@Override
public void run() {
@@ -62,7 +115,11 @@ final class BrokerInterceptor implements Interceptor {
@Override
public void notifyClientDisconnected(final String clientID, final String username ) {
for (final InterceptHandler handler : this.handlers) {
for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) {
if (LOG.isDebugEnabled()) {
LOG.debug("Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, interceptorId = {}.",
clientID, username, handler.getID());
}
executor.execute(new Runnable() {
@Override
public void run() {
@@ -74,7 +131,11 @@ final class BrokerInterceptor implements Interceptor {
@Override
public void notifyClientConnectionLost(final String clientID, final String username) {
for (final InterceptHandler handler : this.handlers) {
for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) {
if (LOG.isDebugEnabled()) {
LOG.debug("Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, interceptorId = {}.",
clientID, username, handler.getID());
}
executor.execute(new Runnable() {
@Override
public void run() {
@@ -86,7 +147,11 @@ final class BrokerInterceptor implements Interceptor {
@Override
public void notifyTopicPublished(final PublishMessage msg, final String clientID, final String username) {
for (final InterceptHandler handler : this.handlers) {
for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) {
if (LOG.isDebugEnabled()) {
LOG.debug("Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, interceptorId = {}.",
clientID, msg.getMessageID(), msg.getTopicName(), handler.getID());
}
executor.execute(new Runnable() {
@Override
public void run() {
@@ -98,7 +163,11 @@ final class BrokerInterceptor implements Interceptor {
@Override
public void notifyTopicSubscribed(final Subscription sub, final String username) {
for (final InterceptHandler handler : this.handlers) {
for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) {
if (LOG.isDebugEnabled()) {
LOG.debug("Notifying MQTT SUBSCRIBE message to interceptor. MqttClientId = {}, topicFilter = {}, interceptorId = {}.",
sub.getClientId(), sub.getTopicFilter(), handler.getID());
}
executor.execute(new Runnable() {
@Override
public void run() {
@@ -110,7 +179,11 @@ final class BrokerInterceptor implements Interceptor {
@Override
public void notifyTopicUnsubscribed(final String topic, final String clientID, final String username) {
for (final InterceptHandler handler : this.handlers) {
for (final InterceptHandler handler : this.handlers.get(InterceptUnsubscribeMessage.class)) {
if (LOG.isDebugEnabled()) {
LOG.debug("Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, interceptorId = {}.",
clientID, topic, handler.getID());
}
executor.execute(new Runnable() {
@Override
public void run() {
@@ -122,7 +195,11 @@ final class BrokerInterceptor implements Interceptor {
@Override
public void notifyMessageAcknowledged( final InterceptAcknowledgedMessage msg ) {
for (final InterceptHandler handler : this.handlers) {
for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) {
if (LOG.isDebugEnabled()) {
LOG.debug("Notifying MQTT ACK message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, interceptorId = {}.",
msg.getMsg().getClientID(), msg.getMsg().getMessageID(), msg.getTopic(), handler.getID());
}
executor.execute(new Runnable() {
@Override
public void run() {
@@ -132,13 +209,31 @@ final class BrokerInterceptor implements Interceptor {
}
}
@Override
public boolean addInterceptHandler(InterceptHandler interceptHandler) {
return this.handlers.add(interceptHandler);
@Override
public void addInterceptHandler(InterceptHandler interceptHandler) {
Class<?>[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler);
LOG.info("Adding MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.",
interceptHandler.getID(), interceptedMessageTypes);
for (Class<?> interceptMessageType : interceptedMessageTypes) {
this.handlers.get(interceptMessageType).add(interceptHandler);
}
}
@Override
public void removeInterceptHandler(InterceptHandler interceptHandler) {
Class<?>[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler);
LOG.info("Removing MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.",
interceptHandler.getID(), interceptedMessageTypes);
for (Class<?> interceptMessageType : interceptedMessageTypes) {
this.handlers.get(interceptMessageType).remove(interceptHandler);
}
}
@Override
public boolean removeInterceptHandler(InterceptHandler interceptHandler) {
return this.handlers.remove(interceptHandler);
}
private static Class<?>[] getInterceptedMessageTypes(InterceptHandler interceptHandler) {
Class<?> interceptedMessageTypes[] = interceptHandler.getInterceptedMessageTypes();
if (interceptedMessageTypes == null) {
return InterceptHandler.ALL_MESSAGE_TYPES;
}
return interceptedMessageTypes;
}
}
@@ -12,7 +12,6 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
class InternalRepublisher {
@@ -30,7 +29,10 @@ class InternalRepublisher {
//fire as retained the message
Integer packetID = storedMsg.getQos() == AbstractMessage.QOSType.MOST_ONE ? null : targetSession.nextPacketId();
if (packetID != null) {
LOG.trace("Adding to inflight <{}>", packetID);
if (LOG.isDebugEnabled()) {
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);
@@ -48,7 +50,10 @@ class InternalRepublisher {
for (IMessagesStore.StoredMessage pubEvt : storedPublishes) {
//put in flight zone
LOG.trace("Adding to inflight <{}>", pubEvt.getMessageID());
if (LOG.isDebugEnabled()) {
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);
//set the PacketIdentifier only for QoS > 0
@@ -2,7 +2,7 @@ package io.moquette.spi.impl;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
@@ -13,19 +13,18 @@ import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired;
class MessagesPublisher {
private static final Logger LOG = LoggerFactory.getLogger(MessagesPublisher.class);
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
private final ConnectionDescriptorStore connectionDescriptors;
private final ISessionsStore m_sessionsStore;
private final IMessagesStore m_messagesStore;
private final PersistentQueueMessageSender messageSender;
public MessagesPublisher(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore,
public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
IMessagesStore messagesStore, PersistentQueueMessageSender messageSender) {
this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore;
@@ -53,18 +52,20 @@ class MessagesPublisher {
if (publishingQos != AbstractMessage.QOSType.MOST_ONE) {
guid = m_messagesStore.storePublishForFuture(pubMsg);
}
LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic);
for (final Subscription sub : topicMatchingSubscriptions) {
AbstractMessage.QOSType qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
boolean targetIsActive = this.connectionDescriptors.containsKey(sub.getClientId());
boolean targetIsActive = this.connectionDescriptors.isConnected(sub.getClientId());
LOG.debug("Broker republishing to client <{}> topicFilter <{}> qos <{}>, active {}",
sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive);
ByteBuffer message = origMessage.duplicate();
if (targetIsActive) {
if (LOG.isDebugEnabled()) {
LOG.debug(
"Sending PUBLISH message to subscriber. MessageId = {}, mqttClientId = {}, topicFilter = {}, qos = {}, active = {}.",
pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive);
}
PublishMessage publishMsg = notRetainedPublish(topic, qos, message);
if (qos != AbstractMessage.QOSType.MOST_ONE) {
//QoS 1 or 2
@@ -76,6 +77,11 @@ class MessagesPublisher {
this.messageSender.sendPublish(targetSession, publishMsg);
} else {
if (!targetSession.isCleanSession()) {
if (LOG.isDebugEnabled()) {
LOG.debug(
"Storing pending PUBLISH message. MessageId = {}, mqttClientId = {}, topicFilter = {}, qos = {}, active = {}.",
pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive);
}
//store the message in targetSession queue to deliver
targetSession.enqueue(pubMsg);
}
@@ -1,13 +1,11 @@
package io.moquette.spi.impl;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.spi.ClientSession;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
@@ -15,41 +13,34 @@ import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
class PersistentQueueMessageSender {
private static final Logger LOG = LoggerFactory.getLogger(PersistentQueueMessageSender.class);
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
private final ConnectionDescriptorStore connectionDescriptorStore;
public PersistentQueueMessageSender(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors) {
this.connectionDescriptors = connectionDescriptors;
public PersistentQueueMessageSender(ConnectionDescriptorStore connectionDescriptorStore) {
this.connectionDescriptorStore = connectionDescriptorStore;
}
void sendPublish(ClientSession clientsession, PublishMessage pubMessage) {
String clientId = clientsession.clientID;
LOG.info("send publish message to <{}> on topic <{}>", clientId, pubMessage.getTopicName());
if (LOG.isDebugEnabled()) {
LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}",
clientId, pubMessage.getTopicName(), pubMessage.getQos(), false, pubMessage.getMessageID());
LOG.debug("content <{}>", DebugUtils.payload2Str(pubMessage.getPayload()));
LOG.debug("Sending PUBLISH message. MessageId = {}, mqttClientId = {}, topic = {}, qos = {}, payload = {}.",
pubMessage.getMessageID(),
clientId, pubMessage.getTopicName(), DebugUtils.payload2Str(pubMessage.getPayload()));
} else {
LOG.info("Sending PUBLISH message. MessageId = {}, mqttClientId = {}, topic = {}.", pubMessage.getMessageID(),
clientId, pubMessage.getTopicName());
}
if (connectionDescriptors == null) {
throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be " +
"initialized, somewhere it's overwritten!!");
}
if (connectionDescriptors.get(clientId) == null) {
//TODO while we were publishing to the target client, that client disconnected,
// could happen is not an error HANDLE IT
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>",
clientId, connectionDescriptors));
}
Channel channel = connectionDescriptors.get(clientId).channel;
LOG.trace("Session for clientId {}", clientId);
if (channel.isWritable()) {
LOG.debug("channel is writable");
//if channel is writable don't enqueue
channel.writeAndFlush(pubMessage);
} else if (pubMessage.getQos() != MOST_ONE) {
//enqueue to the client session
LOG.debug("enqueue to client session");
clientsession.enqueue(asStoredMessage(pubMessage));
}
boolean messageDelivered = connectionDescriptorStore.sendMessage(pubMessage, pubMessage.getMessageID(), clientId);
if (!messageDelivered && pubMessage.getQos() != MOST_ONE && !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);
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);
}
}
}
@@ -23,7 +23,7 @@ import io.moquette.interception.InterceptHandler;
import io.moquette.parser.proto.messages.*;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.ConnectionDescriptor.ConnectionState;
import io.moquette.server.netty.AutoFlushHandler;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.*;
import io.moquette.spi.IMessagesStore.StoredMessage;
@@ -120,7 +120,7 @@ public class ProtocolProcessor {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
protected ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
protected ConnectionDescriptorStore connectionDescriptors;
protected ConcurrentMap<RunningSubscription, SubscriptionState> subscriptionInCourse;
private SubscriptionsStore subscriptions;
@@ -158,6 +158,15 @@ public class ProtocolProcessor {
boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor) {
init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous, allowZeroByteClientId, authorizator,interceptor,null);
}
public void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous,
boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) {
init(new ConnectionDescriptorStore(sessionsStore), subscriptions, storageService, sessionsStore, authenticator,
allowAnonymous, allowZeroByteClientId, authorizator, interceptor, serverPort);
}
/**
* @param subscriptions the subscription store where are stored all the existing
@@ -171,27 +180,32 @@ public class ProtocolProcessor {
* @param authorizator used to apply ACL policies to publishes and subscriptions.
* @param interceptor to notify events to an intercept handler
*/
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
void init(ConnectionDescriptorStore connectionDescriptors, SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous,
boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) {
this.connectionDescriptors = new ConcurrentHashMap<>();
LOG.info("Initializing MQTT protocol processor...");
this.connectionDescriptors = connectionDescriptors;
this.subscriptionInCourse = new ConcurrentHashMap<>();
this.m_interceptor = interceptor;
this.subscriptions = subscriptions;
this.allowAnonymous = allowAnonymous;
this.allowZeroByteClientId = allowZeroByteClientId;
m_authorizator = authorizator;
LOG.trace("subscription tree on init {}", subscriptions.dumpTree());
if (LOG.isDebugEnabled()) {
LOG.debug("Initial subscriptions tree = {}.", subscriptions.dumpTree());
}
m_authenticator = authenticator;
m_messagesStore = storageService;
m_sessionsStore = sessionsStore;
m_server_port = serverPort;
LOG.info("Initializing messages publisher...");
final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors);
this.messagesPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, m_messagesStore, messageSender);
LOG.info("Initializing QoS publish handlers...");
this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, subscriptions, m_messagesStore,
m_interceptor, this.messagesPublisher);
this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, subscriptions, m_messagesStore,
@@ -199,16 +213,19 @@ public class ProtocolProcessor {
this.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore,
m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port, this.messagesPublisher);
LOG.info("Initializing internal republisher...");
this.internalRepublisher = new InternalRepublisher(messageSender);
}
public void processConnect(Channel channel, ConnectMessage msg) {
LOG.info("CONNECT for client <{}>", msg.getClientID());
LOG.info("Processing CONNECT message. MqttClientId = {}, username = {}, password = {}.", msg.getClientID(),
msg.getUsername(), msg.getPassword());
if (msg.getProtocolVersion() != VERSION_3_1 && msg.getProtocolVersion() != VERSION_3_1_1) {
ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
LOG.warn("CONNECT sent bad proto ConnAck");
LOG.error("The MQTT protocol version is not valid. MqttClientId = {}, protocolVersion = {}.",
msg.getProtocolVersion());
channel.writeAndFlush(badProto);
channel.close();
return;
@@ -220,14 +237,16 @@ public class ProtocolProcessor {
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
channel.writeAndFlush(okResp);
channel.close();
LOG.warn("CONNECT sent rejected identifier ConnAck");
LOG.error("The MQTT client ID cannot be empty., Username = {}, password = {}.", msg.getUsername(),
msg.getPassword());
return;
}
// Generating client id.
String randomIdentifier = UUID.randomUUID().toString().replace("-", "");
msg.setClientID(randomIdentifier);
LOG.info("Client connected with server generated identifier: {}", randomIdentifier);
LOG.info("The client has connected with a server generated identifier. MqttClientId = {}, username = {}, password = {}.",
randomIdentifier, msg.getUsername(), msg.getPassword());
}
if (!login(channel, msg)) {
@@ -237,9 +256,10 @@ public class ProtocolProcessor {
final String clientID = msg.getClientID();
ConnectionDescriptor descriptor = new ConnectionDescriptor(clientID, channel, msg.isCleanSession());
ConnectionDescriptor existing = this.connectionDescriptors.putIfAbsent(clientID, descriptor);
ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor);
if (existing != null) {
LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID());
LOG.info("The client ID is being used in an existing connection. It will be closed. MqttClientId = {}.",
msg.getClientID());
existing.abort();
return;
}
@@ -268,10 +288,10 @@ public class ProtocolProcessor {
final boolean success = descriptor.assignState(ConnectionState.MESSAGES_REPUBLISHED, ConnectionState.ESTABLISHED);
if (!success) {
channel.close();
} else {
LOG.info("Connection established");
}
LOG.info("CONNECT processed");
LOG.info("The CONNECT message has been processed. MqttClientId = {}, username = {}, password = {}.",
msg.getClientID(), msg.getUsername(), msg.getPassword());
}
private boolean login(Channel channel, ConnectMessage msg) {
@@ -281,15 +301,21 @@ public class ProtocolProcessor {
if (msg.isPasswordFlag()) {
pwd = msg.getPassword();
} else if (!this.allowAnonymous) {
LOG.error("The client didn't supply any password and MQTT anonymous mode is disabled. MqttClientId = {}.",
msg.getClientID());
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);
failedCredentials(channel);
return false;
}
NettyUtils.userName(channel, msg.getUsername());
} else if (!this.allowAnonymous) {
LOG.error("The client didn't supply any credentials and MQTT anonymous mode is disabled. MqttClientId = {}.",
msg.getClientID());
failedCredentials(channel);
return false;
}
@@ -297,6 +323,7 @@ public class ProtocolProcessor {
}
private boolean sendAck(ConnectionDescriptor descriptor, ConnectMessage msg) {
LOG.info("Sending connect ACK. MqttClientId = {}.", msg.getClientID());
final boolean success = descriptor.assignState(ConnectionState.DISCONNECTED, ConnectionState.SENDACK);
if (!success) {
return false;
@@ -310,36 +337,42 @@ public class ProtocolProcessor {
okResp.setSessionPresent(true);
}
if (isSessionAlreadyStored) {
LOG.info("Cleaning session. MqttClientId = {}.", msg.getClientID());
clientSession.cleanSession(msg.isCleanSession());
}
descriptor.channel.writeAndFlush(okResp);
descriptor.writeAndFlush(okResp);
LOG.info("The connect ACK has been sent. MqttClientId = {}.", msg.getClientID());
return true;
}
private void initializeKeepAliveTimeout(Channel channel, ConnectMessage msg) {
int keepAlive = msg.getKeepAlive();
LOG.debug("Connect with keepAlive {} s", keepAlive);
LOG.info("Configuring connection. MqttClientId = {}.", msg.getClientID());
NettyUtils.keepAlive(channel, keepAlive);
//session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession());
NettyUtils.cleanSession(channel, msg.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());
LOG.debug("Connect create session <{}>", channel);
int idleTime = Math.round(keepAlive * 1.5f);
setIdleTime(channel.pipeline(), idleTime);
setIdleTime(channel.pipeline(), Math.round(keepAlive * 1.5f));
LOG.info("The connection has been configured. MqttClientId = {}, keepAlive = {}, cleanSession = {}, idleTime = {}.",
msg.getClientID(), keepAlive, msg.isCleanSession(), idleTime);
}
private void storeWillMessage(ConnectMessage msg) {
//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();
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("Session for clientID <{}> with will to topic {}", msg.getClientID(), msg.getWillTopic());
LOG.info("MQTT last will and testament has been configured. MqttClientId = {}.", msg.getClientID());
}
}
@@ -352,13 +385,12 @@ public class ProtocolProcessor {
ClientSession clientSession = m_sessionsStore.sessionForClient(msg.getClientID());
boolean isSessionAlreadyStored = clientSession != null;
if (!isSessionAlreadyStored) {
LOG.debug("Create persistent session for clientID <{}>", msg.getClientID());
clientSession = m_sessionsStore.createNewSession(msg.getClientID(), msg.isCleanSession());
}
if (msg.isCleanSession()) {
LOG.info("Cleaning session. MqttClientId = {}.", msg.getClientID());
clientSession.cleanSession();
}
LOG.debug("Created session for client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
return clientSession;
}
@@ -373,7 +405,7 @@ public class ProtocolProcessor {
republishStoredInSession(clientSession);
}
int flushIntervalMs = 500/*(keepAlive * 1000) / 2*/;
setupAutoFlusher(descriptor.channel.pipeline(), flushIntervalMs);
descriptor.setupAutoFlusher(flushIntervalMs);
return true;
}
@@ -384,15 +416,6 @@ public class ProtocolProcessor {
LOG.info("Client {} failed to connect with bad username or password.", session);
}
private void setupAutoFlusher(ChannelPipeline pipeline, int flushIntervalMs) {
try {
pipeline.addAfter("idleEventHandler", "autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
} catch (NoSuchElementException nseex) {
//the idleEventHandler is not present on the pipeline
pipeline.addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
}
}
private void setIdleTime(ChannelPipeline pipeline, int idleTime) {
if (pipeline.names().contains("idleStateHandler")) {
pipeline.remove("idleStateHandler");
@@ -404,14 +427,13 @@ public class ProtocolProcessor {
* Republish QoS1 and QoS2 messages stored into the session for the clientID.
* */
private void republishStoredInSession(ClientSession clientSession) {
LOG.trace("republishStoredInSession for client <{}>", clientSession);
LOG.info("Republishing stored publish events. MqttClientId = {}.", clientSession.clientID);
BlockingQueue<StoredMessage> publishedEvents = clientSession.queue();
if (publishedEvents.isEmpty()) {
LOG.info("No stored messages for client <{}>", clientSession.clientID);
LOG.info("There are no stored publish events. ClientId = {}.", clientSession.clientID);
return;
}
LOG.info("republishing stored messages to client <{}>", clientSession.clientID);
this.internalRepublisher.publishStored(clientSession, publishedEvents);
}
@@ -448,7 +470,8 @@ public class ProtocolProcessor {
}
public void processPublish(Channel channel, PublishMessage msg) {
LOG.info("PUB --PUBLISH--> SRV executePublish invoked with {}", msg);
LOG.info("Processing PUBLISH message. MqttClientId = {}, topic = {}, messageId = {}, qos = {}.",
msg.getClientId(), msg.getTopicName(), msg.getMessageID(), msg.getQos());
final AbstractMessage.QOSType qos = msg.getQos();
switch (qos) {
case MOST_ONE:
@@ -460,6 +483,8 @@ public class ProtocolProcessor {
case EXACTLY_ONCE:
this.qos2PublishHandler.receivedPublishQos2(channel, msg);
break;
default:
break;
}
}
@@ -476,7 +501,7 @@ public class ProtocolProcessor {
public void internalPublish(PublishMessage msg) {
final AbstractMessage.QOSType qos = msg.getQos();
final String topic = msg.getTopicName();
LOG.info("embedded PUBLISH on topic <{}> with QoS {}", topic, qos);
LOG.info("Sending PUBLISH message. Topic = {}, qos = {}.", topic, qos);
MessageGUID guid = null;
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
@@ -524,6 +549,7 @@ public class ProtocolProcessor {
String topic = tobeStored.getTopic();
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
LOG.info("Publishing will message. MqttClientId = {}, messageId = {}, topic = {}.", clientID, messageId, topic);
this.messagesPublisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions);
}
@@ -551,7 +577,9 @@ public class ProtocolProcessor {
int messageID = msg.getMessageID();
targetSession.moveInFlightToSecondPhaseAckWaiting(messageID);
//once received a PUBREC reply with a PUBREL(messageID)
LOG.debug("\t\tSRV <--PUBREC-- SUB processPubRec invoked for clientID {} ad messageID {}", clientID, messageID);
if (LOG.isDebugEnabled()) {
LOG.debug("Processing PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID);
}
PubRelMessage pubRelMessage = new PubRelMessage();
pubRelMessage.setMessageID(messageID);
pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE);
@@ -562,7 +590,9 @@ public class ProtocolProcessor {
public void processPubComp(Channel channel, PubCompMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
if (LOG.isDebugEnabled()) {
LOG.debug("Processing PUBCOMP message. MqttClientId = {}, 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);
@@ -572,47 +602,55 @@ public class ProtocolProcessor {
}
public void processDisconnect(Channel channel) throws InterruptedException {
channel.flush();
final String clientID = NettyUtils.clientID(channel);
final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.get(clientID);
final String clientID = NettyUtils.clientID(channel);
LOG.info("Processing DISCONNECT message. MqttClientId = {}.", clientID);
channel.flush();
final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID);
if (existingDescriptor == null) {
//another client with same ID removed the descriptor, we must exit
channel.close();
return;
}
if (existingDescriptor.channel != channel) {
if (existingDescriptor.doesNotUseChannel(channel)) {
//another client saved it's descriptor, exit
channel.close();
LOG.warn("Another client is using the connection descriptor. Closing connection. MqttClientId = {}.",
clientID);
existingDescriptor.abort();
return;
}
if (!removeSubscriptions(existingDescriptor, clientID)) {
channel.close();
LOG.warn("Unable to remove subscriptions. Closing connection. MqttClientId = {}.", clientID);
existingDescriptor.abort();
return;
}
if (!dropStoredMessages(existingDescriptor, clientID)) {
channel.close();
LOG.warn("Unable to drop stored messages. Closing connection. MqttClientId = {}.", clientID);
existingDescriptor.abort();
return;
}
if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) {
channel.close();
LOG.warn("Unable to drop will message. Closing connection. MqttClientId = {}.", clientID);
existingDescriptor.abort();
return;
}
if (!closeChannel(existingDescriptor)) {
if (!existingDescriptor.close()) {
LOG.info("The connection has been closed. MqttClientId = {}.", clientID);
return;
}
boolean stillPresent = this.connectionDescriptors.remove(clientID, existingDescriptor);
boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor);
if (!stillPresent) {
//another descriptor was inserted
return;
LOG.warn("Another descriptor has been inserted. MqttClientId = {}.", clientID);
return;
}
LOG.info("DISCONNECT client <{}> finished", clientID);
LOG.info("The DISCONNECT message has been processed. MqttClientId = {}.", clientID);
}
private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) {
@@ -622,9 +660,9 @@ public class ProtocolProcessor {
}
if (descriptor.cleanSession) {
LOG.info("cleaning old saved subscriptions for client <{}>", clientID);
LOG.info("Removing saved subscriptions. MqttClientId = {}.", descriptor.clientID);
m_sessionsStore.wipeSubscriptions(clientID);
LOG.debug("Wiped subscriptions for client <{}>", clientID);
LOG.info("The saved subscriptions have been removed. MqttClientId = {}.", descriptor.clientID);
}
return true;
}
@@ -636,9 +674,9 @@ public class ProtocolProcessor {
}
if (descriptor.cleanSession) {
LOG.debug("Removing messages in session's queue for client <{}>", clientID);
LOG.debug("Removing messages of session. MqttClientId = {}.", descriptor.clientID);
this.m_sessionsStore.dropQueue(clientID);
LOG.debug("Removed messages in session for client's queue <{}>", clientID);
LOG.debug("The messages of the session have been removed. MqttClientId = {}.", descriptor.clientID);
}
return true;
}
@@ -649,25 +687,18 @@ public class ProtocolProcessor {
return false;
}
LOG.info("Removing will message. ClientId = {}.", descriptor.clientID);
//cleanup the will store
m_willStore.remove(clientID);
String username = NettyUtils.userName(descriptor.channel);
String username = descriptor.getUsername();
m_interceptor.notifyClientDisconnected(clientID, username);
return true;
}
private boolean closeChannel(ConnectionDescriptor descriptor) {
final boolean success = descriptor.assignState(ConnectionState.INTERCEPTORS_NOTIFIED, ConnectionState.DISCONNECTED);
if (!success) {
return false;
}
descriptor.channel.close();
return true;
}
public void processConnectionLost(String clientID, Channel channel) {
LOG.info("Processing connection lost event. MqttClientId = {}.", clientID);
ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true);
connectionDescriptors.remove(clientID, oldConnDescr);
connectionDescriptors.removeConnection(oldConnDescr);
//publish the Will message (if any) for the clientID
if (m_willStore.containsKey(clientID)) {
WillMessage will = m_willStore.get(clientID);
@@ -689,7 +720,7 @@ public class ProtocolProcessor {
List<String> topics = msg.topicFilters();
String clientID = NettyUtils.clientID(channel);
LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID);
LOG.info("Processing UNSUBSCRIBE message. MqttClientId = {}, topics = {}.", clientID, topics);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
for (String topic : topics) {
@@ -697,10 +728,14 @@ public class ProtocolProcessor {
if (!validTopic) {
//close the connection, not valid topicFilter is a protocol violation
channel.close();
LOG.warn("UNSUBSCRIBE found an invalid topic filter <{}> for clientID <{}>", topic, clientID);
LOG.error("The topic filter is not valid. MqttClientId = {}, topics = {}, badTopicFilter = {}.",
clientID, topics, topic);
return;
}
if (LOG.isDebugEnabled()) {
LOG.debug("Removing subscription. MqttClientId = {}, topic = {}.", clientID, topic);
}
subscriptions.removeSubscription(topic, clientID);
clientSession.unsubscribeFrom(topic);
String username = NettyUtils.userName(channel);
@@ -712,43 +747,45 @@ public class ProtocolProcessor {
UnsubAckMessage ackMessage = new UnsubAckMessage();
ackMessage.setMessageID(messageID);
LOG.info("replying with UnsubAck to MSG ID {}", messageID);
LOG.info("Sending UNSUBACK message. MqttClientId = {}, topics = {}, messageId = {}.", clientID, topics,
messageID);
channel.writeAndFlush(ackMessage);
}
public void processSubscribe(Channel channel, SubscribeMessage msg) {
String clientID = NettyUtils.clientID(channel);
LOG.info("SUBSCRIBE client <{}>", clientID);
int messageID = msg.getMessageID();
LOG.debug("SUBSCRIBE client <{}> on server {} packetID {}", clientID, m_server_port, messageID);
LOG.info("Processing SUBSCRIBE message. MqttClientId = {}, messageId = {}.", clientID, msg.getMessageID());
RunningSubscription executionKey = new RunningSubscription(clientID, messageID);
SubscriptionState currentStatus = this.subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED);
if (currentStatus != null) {
LOG.debug("The client <{}> sent another SUBSCRIBE while this one was processing", clientID);
LOG.warn("The client sent another SUBSCRIBE message while this one was being processed. MqttClientId = {}, messageId = {}.",
clientID, msg.getMessageID());
return;
}
String username = NettyUtils.userName(channel);
List<SubscribeMessage.Couple> ackTopics = doVerify(clientID, username, msg);
SubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics);
if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) {
LOG.debug("The client {} sent another SUBSCRIBE while this one was verifing topicFilters");
LOG.warn("The client sent another SUBSCRIBE message while the topic filters were being verified. MqttClientId = {}, messageId = {}.",
clientID, msg.getMessageID());
return;
}
LOG.info("Creating and storing subscriptions. MqttClientId = {}, messageId = {}, topics = {}.", clientID,
msg.getMessageID(), ackTopics);
ackMessage.setMessageID(messageID);
List<Subscription> newSubscriptions = doStoreSubscription(ackTopics, clientID);
//save session, persist subscriptions from session
LOG.debug("SUBACK for packetID {}", messageID);
if (LOG.isTraceEnabled()) {
LOG.trace("subscription tree {}", subscriptions.dumpTree());
}
for (Subscription subscription : newSubscriptions) {
LOG.debug("Persisting subscription {}", subscription);
subscriptions.add(subscription.asClientTopicCouple());
}
LOG.info("Sending SUBACK response. MqttClientId = {}, messageId = {}.", clientID, msg.getMessageID());
channel.writeAndFlush(ackMessage);
//fire the persisted messages in session
@@ -758,7 +795,8 @@ public class ProtocolProcessor {
boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED);
if (!success) {
LOG.warn("Failed to remove the descriptor, something bad happened");
LOG.warn("Unable to perform the final subscription state update. MqttClientId = {}, messageId = {}.",
clientID, msg.getMessageID());
}
}
@@ -791,13 +829,20 @@ public class ProtocolProcessor {
for (SubscribeMessage.Couple req : msg.subscriptions()) {
if (!m_authorizator.canRead(req.topicFilter, username, clientSession.clientID)) {
//send SUBACK with 0x80, the user hasn't credentials to read the topic
LOG.debug("topic {} doesn't have read credentials", req.topicFilter);
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));
} else {
boolean validTopic = SubscriptionsStore.validate(req.topicFilter);
AbstractMessage.QOSType qos = validTopic ?
AbstractMessage.QOSType.valueOf(req.qos)
: AbstractMessage.QOSType.FAILURE;
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);
} else {
LOG.error("The topic filter is not valid. MqttClientId = {}, username = {}, messageId = {}, topic = {}.",
clientID, username, msg.getMessageID(), req.topicFilter);
qos = AbstractMessage.QOSType.FAILURE;
}
ackTopics.add(new SubscribeMessage.Couple(qos.byteValue(), req.topicFilter));
}
}
@@ -817,7 +862,8 @@ public class ProtocolProcessor {
}
private void publishRetainedMessagesInSession(final Subscription newSubscription, String username) {
LOG.debug("Publish persisted messages in session {}", newSubscription);
LOG.info("Retrieving retained messages. MqttClientId = {}, topics = {}.", newSubscription.getClientId(),
newSubscription.getTopicFilter());
//scans retained messages to be published to the new subscription
//TODO this is ugly, it does a linear scan on potential big dataset
@@ -828,7 +874,10 @@ public class ProtocolProcessor {
}
});
LOG.debug("Found {} messages to republish", messages.size());
if (!messages.isEmpty()) {
LOG.info("Publishing retained messages. MqttClientId = {}, topics = {}, messagesNo = {}.",
newSubscription.getClientId(), newSubscription.getTopicFilter(), messages.size());
}
ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId());
this.internalRepublisher.publishRetained(targetSession, messages);
@@ -854,11 +903,11 @@ public class ProtocolProcessor {
channel.flush();
}
public boolean addInterceptHandler(InterceptHandler interceptHandler) {
return this.m_interceptor.addInterceptHandler(interceptHandler);
public void addInterceptHandler(InterceptHandler interceptHandler) {
this.m_interceptor.addInterceptHandler(interceptHandler);
}
public boolean removeInterceptHandler(InterceptHandler interceptHandler) {
return this.m_interceptor.removeInterceptHandler(interceptHandler);
public void removeInterceptHandler(InterceptHandler interceptHandler) {
this.m_interceptor.removeInterceptHandler(interceptHandler);
}
}
@@ -17,6 +17,7 @@ package io.moquette.spi.impl;
import io.moquette.BrokerConstants;
import io.moquette.interception.InterceptHandler;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.Server;
import io.moquette.server.config.IConfig;
import io.moquette.server.config.IResourceLoader;
@@ -35,7 +36,6 @@ import io.moquette.spi.security.IAuthorizator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.text.ParseException;
@@ -60,6 +60,8 @@ public class ProtocolProcessorBootstrapper {
private BrokerInterceptor m_interceptor;
private final ProtocolProcessor m_processor = new ProtocolProcessor();
private ConnectionDescriptorStore connectionDescriptors;
public ProtocolProcessorBootstrapper() {
}
@@ -81,37 +83,32 @@ public class ProtocolProcessorBootstrapper {
IAuthenticator authenticator, IAuthorizator authorizator, Server server) {
subscriptions = new SubscriptionsStore();
LOG.info("Initializing messages and sessions stores...");
m_mapStorage = new MapDBPersistentStore(props);
m_mapStorage.initStore();
IMessagesStore messagesStore = m_mapStorage.messagesStore();
m_sessionsStore = m_mapStorage.sessionsStore();
LOG.info("Configuring message interceptors...");
List<InterceptHandler> observers = new ArrayList<>(embeddedObservers);
String interceptorClassName = props.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
if (interceptorClassName != null && !interceptorClassName.isEmpty()) {
try {
InterceptHandler handler;
try {
final Constructor<? extends InterceptHandler> constructor = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).getConstructor(Server.class);
handler = constructor.newInstance(server);
} catch (NoSuchMethodException nsme){
handler = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).newInstance();
InterceptHandler handler = loadClass(interceptorClassName, InterceptHandler.class, Server.class, server);
if (handler != null) {
observers.add(handler);
}
observers.add(handler);
} catch (Throwable ex) {
LOG.error("Can't load the intercept handler {}", ex);
}
}
m_interceptor = new BrokerInterceptor(observers);
m_interceptor = new BrokerInterceptor(props, observers);
LOG.info("Initializing subscriptions store...");
subscriptions.init(m_sessionsStore);
String configPath = System.getProperty("moquette.path", null);
LOG.info("Configuring MQTT authenticator...");
String authenticatorClassName = props.getProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, "");
if (!authenticatorClassName.isEmpty()) {
authenticator = (IAuthenticator)loadClass(authenticatorClassName, IAuthenticator.class, props);
LOG.info("Loaded custom authenticator {}", authenticatorClassName);
if (authenticator == null && !authenticatorClassName.isEmpty()) {
authenticator = loadClass(authenticatorClassName, IAuthenticator.class, IConfig.class, props);
}
IResourceLoader resourceLoader = props.getResourceLoader();
@@ -122,12 +119,13 @@ public class ProtocolProcessorBootstrapper {
} else {
authenticator = new ResourceAuthenticator(resourceLoader, passwdPath);
}
LOG.info("An {} authenticator instance will be used.", authenticator.getClass().getName());
}
LOG.info("Configuring MQTT authorizator...");
String authorizatorClassName = props.getProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, "");
if (!authorizatorClassName.isEmpty()) {
authorizator = (IAuthorizator)loadClass(authorizatorClassName, IAuthorizator.class, props);
LOG.info("Loaded custom authorizator {}", authorizatorClassName);
if (authorizator == null && !authorizatorClassName.isEmpty()) {
authorizator = loadClass(authorizatorClassName, IAuthorizator.class, IConfig.class, props);
}
if (authorizator == null) {
@@ -135,70 +133,85 @@ public class ProtocolProcessorBootstrapper {
if (aclFilePath != null && !aclFilePath.isEmpty()) {
authorizator = new DenyAllAuthorizator();
try {
LOG.info("Parsing ACL file. Path = {}.", aclFilePath);
authorizator = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath));
} catch (ParseException pex) {
LOG.error(String.format("Format error in parsing acl %s %s", resourceLoader.getName(), aclFilePath), pex);
LOG.error("Unable to parse ACL file. Path = {}, cause = {}, errorMessage = {}.", aclFilePath,
pex.getCause(), pex.getMessage());
}
LOG.info("Using acl file defined at path {}", aclFilePath);
} else {
authorizator = new PermitAllAuthorizator();
LOG.info("Starting without ACL definition");
}
LOG.info("An {} authorizator instance will be used.", authorizator.getClass().getName());
}
LOG.info("Initializing connection descriptor store...");
connectionDescriptors = new ConnectionDescriptorStore(m_sessionsStore);
LOG.info("Initializing MQTT protocol processor...");
boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
boolean allowZeroByteClientId = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false"));
m_processor.init(subscriptions, messagesStore, m_sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
return m_processor;
}
private Object loadClass(String className, Class<?> cls, IConfig props) {
Object instance = null;
try {
Class<?> clazz = Class.forName(className);
@SuppressWarnings("unchecked")
private <T, U> T loadClass(String className, Class<T> iface, Class<U> constructorArgClass, U props) {
T instance = null;
try {
LOG.info("Loading class. ClassName = {}, interfaceName = {}.", className, iface.getName());
Class<?> clazz = Class.forName(className);
// check if method getInstance exists
Method method = clazz.getMethod("getInstance", new Class[] {});
try {
instance = method.invoke(null, new Object[] {});
} catch (IllegalArgumentException | InvocationTargetException | IllegalAccessException ex) {
LOG.error(null, ex);
throw new RuntimeException("Cannot call method "+ className +".getInstance", ex);
}
}
catch (NoSuchMethodException nsmex) {
try {
// check if constructor with IConfig parameter exists
instance = this.getClass().getClassLoader()
.loadClass(className)
.asSubclass(cls)
.getConstructor(IConfig.class).newInstance(props);
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
LOG.error(null, ex);
throw new RuntimeException("Cannot load custom authenticator class " + className, ex);
} catch (NoSuchMethodException | InvocationTargetException e) {
try {
// fallback to default constructor
instance = this.getClass().getClassLoader()
.loadClass(className)
.asSubclass(cls)
.newInstance();
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
LOG.error(null, ex);
throw new RuntimeException("Cannot load custom authenticator class " + className, ex);
}
}
} catch (ClassNotFoundException ex) {
LOG.error(null, ex);
throw new RuntimeException("Class " + className + " not found", ex);
} catch (SecurityException ex) {
LOG.error(null, ex);
throw new RuntimeException("Cannot call method "+ className +".getInstance", ex);
}
// check if method getInstance exists
Method method = clazz.getMethod("getInstance", new Class[] {});
try {
LOG.info("Invoking getInstance() method. ClassName = {}, interfaceName = {}.", className,
iface.getName());
instance = (T) method.invoke(null, new Object[] {});
} catch (IllegalArgumentException | InvocationTargetException | IllegalAccessException ex) {
LOG.error(
"Unable to invoke getInstance() method. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
className, iface.getName(), ex.getCause(), ex.getMessage());
return null;
}
} catch (NoSuchMethodException nsmex) {
try {
// check if constructor with constructor arg class parameter
// exists
LOG.info("Invoking constructor with {} argument. ClassName = {}, interfaceName = {}.",
constructorArgClass.getName(), className, iface.getName());
instance = this.getClass().getClassLoader().loadClass(className).asSubclass(iface)
.getConstructor(constructorArgClass).newInstance(props);
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
LOG.warn(
"Unable to invoke constructor with {} argument. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
constructorArgClass.getName(), className, iface.getName(), ex.getCause(), ex.getMessage());
return null;
} catch (NoSuchMethodException | InvocationTargetException e) {
try {
LOG.info("Invoking default constructor. ClassName = {}, interfaceName = {}.", className,
iface.getName());
// fallback to default constructor
instance = this.getClass().getClassLoader().loadClass(className).asSubclass(iface).newInstance();
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
LOG.error(
"Unable to invoke default constructor. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
className, iface.getName(), ex.getCause(), ex.getMessage());
return null;
}
}
} catch (ClassNotFoundException ex) {
LOG.error("The class does not exist. ClassName = {}, interfaceName = {}.", className, iface.getName());
return null;
} catch (SecurityException ex) {
LOG.error(
"Unable to load class due to a security violation. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
className, iface.getName(), ex.getCause(), ex.getMessage());
return null;
}
return instance;
}
return instance;
}
public List<Subscription> getSubscriptions() {
return m_sessionsStore.getSubscriptions();
@@ -207,4 +220,8 @@ public class ProtocolProcessorBootstrapper {
public void shutdown() {
this.m_mapStorage.close();
}
public ConnectionDescriptorStore getConnectionDescriptors() {
return connectionDescriptors;
}
}
@@ -14,11 +14,10 @@ import java.util.List;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
class Qos0PublishHandler {
class Qos0PublishHandler extends QosPublishHandler {
private static final Logger LOG = LoggerFactory.getLogger(Qos0PublishHandler.class);
private final IAuthorizator m_authorizator;
private final SubscriptionsStore subscriptions;
private final IMessagesStore m_messagesStore;
private final BrokerInterceptor m_interceptor;
@@ -27,7 +26,7 @@ class Qos0PublishHandler {
public Qos0PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor,
MessagesPublisher messagesPublisher) {
this.m_authorizator = authorizator;
super(authorizator);
this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore;
this.m_interceptor = interceptor;
@@ -46,11 +45,15 @@ class Qos0PublishHandler {
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic);
if (LOG.isTraceEnabled()) {
LOG.trace("content <{}>", DebugUtils.payload2Str(toStoreMsg.getMessage()));
LOG.trace("subscription tree {}", subscriptions.dumpTree());
}
if (LOG.isDebugEnabled()) {
LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID,
topic, msg.getMessageID());
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
@@ -62,14 +65,4 @@ class Qos0PublishHandler {
String username = NettyUtils.userName(channel);
m_interceptor.notifyTopicPublished(msg, clientID, username);
}
boolean checkWriteOnTopic(String topic, Channel channel) {
String clientID = NettyUtils.clientID(channel);
String username = NettyUtils.userName(channel);
if (!m_authorizator.canWrite(topic, username, clientID)) {
LOG.debug("topic {} doesn't have write credentials", topic);
return true;
}
return false;
}
}
@@ -2,7 +2,7 @@ package io.moquette.spi.impl;
import io.moquette.parser.proto.messages.PubAckMessage;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.MessageGUID;
@@ -14,31 +14,27 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
class Qos1PublishHandler {
class Qos1PublishHandler extends QosPublishHandler {
private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class);
private final IAuthorizator m_authorizator;
private final SubscriptionsStore subscriptions;
private final IMessagesStore m_messagesStore;
private final BrokerInterceptor m_interceptor;
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
private final String brokerPort;
private final ConnectionDescriptorStore connectionDescriptors;
private final MessagesPublisher publisher;
public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors,
ConnectionDescriptorStore connectionDescriptors,
String brokerPort, MessagesPublisher messagesPublisher) {
this.m_authorizator = authorizator;
super(authorizator);
this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore;
this.m_interceptor = interceptor;
this.connectionDescriptors = connectionDescriptors;
this.brokerPort = brokerPort;
this.publisher = messagesPublisher;
}
@@ -54,11 +50,15 @@ class Qos1PublishHandler {
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
LOG.debug("publish2Subscribers_qos1 republishing to existing subscribers that matches the topic {}", topic);
if (LOG.isTraceEnabled()) {
LOG.trace("content <{}>", DebugUtils.payload2Str(toStoreMsg.getMessage()));
LOG.trace("subscription tree {}", subscriptions.dumpTree());
}
if (LOG.isDebugEnabled()) {
LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID,
topic, msg.getMessageID());
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
@@ -67,7 +67,6 @@ class Qos1PublishHandler {
if (msg.isLocal()) {
sendPubAck(clientID, messageID);
}
LOG.info("server {} replying with PubAck to MSG ID {}", brokerPort, messageID);
if (msg.isRetainFlag()) {
if (!msg.getPayload().hasRemaining()) {
@@ -83,16 +82,6 @@ class Qos1PublishHandler {
m_interceptor.notifyTopicPublished(msg, clientID, username);
}
boolean checkWriteOnTopic(String topic, Channel channel) {
String clientID = NettyUtils.clientID(channel);
String username = NettyUtils.userName(channel);
if (!m_authorizator.canWrite(topic, username, clientID)) {
LOG.debug("topic {} doesn't have write credentials", topic);
return true;
}
return false;
}
private void sendPubAck(String clientId, int messageID) {
LOG.trace("sendPubAck invoked");
PubAckMessage pubAckMessage = new PubAckMessage();
@@ -103,10 +92,10 @@ class Qos1PublishHandler {
throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be initialized, somewhere it's overwritten!!");
}
LOG.debug("clientIDs are {}", connectionDescriptors);
if (connectionDescriptors.get(clientId) == null) {
if (!connectionDescriptors.isConnected(clientId)) {
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, connectionDescriptors));
}
connectionDescriptors.get(clientId).channel.writeAndFlush(pubAckMessage);
connectionDescriptors.sendMessage(pubAckMessage, messageID, clientId);
} catch(Throwable t) {
LOG.error(null, t);
}
@@ -1,7 +1,7 @@
package io.moquette.spi.impl;
import io.moquette.parser.proto.messages.*;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
@@ -14,40 +14,36 @@ import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
class Qos2PublishHandler {
import java.util.List;
class Qos2PublishHandler extends QosPublishHandler {
private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class);
private final IAuthorizator m_authorizator;
private final SubscriptionsStore subscriptions;
private final IMessagesStore m_messagesStore;
private final BrokerInterceptor m_interceptor;
private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
private final ConnectionDescriptorStore connectionDescriptors;
private final ISessionsStore m_sessionsStore;
private final String brokerPort;
private final MessagesPublisher publisher;
public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors,
ConnectionDescriptorStore connectionDescriptors,
ISessionsStore sessionsStore, String brokerPort, MessagesPublisher messagesPublisher) {
this.m_authorizator = authorizator;
super(authorizator);
this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore;
this.m_interceptor = interceptor;
this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore;
this.brokerPort = brokerPort;
this.publisher = messagesPublisher;
}
void receivedPublishQos2(Channel channel, PublishMessage msg) {
final AbstractMessage.QOSType qos = AbstractMessage.QOSType.EXACTLY_ONCE;
final String topic = msg.getTopicName();
//check if the topic can be wrote
if (checkWriteOnTopic(topic, channel)) {
@@ -58,7 +54,16 @@ class Qos2PublishHandler {
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
LOG.info("PUBLISH on server {} from clientID <{}> on topic <{}> with QoS {}", this.brokerPort, clientID, topic, qos);
if (LOG.isDebugEnabled()) {
LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID,
topic, msg.getMessageID());
}
//QoS2
MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg);
if (msg.isLocal()) {
@@ -85,16 +90,20 @@ class Qos2PublishHandler {
void processPubRel(Channel channel, PubRelMessage msg) {
String clientID = NettyUtils.clientID(channel);
int messageID = msg.getMessageID();
LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
LOG.info("Processing PUBREL message. MqttClientId = {}, messageId = {}.", clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID);
final String topic = evt.getTopic();
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic);
if (LOG.isTraceEnabled()) {
LOG.trace("content <{}>", DebugUtils.payload2Str(evt.getMessage()));
LOG.trace("subscription tree {}", subscriptions.dumpTree());
}
if (LOG.isDebugEnabled()) {
LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(evt.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID,
topic, msg.getMessageID());
}
this.publisher.publish2Subscribers(evt, topicMatchingSubscriptions);
if (evt.isRetained()) {
@@ -108,27 +117,22 @@ class Qos2PublishHandler {
sendPubComp(clientID, messageID);
}
private boolean checkWriteOnTopic(String topic, Channel channel) {
String clientID = NettyUtils.clientID(channel);
String username = NettyUtils.userName(channel);
if (!m_authorizator.canWrite(topic, username, clientID)) {
LOG.debug("topic {} doesn't have write credentials", topic);
return true;
}
return false;
}
private void sendPubRec(String clientID, int messageID) {
LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID);
if (LOG.isDebugEnabled()) {
LOG.debug("Sending PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID);
}
PubRecMessage pubRecMessage = new PubRecMessage();
pubRecMessage.setMessageID(messageID);
connectionDescriptors.get(clientID).channel.writeAndFlush(pubRecMessage);
connectionDescriptors.sendMessage(pubRecMessage,messageID, clientID);
}
private void sendPubComp(String clientID, int messageID) {
LOG.debug("PUB <--PUBCOMP-- SRV sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
if (LOG.isDebugEnabled()) {
LOG.debug("Sending PUBCOMP message. MqttClientId = {}, messageId = {}.", clientID, messageID);
}
PubCompMessage pubCompMessage = new PubCompMessage();
pubCompMessage.setMessageID(messageID);
connectionDescriptors.get(clientID).channel.writeAndFlush(pubCompMessage);
connectionDescriptors.sendMessage(pubCompMessage, messageID, clientID);
}
}
@@ -0,0 +1,30 @@
package io.moquette.spi.impl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
abstract class QosPublishHandler {
private static final Logger LOG = LoggerFactory.getLogger(QosPublishHandler.class);
private final IAuthorizator m_authorizator;
protected QosPublishHandler(IAuthorizator m_authorizator) {
this.m_authorizator = m_authorizator;
}
public boolean checkWriteOnTopic(String topic, Channel channel) {
String clientID = NettyUtils.clientID(channel);
String username = NettyUtils.userName(channel);
if (!m_authorizator.canWrite(topic, username, clientID)) {
LOG.error("The MQTT client is not authorized to publish on topic. MqttClientId = {}, topic = {}.", clientID,
topic);
return true;
}
return false;
}
}
@@ -35,9 +35,9 @@ class AuthorizationsCollector implements IAuthorizator {
private static final Logger LOG = LoggerFactory.getLogger(AuthorizationsCollector.class);
private List<Authorization> m_globalAuthorizations = new ArrayList();
private List<Authorization> m_patternAuthorizations = new ArrayList();
private Map<String, List<Authorization>> m_userAuthorizations = new HashMap();
private List<Authorization> m_globalAuthorizations = new ArrayList<>();
private List<Authorization> m_patternAuthorizations = new ArrayList<>();
private Map<String, List<Authorization>> m_userAuthorizations = new HashMap<>();
private boolean m_parsingUsersSpecificSection = false;
private boolean m_parsingPatternSpecificSection = false;
private String m_currentUser = "";
@@ -59,7 +59,7 @@ class AuthorizationsCollector implements IAuthorizator {
if (m_parsingUsersSpecificSection) {
//TODO in java 8 switch to m_userAuthorizations.putIfAbsent(m_currentUser, new ArrayList());
if (!m_userAuthorizations.containsKey(m_currentUser)) {
m_userAuthorizations.put(m_currentUser, new ArrayList());
m_userAuthorizations.put(m_currentUser, new ArrayList<Authorization>());
}
List<Authorization> userAuths = m_userAuthorizations.get(m_currentUser);
userAuths.add(acl);
@@ -22,7 +22,6 @@ public class DBAuthenticator implements IAuthenticator {
private static final Logger LOG = LoggerFactory.getLogger(DBAuthenticator.class);
private final String sqlQuery;
private final MessageDigest messageDigest;
private final PreparedStatement preparedStatement;
@@ -44,7 +43,6 @@ public class DBAuthenticator implements IAuthenticator {
*/
public DBAuthenticator(String driver, String jdbcUrl, String sqlQuery, String digestMethod){
this.sqlQuery = sqlQuery;
try {
Class.forName(driver);
final Connection connection = DriverManager.getConnection(jdbcUrl);
@@ -17,7 +17,6 @@ package io.moquette.spi.impl.security;
import io.moquette.server.config.IResourceLoader;
import io.moquette.spi.security.IAuthenticator;
import org.apache.commons.codec.binary.Hex;
import org.apache.commons.codec.digest.DigestUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -28,7 +28,8 @@ import io.moquette.spi.ISessionsStore.ClientTopicCouple;
*/
public final class Subscription implements Serializable {
final QOSType requestedQos; //max QoS acceptable
private static final long serialVersionUID = 1L;
final QOSType requestedQos; //max QoS acceptable
final String clientId;
final String topicFilter;
final boolean active;
@@ -59,6 +60,10 @@ public final class Subscription implements Serializable {
return topicFilter;
}
public boolean isActive() {
return active;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
@@ -51,10 +51,11 @@ public class SubscriptionsStore {
try {
parseTopic(topicFilter);
return true;
} catch (ParseException pex) {
LOG.info("Bad matching topic filter <{}>", topicFilter);
return false;
}
} catch (ParseException pex) {
LOG.warn("The topic filter is malformed. TopicFilter = {}, cause = {}, errorMessage = {}.", topicFilter,
pex.getCause(), pex.getMessage());
return false;
}
}
public interface IVisitor<T> {
@@ -103,24 +104,27 @@ public class SubscriptionsStore {
* @param sessionsStore to be used as backing store from the subscription store.
*/
public void init(ISessionsStore sessionsStore) {
LOG.debug("init invoked");
LOG.info("Initializing subscriptions store...");
m_sessionsStore = sessionsStore;
List<ClientTopicCouple> subscriptions = sessionsStore.listAllSubscriptions();
//reload any subscriptions persisted
if (LOG.isDebugEnabled()) {
LOG.debug("Reloading all stored subscriptions...subscription tree before {}", dumpTree());
LOG.debug("Reloading all stored subscriptions. SubscriptionTree = {}.", dumpTree());
}
for (ClientTopicCouple clientTopic : subscriptions) {
LOG.debug("Re-subscribing {} to topic {}", clientTopic.clientID, clientTopic.topicFilter);
LOG.info("Re-subscribing client to topic. ClientId = {}, topicFilter = {}.", clientTopic.clientID, clientTopic.topicFilter);
add(clientTopic);
}
if (LOG.isTraceEnabled()) {
LOG.trace("Finished loading. Subscription tree after {}", dumpTree());
if (LOG.isDebugEnabled()) {
LOG.debug("The stored subscriptions have been reloaded. SubscriptionTree = {}.", dumpTree());
}
}
public void add(ClientTopicCouple newSubscription) {
/*
* The topic filters have already been validated at the ProtocolProcessor. We can assume they are valid.
*/
TreeNode oldRoot;
NodeCouple couple;
do {
@@ -129,7 +133,9 @@ public class SubscriptionsStore {
couple.createdNode.addSubscription(newSubscription); //createdNode could be null?
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, couple.root));
LOG.debug("root ref {}, original root was {}", couple.root, oldRoot);
if (LOG.isDebugEnabled()) {
LOG.debug("A subscription has been added. Root = {}, oldRoot = {}.", couple.root, oldRoot);
}
}
@@ -139,7 +145,9 @@ public class SubscriptionsStore {
tokens = parseTopic(topic);
} catch (ParseException ex) {
//TODO handle the parse exception
LOG.error(null, ex);
LOG.error("The topic is malformed. Topic = {}, cause = {}, errorMessage = {}.", topic, ex.getCause(),
ex.getMessage());
throw new IllegalArgumentException(ex.getMessage());
}
final TreeNode newRoot = oldRoot.copy();
@@ -167,6 +175,9 @@ public class SubscriptionsStore {
}
public void removeSubscription(String topic, String clientID) {
/*
* The topic filters have already been validated at the ProtocolProcessor. We can assume they are valid.
*/
TreeNode oldRoot;
NodeCouple couple;
do {
@@ -208,7 +219,7 @@ public class SubscriptionsStore {
tokens = parseTopic(topic);
} catch (ParseException ex) {
//TODO handle the parse exception
LOG.error(null, ex);
LOG.warn("The topic is malformed. Topic = {}, cause = {}, errorMessage = {}.", topic, ex.getCause(), ex.getMessage());
return Collections.emptyList();
}
@@ -247,7 +258,7 @@ public class SubscriptionsStore {
return visitor.getResult();
}
private void bfsVisit(TreeNode node, IVisitor visitor, int deep) {
private void bfsVisit(TreeNode node, IVisitor<?> visitor, int deep) {
if (node == null) {
return;
}
@@ -294,8 +305,10 @@ public class SubscriptionsStore {
// }
return i == msgTokens.size();
} catch (ParseException ex) {
LOG.error(null, ex);
throw new RuntimeException(ex);
LOG.error(
"The message topic, the subscription topic or both are malformed. MsgTopic = {}, subscriptionTopic = {}, cause = {}, errorMessage = {}.",
msgTopic, subscriptionTopic, ex.getCause(), ex.getMessage());
throw new IllegalStateException(ex.getMessage());
}
}
@@ -48,18 +48,24 @@ class MapDBMessagesStore implements IMessagesStore {
@Override
public void initStore() {
LOG.info("Initializing store...");
m_retainedStore = m_db.getHashMap("retained");
m_persistentMessageStore = m_db.getHashMap("persistedMessages");
}
@Override
public void storeRetained(String topic, MessageGUID guid) {
if (LOG.isDebugEnabled()) {
LOG.debug("Storing retained messages. Topic = {}, guid = {}.", topic, guid);
}
m_retainedStore.put(topic, guid);
}
@Override
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
if (LOG.isDebugEnabled()) {
LOG.debug("Scanning retained messages...");
}
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, MessageGUID> entry : m_retainedStore.entrySet()) {
@@ -70,19 +76,22 @@ class MapDBMessagesStore implements IMessagesStore {
}
}
if (LOG.isDebugEnabled()) {
LOG.debug("The retained messages have been scanned. MatchingMessages = {}.", results);
}
return results;
}
@Override
public MessageGUID storePublishForFuture(StoredMessage storedMessage) {
LOG.debug("storePublishForFuture store evt {}", storedMessage);
if (storedMessage.getClientID() == null) {
LOG.error("persisting a message without a clientID, bad programming error msg: {}", storedMessage);
throw new IllegalArgumentException("persisting a message without a clientID, bad programming error");
}
assert storedMessage.getClientID() != null : "The message to be persisted must have a valid client ID";
MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
storedMessage.setGuid(guid);
LOG.debug("storePublishForFuture guid <{}>", guid);
if (LOG.isDebugEnabled()) {
LOG.debug("Storing publish event. MqttClientId = {}, messageId = {}, guid = {}, topic = {}.", storedMessage.getClientID(),
storedMessage.getMessageID(), guid, storedMessage.getTopic());
}
m_persistentMessageStore.put(guid, storedMessage);
ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(storedMessage.getClientID()));
messageIdToGuid.put(storedMessage.getMessageID(), guid);
@@ -91,6 +100,9 @@ class MapDBMessagesStore implements IMessagesStore {
@Override
public void dropMessagesInSession(String clientID) {
if (LOG.isDebugEnabled()) {
LOG.debug("Dropping stored messages. ClientId = {}.", clientID);
}
ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID));
for (MessageGUID guid : messageIdToGuid.values()) {
removeStoredMessage(guid);
@@ -102,18 +114,34 @@ class MapDBMessagesStore implements IMessagesStore {
//remove only the not retained and no more referenced
StoredMessage storedMessage = m_persistentMessageStore.get(guid);
if (!storedMessage.isRetained()) {
LOG.debug("Cleaning not retained message guid {}", guid);
if (LOG.isDebugEnabled()) {
LOG.debug("Dropping stored message. ClientId = {}, messageId = {}, guid = {}, topic = {}.",
storedMessage.getClientID(), storedMessage.getMessageID(), guid, storedMessage.getTopic());
}
m_persistentMessageStore.remove(guid);
}
}
@Override
public StoredMessage getMessageByGuid(MessageGUID guid) {
if (LOG.isDebugEnabled()) {
LOG.debug("Retrieving stored message. Guid = {}.", guid);
}
return m_persistentMessageStore.get(guid);
}
@Override
public void cleanRetained(String topic) {
if (LOG.isDebugEnabled()) {
LOG.debug("Cleaning retained messages. Topic = {}.", topic);
}
m_retainedStore.remove(topic);
}
@Override
public int getPendingPublishMessages(String clientID) {
ConcurrentMap<Integer, MessageGUID> messageIdToGuidMap = m_db
.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID));
return messageIdToGuidMap.size();
}
}
@@ -45,7 +45,8 @@ public class MapDBPersistentStore {
* a session.
* */
public static class PersistentSession implements Serializable {
public final boolean cleanSession;
private static final long serialVersionUID = 1L;
public final boolean cleanSession;
public PersistentSession(boolean cleanSession) {
this.cleanSession = cleanSession;
@@ -80,23 +81,30 @@ public class MapDBPersistentStore {
}
public void initStore() {
LOG.info("Initializing MapDB store...");
if (m_storePath == null || m_storePath.isEmpty()) {
LOG.warn("The MapDB store file path is empty. Using in-memory store.");
m_db = DBMaker.newMemoryDB().make();
} else {
File tmpFile;
try {
LOG.info("Using user-defined MapDB store file. Path = {}.", m_storePath);
tmpFile = new File(m_storePath);
boolean fileNewlyCreated = tmpFile.createNewFile();
LOG.info("Starting with {} [{}] db file", fileNewlyCreated ? "fresh" : "existing", m_storePath);
LOG.warn("Using {} MapDB store file. Path = {}.", fileNewlyCreated ? "fresh" : "existing", m_storePath);
} catch (IOException ex) {
LOG.error(null, 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);
}
m_db = DBMaker.newFileDB(tmpFile).make();
}
LOG.info("Scheduling MapDB commit task...");
m_scheduler.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
if (LOG.isDebugEnabled()) {
LOG.debug("Committing to MapDB...");
}
m_db.commit();
}
}, this.m_autosaveInterval, this.m_autosaveInterval, TimeUnit.SECONDS);
@@ -111,14 +119,22 @@ public class MapDBPersistentStore {
public void close() {
if (this.m_db.isClosed()) {
LOG.debug("already closed");
LOG.warn("The MapDB store is already closed. Nothing will be done.");
return;
}
LOG.info("Performing last commit to MapDB...");
this.m_db.commit();
//LOG.debug("persisted subscriptions {}", m_persistentSubscriptions);
LOG.info("Closing MapDB store...");
this.m_db.close();
LOG.debug("closed disk storage");
LOG.info("Stopping MapDB commit tasks...");
this.m_scheduler.shutdown();
LOG.debug("Persistence commit scheduler is shutdown");
try {
m_scheduler.awaitTermination(10L, TimeUnit.SECONDS);
} catch (InterruptedException e) {}
if (!m_scheduler.isTerminated()) {
LOG.warn("Forcing shutdown of MapDB commit tasks...");
m_scheduler.shutdown();
}
LOG.info("The MapDB store has been closed successfully.");
}
}
@@ -16,7 +16,6 @@
package io.moquette.spi.persistence;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.MessageGUID;
@@ -67,38 +66,44 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public void addNewSubscription(Subscription newSubscription) {
LOG.debug("addNewSubscription invoked with subscription {}", newSubscription);
LOG.info("Adding new subscription. MqttClientId = {}, topics = {}.", newSubscription.getClientId(),
newSubscription.getTopicFilter());
final String clientID = newSubscription.getClientId();
m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription);
if (LOG.isTraceEnabled()) {
LOG.trace("subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID));
if (LOG.isDebugEnabled()) {
LOG.debug("The subscription has been added. MqttClientId = {}, topics = {}, clientSubscriptions = {}.",
newSubscription.getClientId(), newSubscription.getTopicFilter(),
m_db.getHashMap("subscriptions_" + clientID));
}
}
@Override
public void removeSubscription(String topicFilter, String clientID) {
LOG.debug("removeSubscription topic filter: {} for clientID: {}", topicFilter, clientID);
LOG.info("Removing subscription. MqttClientId = {}, 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 = {}.",
clientID, topicFilter, m_db.getHashMap("subscriptions_" + clientID));
}
}
@Override
public void wipeSubscriptions(String clientID) {
LOG.debug("wipeSubscriptions");
if (LOG.isTraceEnabled()) {
LOG.trace("Subscription pre wipe: subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID));
}
LOG.info("Wiping subscriptions. MqttClientId = {}.", clientID);
m_db.delete("subscriptions_" + clientID);
if (LOG.isTraceEnabled()) {
LOG.trace("Subscription post wipe: subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID));
if (LOG.isDebugEnabled()) {
LOG.debug("The subscriptions have been removed. MqttClientId = {}, clientSubscriptions = {}.", clientID,
m_db.getHashMap("subscriptions_" + clientID));
}
}
@Override
public List<ClientTopicCouple> listAllSubscriptions() {
LOG.info("Retrieving existing subscriptions...");
final List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
for (String clientID : m_persistentSessions.keySet()) {
ConcurrentMap<String, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
@@ -106,24 +111,31 @@ class MapDBSessionsStore implements ISessionsStore {
allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter));
}
}
LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions);
if (LOG.isDebugEnabled()) {
LOG.debug("The existing subscriptions have been retrieved. Result = {}.", allSubscriptions);
}
return allSubscriptions;
}
@Override
public Subscription getSubscription(ClientTopicCouple couple) {
ConcurrentMap<String, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID);
LOG.debug("subscriptions_{}: {}", couple.clientID, clientSubscriptions);
LOG.info("Retrieving subscriptions. MqttClientId = {}, subscriptions = {}.", couple.clientID,
clientSubscriptions);
return clientSubscriptions.get(couple.topicFilter);
}
@Override
public List<Subscription> getSubscriptions() {
LOG.info("Retrieving existing subscriptions...");
List<Subscription> subscriptions = new ArrayList<>();
for (String clientID : m_persistentSessions.keySet()) {
ConcurrentMap<String, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
subscriptions.addAll(clientSubscriptions.values());
}
if (LOG.isDebugEnabled()) {
LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions);
}
return subscriptions;
}
@@ -134,28 +146,41 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public ClientSession createNewSession(String clientID, boolean cleanSession) {
LOG.debug("createNewSession for client <{}> with clean flag <{}>", clientID, cleanSession);
if (m_persistentSessions.containsKey(clientID)) {
LOG.error("already exists a session for client <{}>, bad condition", clientID);
LOG.error(
"Unable to create a new session: the client ID is already in use. MqttClientId = {}, cleanSession = {}.",
clientID, cleanSession);
throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
}
LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID);
LOG.info("Creating new session. MqttClientId = {}, 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);
if (!m_persistentSessions.containsKey(clientID)) {
LOG.warn("The session does not exist. MqttClientId = {}.", clientID);
return null;
}
PersistentSession storedSession = m_persistentSessions.get(clientID);
return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
}
@Override
public Collection<ClientSession> getAllSessions() {
Collection<ClientSession> result = new ArrayList<ClientSession>();
for (Map.Entry<String, PersistentSession> entry : m_persistentSessions.entrySet()) {
result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession));
}
return result;
}
@Override
public void updateCleanStatus(String clientID, boolean cleanSession) {
LOG.info("Updating cleanSession flag. MqttClientId = {}, cleanSession = {}.", clientID, cleanSession);
m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
}
@@ -164,6 +189,9 @@ class MapDBSessionsStore implements ISessionsStore {
* */
@Override
public int nextPacketID(String clientID) {
if (LOG.isDebugEnabled()) {
LOG.debug("Generating next packet ID. MqttClientId = {}.", clientID);
}
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient == null) {
int nextPacketId = 1;
@@ -177,15 +205,20 @@ class MapDBSessionsStore implements ISessionsStore {
int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
int nextPacketId = (maxId + 1) % 0xFFFF;
inFlightForClient.add(nextPacketId);
if (LOG.isDebugEnabled()) {
LOG.debug("The next packet ID has been generated. MqttClientId = {}, result = {}.", clientID, nextPacketId);
}
return nextPacketId;
}
@Override
public void inFlightAck(String clientID, int messageID) {
LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID);
if (LOG.isDebugEnabled()) {
LOG.debug("Acknowledging inflight message. MqttClientId = {}, messageId = {}.", clientID, messageID);
}
Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
if (m == null) {
LOG.error("Can't find the inFlight record for client <{}>", clientID);
LOG.warn("Unable to retrieve inflight message record. MqttClientId = {}, messageId = {}.", clientID, messageID);
return;
}
m.remove(messageID);
@@ -199,6 +232,10 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public void inFlight(String clientID, int messageID, MessageGUID guid) {
if (LOG.isDebugEnabled()) {
LOG.debug("Storing inflight message. MqttClientId = {}, messageId = {}, guid = {}.", clientID, messageID,
guid);
}
ConcurrentMap<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
if (m == null) {
m = new ConcurrentHashMap<>();
@@ -210,20 +247,26 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public BlockingQueue<StoredMessage> queue(String clientID) {
LOG.info("Queuing pending message. MqttClientId = {}, guid = {}.", clientID);
return this.m_db.getQueue(clientID);
}
@Override
public void dropQueue(String clientID) {
LOG.info("Removing pending messages. MqttClientId = {}.", clientID);
this.m_db.delete(clientID);
}
@Override
public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) {
LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID);
if (LOG.isDebugEnabled()) {
LOG.debug("Moving inflight message to second phase ack state. MqttClientId = {}, messageID = {}.", clientID,
messageID);
}
Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
if (m == null) {
LOG.error("Can't find the inFlight record for client <{}>", clientID);
LOG.warn("Unable to retrieve inflight message record. MqttClientId = {}, messageId = {}.", clientID,
messageID);
return;
}
MessageGUID guid = m.remove(messageID);
@@ -234,7 +277,6 @@ class MapDBSessionsStore implements ISessionsStore {
inFlightForClient.remove(messageID);
}
LOG.info("Moving to second phase store");
Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
messageIDs.put(messageID, guid);
m_secondPhaseStore.put(clientID, messageIDs);
@@ -242,6 +284,9 @@ 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);
}
Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
MessageGUID guid = messageIDs.remove(messageID);
m_secondPhaseStore.put(clientID, messageIDs);
@@ -250,8 +295,16 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public MessageGUID mapToGuid(String clientID, int messageID) {
if (LOG.isDebugEnabled()) {
LOG.debug("Mapping message ID to GUID. MqttClientId = {}, messageId = {}.", clientID, messageID);
}
ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
return messageIdToGuid.get(messageID);
MessageGUID result = messageIdToGuid.get(messageID);
if (LOG.isDebugEnabled()) {
LOG.debug("The message ID has been mapped to a GUID. MqttClientId = {}, messageId = {}, guid = {}.",
clientID, messageID, result);
}
return result;
}
static String messageId2GuidsMapName(String clientID) {
@@ -260,13 +313,39 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public StoredMessage getInflightMessage(String clientID, int messageID) {
LOG.info("Retrieving inflight message. MqttClientId = {}, messageId = {}.", clientID, messageID);
Map<Integer, MessageGUID> clientEntries = m_inflightStore.get(clientID);
if (clientEntries == null)
return null;
if (clientEntries == null) {
LOG.warn("The client has no inflight messages. MqttClientId = {}, messageId = {}.", clientID, messageID);
return null;
}
MessageGUID guid = clientEntries.get(messageID);
LOG.info("inflight messageID {} guid <{}>", messageID, guid);
if (guid == null)
return null;
if (guid == null) {
LOG.warn("The message ID does not have an associated GUID. MqttClientId = {}, messageId = {}.", clientID,
messageID);
return null;
}
return m_messagesStore.getMessageByGuid(guid);
}
@Override
public int getInflightMessagesNo(String clientID) {
if (!m_inflightStore.containsKey(clientID))
return 0;
else
return m_inflightStore.get(clientID).size();
}
@Override
public int getPendingPublishMessagesNo(String clientID) {
return m_messagesStore.getPendingPublishMessages(clientID);
}
@Override
public int getSecondPhaseAckPendingMessages(String clientID) {
if (!m_secondPhaseStore.containsKey(clientID))
return 0;
else
return m_secondPhaseStore.get(clientID).size();
}
}