diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java
index f123652f..87074b32 100644
--- a/broker/src/main/java/io/moquette/server/Server.java
+++ b/broker/src/main/java/io/moquette/server/Server.java
@@ -13,6 +13,7 @@
*
* You may elect to redistribute this code under either of these licenses.
*/
+
package io.moquette.server;
import com.hazelcast.config.ClasspathXmlConfig;
@@ -38,26 +39,25 @@ import io.moquette.spi.security.ISslContextCreator;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
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.
+ * Launch a configured version of the server.
+ *
* @author andrea
*/
public class Server {
-
+
private static final Logger LOG = LoggerFactory.getLogger(Server.class);
private static final String HZ_INTERCEPT_HANDLER = HazelcastInterceptHandler.class.getCanonicalName();
-
+
private ServerAcceptor m_acceptor;
private volatile boolean m_initialized;
@@ -72,22 +72,24 @@ public class Server {
final Server server = new Server();
server.startServer();
System.out.println("Server started, version 0.10-SNAPSHOT");
- //Bind a shutdown hook
+ // Bind a shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread() {
+
@Override
public void run() {
server.stopServer();
}
});
}
-
+
/**
- * Starts Moquette bringing the configuration from the file
- * located at m_config/moquette.conf
- * @throws IOException in case of any IO error.
+ * Starts Moquette bringing the configuration from the file located at m_config/moquette.conf
+ *
+ * @throws IOException
+ * in case of any IO error.
*/
public void startServer() throws IOException {
- File defaultConfigurationFile = 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);
@@ -101,8 +103,11 @@ public class Server {
/**
* Starts Moquette bringing the configuration from the given file
- * @param configFile text file that contains the configuration.
- * @throws IOException in case of any IO Error.
+ *
+ * @param configFile
+ * text file that contains the configuration.
+ * @throws IOException
+ * in case of any IO Error.
*/
public void startServer(File configFile) throws IOException {
LOG.info("Starting Moquette server. Configuration file path = {}.", configFile.getAbsolutePath());
@@ -110,50 +115,58 @@ public class Server {
final IConfig config = new ResourceLoaderConfig(filesystemLoader);
startServer(config);
}
-
+
/**
* Starts the server with the given properties.
- *
+ *
* Its suggested to at least have the following properties:
*
- * - port
- * - password_file
+ * - port
+ * - password_file
*
*
- * @param configProps the properties map to use as configuration.
- * @throws IOException in case of any IO Error.
+ * @param configProps
+ * the properties map to use as configuration.
+ * @throws IOException
+ * in case of any IO Error.
*/
public void startServer(Properties configProps) throws IOException {
- LOG.info("Starting Moquette server using properties object...");
+ LOG.info("Starting Moquette server using properties object...");
final IConfig config = new MemoryConfig(configProps);
startServer(config);
}
/**
* Starts Moquette bringing the configuration files from the given Config implementation.
- * @param config the configuration to use to start the broker.
- * @throws IOException in case of any IO Error.
+ *
+ * @param config
+ * the configuration to use to start the broker.
+ * @throws IOException
+ * in case of any IO Error.
*/
public void startServer(IConfig config) throws IOException {
- LOG.info("Starting Moquette server using IConfig instance...");
+ LOG.info("Starting Moquette server using IConfig instance...");
startServer(config, null);
}
/**
- * Starts Moquette with config provided by an implementation of IConfig class and with the
- * set of InterceptHandler.
- * @param config the configuration to use to start the broker.
- * @param handlers the handlers to install in the broker.
- * @throws IOException in case of any IO Error.
- * */
+ * Starts Moquette with config provided by an implementation of IConfig class and with the set
+ * of InterceptHandler.
+ *
+ * @param config
+ * the configuration to use to start the broker.
+ * @param handlers
+ * the handlers to install in the broker.
+ * @throws IOException
+ * in case of any IO Error.
+ */
public void startServer(IConfig config, List handlers) throws IOException {
- LOG.info("Starting moquette server using IConfig instance and intercept handlers...");
+ LOG.info("Starting moquette server using IConfig instance and intercept handlers...");
startServer(config, handlers, null, null, null);
}
- public void startServer(IConfig config, List handlers,
- ISslContextCreator sslCtxCreator, IAuthenticator authenticator,
- IAuthorizator authorizator) throws IOException {
+ public void startServer(IConfig config, List handlers, ISslContextCreator sslCtxCreator,
+ IAuthenticator authenticator, IAuthorizator authorizator) throws IOException {
if (handlers == null) {
handlers = Collections.emptyList();
}
@@ -164,10 +177,13 @@ public class Server {
config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp);
}
configureCluster(config);
- LOG.info("Configuring Using persistent store file. Path = {}.", config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME));
+ 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);
+ final ProtocolProcessor processor = m_processorBootstrapper
+ .init(config, handlers, authenticator, authorizator, this);
if (sslCtxCreator == null) {
LOG.warn("Using default SSL context creator...");
@@ -184,19 +200,19 @@ public class Server {
}
private void configureCluster(IConfig config) throws FileNotFoundException {
- LOG.info("Configuring embedded Hazelcast instance...");
+ LOG.info("Configuring embedded Hazelcast instance...");
String interceptHandlerClassname = config.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
if (interceptHandlerClassname == null || !HZ_INTERCEPT_HANDLER.equals(interceptHandlerClassname)) {
- LOG.info("There are no Hazelcast intercept handlers. The server won't start a Hazelcast instance.");
+ LOG.info("There are no Hazelcast intercept handlers. The server won't start a Hazelcast instance.");
return;
}
String hzConfigPath = config.getProperty(BrokerConstants.HAZELCAST_CONFIGURATION);
if (hzConfigPath != null) {
boolean isHzConfigOnClasspath = this.getClass().getClassLoader().getResource(hzConfigPath) != null;
- Config hzconfig = isHzConfigOnClasspath ?
- new ClasspathXmlConfig(hzConfigPath) :
- new FileSystemXmlConfig(hzConfigPath);
- LOG.info("Starting Hazelcast instance. ConfigurationFile = {}.",hzconfig);
+ Config hzconfig = isHzConfigOnClasspath
+ ? new ClasspathXmlConfig(hzConfigPath)
+ : new FileSystemXmlConfig(hzConfigPath);
+ LOG.info("Starting Hazelcast instance. ConfigurationFile = {}.", hzconfig);
hazelcastInstance = Hazelcast.newHazelcastInstance(hzconfig);
} else {
LOG.info("Starting Hazelcast instance with default configuration.");
@@ -206,43 +222,48 @@ public class Server {
}
private void listenOnHazelCastMsg() {
- LOG.info("Subscribing to Hazelcast topic. TopicName = {}.", "moquette");
+ LOG.info("Subscribing to Hazelcast topic. TopicName = {}.", "moquette");
HazelcastInstance hz = getHazelcastInstance();
ITopic topic = hz.getTopic("moquette");
topic.addMessageListener(new HazelcastListener(this));
}
- public HazelcastInstance getHazelcastInstance(){
+ public HazelcastInstance getHazelcastInstance() {
return hazelcastInstance;
}
/**
- * Use the broker to publish a message. It's intended for embedding applications.
- * It can be used only after the server is correctly started with startServer.
+ * Use the broker to publish a message. It's intended for embedding applications. It can be used
+ * only after the server is correctly started with startServer.
*
- * @param msg the message to forward.
- * @param clientId the id of the sending server.
- * @throws IllegalStateException if the server is not yet started
- * */
+ * @param msg
+ * the message to forward.
+ * @param clientId
+ * the id of the sending server.
+ * @throws IllegalStateException
+ * if the server is not yet started
+ */
public void internalPublish(MqttPublishMessage msg, final String clientId) {
final int messageID = msg.variableHeader().messageId();
if (!m_initialized) {
- LOG.error("The server is not started. The message cannot be published. MqttClientId = {}, messageId = {}",
- clientId, messageID);
+ LOG.error(
+ "The server is not started. The message cannot be published. MqttClientId = {}, messageId = {}",
+ clientId,
+ messageID);
throw new IllegalStateException("Can't publish on a server is not yet started");
}
LOG.debug("Publishing message. MqttClientId = {}, messageId = {}", clientId, messageID);
m_processor.internalPublish(msg, clientId);
}
-
+
public void stopServer() {
- LOG.info("Unbinding server from the configured ports...");
+ LOG.info("Unbinding server from the configured ports...");
m_acceptor.close();
LOG.info("Stopping MQTT protocol processor...");
m_processorBootstrapper.shutdown();
m_initialized = false;
if (hazelcastInstance != null) {
- LOG.info("Stopping embedded Hazelcast instance...");
+ LOG.info("Stopping embedded Hazelcast instance...");
try {
hazelcastInstance.shutdown();
} catch (HazelcastInstanceNotActiveException e) {
@@ -253,8 +274,8 @@ public class Server {
}
/**
- * SPI method used by Broker embedded applications to get list of subscribers.
- * Returns null if the broker is not started.
+ * SPI method used by Broker embedded applications to get list of subscribers. Returns null if
+ * the broker is not started.
*
* @return list of subscriptions.
*/
@@ -267,14 +288,16 @@ public class Server {
/**
* SPI method used by Broker embedded applications to add intercept handlers.
- * @param interceptHandler the handler to add.
- * */
+ *
+ * @param interceptHandler
+ * the handler to add.
+ */
public void addInterceptHandler(InterceptHandler interceptHandler) {
if (!m_initialized) {
- LOG.error("The server is not started. The MQTT message interceptor cannot be added. InterceptorId = {}.",
- interceptHandler.getID());
+ LOG.error(
+ "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");
-
}
LOG.info("Adding MQTT message interceptor. InterceptorId = {}.", interceptHandler.getID());
m_processor.addInterceptHandler(interceptHandler);
@@ -282,12 +305,15 @@ public class Server {
/**
* SPI method used by Broker embedded applications to remove intercept handlers.
- * @param interceptHandler the handler to remove.
- * */
+ *
+ * @param interceptHandler
+ * the handler to remove.
+ */
public void removeInterceptHandler(InterceptHandler interceptHandler) {
if (!m_initialized) {
- LOG.error("The server is not started. The MQTT message interceptor cannot be removed. InterceptorId = {}.",
- interceptHandler.getID());
+ LOG.error(
+ "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");
}
LOG.info("Removing MQTT message interceptor. InterceptorId = {}.", interceptHandler.getID());
@@ -296,7 +322,7 @@ public class Server {
/**
* Returns the connections manager of this broker.
- *
+ *
* @return IConnectionsManager the instance used bt the broker.
*/
public IConnectionsManager getConnectionsManager() {