323 lines
13 KiB
Java
323 lines
13 KiB
Java
/*
|
|
* Copyright (c) 2012-2018 The original author or authors
|
|
* ------------------------------------------------------
|
|
* All rights reserved. This program and the accompanying materials
|
|
* are made available under the terms of the Eclipse Public License v1.0
|
|
* and Apache License v2.0 which accompanies this distribution.
|
|
*
|
|
* The Eclipse Public License is available at
|
|
* http://www.eclipse.org/legal/epl-v10.html
|
|
*
|
|
* The Apache License v2.0 is available at
|
|
* http://www.opensource.org/licenses/apache2.0.php
|
|
*
|
|
* You may elect to redistribute this code under either of these licenses.
|
|
*/
|
|
|
|
package io.moquette.server;
|
|
|
|
import com.hazelcast.config.ClasspathXmlConfig;
|
|
import com.hazelcast.config.Config;
|
|
import com.hazelcast.config.FileSystemXmlConfig;
|
|
import com.hazelcast.core.Hazelcast;
|
|
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;
|
|
import io.moquette.server.config.*;
|
|
import io.moquette.server.netty.NettyAcceptor;
|
|
import io.moquette.spi.impl.ProtocolProcessor;
|
|
import io.moquette.spi.impl.ProtocolProcessorBootstrapper;
|
|
import io.moquette.spi.impl.subscriptions.Subscription;
|
|
import io.moquette.spi.security.IAuthenticator;
|
|
import io.moquette.spi.security.IAuthorizator;
|
|
import io.moquette.spi.security.ISslContextCreator;
|
|
import io.netty.handler.codec.mqtt.MqttPublishMessage;
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
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 java.util.concurrent.Executors;
|
|
import java.util.concurrent.ScheduledExecutorService;
|
|
|
|
import static io.moquette.logging.LoggingUtils.getInterceptorIds;
|
|
|
|
/**
|
|
* Launch a configured version of the server.
|
|
*/
|
|
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;
|
|
|
|
private ProtocolProcessor m_processor;
|
|
|
|
private HazelcastInstance hazelcastInstance;
|
|
|
|
private ProtocolProcessorBootstrapper m_processorBootstrapper;
|
|
|
|
private ScheduledExecutorService scheduler;
|
|
|
|
public static void main(String[] args) throws IOException {
|
|
final Server server = new Server();
|
|
server.startServer();
|
|
System.out.println("Server started, version 0.12-SNAPSHOT");
|
|
//Bind a shutdown hook
|
|
Runtime.getRuntime().addShutdownHook(new Thread(server::stopServer));
|
|
}
|
|
|
|
/**
|
|
* 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();
|
|
LOG.info("Starting Moquette server. Configuration file path={}", defaultConfigurationFile.getAbsolutePath());
|
|
IResourceLoader filesystemLoader = new FileResourceLoader(defaultConfigurationFile);
|
|
final IConfig config = new ResourceLoaderConfig(filesystemLoader);
|
|
startServer(config);
|
|
}
|
|
|
|
private static File defaultConfigFile() {
|
|
String configPath = System.getProperty("moquette.path", null);
|
|
return new File(configPath, IConfig.DEFAULT_CONFIG);
|
|
}
|
|
|
|
/**
|
|
* 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.
|
|
*/
|
|
public void startServer(File configFile) throws IOException {
|
|
LOG.info("Starting Moquette server. Configuration file path={}", configFile.getAbsolutePath());
|
|
IResourceLoader filesystemLoader = new FileResourceLoader(configFile);
|
|
final IConfig config = new ResourceLoaderConfig(filesystemLoader);
|
|
startServer(config);
|
|
}
|
|
|
|
/**
|
|
* Starts the server with the given properties.
|
|
* <p>
|
|
* Its suggested to at least have the following properties:
|
|
* <ul>
|
|
* <li>port</li>
|
|
* <li>password_file</li>
|
|
* </ul>
|
|
*
|
|
* @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.debug("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.
|
|
*/
|
|
public void startServer(IConfig config) throws IOException {
|
|
LOG.debug("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.
|
|
*/
|
|
public void startServer(IConfig config, List<? extends InterceptHandler> handlers) throws IOException {
|
|
LOG.debug("Starting moquette server using IConfig instance and intercept handlers");
|
|
startServer(config, handlers, null, null, null);
|
|
}
|
|
|
|
public void startServer(IConfig config, List<? extends InterceptHandler> handlers, ISslContextCreator sslCtxCreator,
|
|
IAuthenticator authenticator, IAuthorizator authorizator) throws IOException {
|
|
final long start = System.currentTimeMillis();
|
|
if (handlers == null) {
|
|
handlers = Collections.emptyList();
|
|
}
|
|
LOG.trace("Starting Moquette Server. MQTT message interceptors={}", getInterceptorIds(handlers));
|
|
|
|
scheduler = Executors.newScheduledThreadPool(1);
|
|
|
|
final String handlerProp = System.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
|
|
if (handlerProp != null) {
|
|
config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp);
|
|
}
|
|
configureCluster(config);
|
|
final String persistencePath = config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME);
|
|
LOG.debug("Configuring Using persistent store file, path={}", persistencePath);
|
|
m_processorBootstrapper = new ProtocolProcessorBootstrapper();
|
|
final ProtocolProcessor processor = m_processorBootstrapper.init(config, handlers, authenticator, authorizator,
|
|
this);
|
|
LOG.debug("Initialized MQTT protocol processor");
|
|
if (sslCtxCreator == null) {
|
|
LOG.info("Using default SSL context creator");
|
|
sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
|
|
}
|
|
|
|
m_acceptor = new NettyAcceptor();
|
|
m_acceptor.initialize(processor, config, sslCtxCreator);
|
|
m_processor = processor;
|
|
|
|
final long startTime = System.currentTimeMillis() - start;
|
|
LOG.info("Moquette server has been started successfully in {} ms", startTime);
|
|
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);
|
|
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);
|
|
hazelcastInstance = Hazelcast.newHazelcastInstance(hzconfig);
|
|
} else {
|
|
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));
|
|
}
|
|
|
|
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.
|
|
*
|
|
* @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("Moquette is not started, internal message cannot be published. CId={}, messageId={}", clientId,
|
|
messageID);
|
|
throw new IllegalStateException("Can't publish on a server is not yet started");
|
|
}
|
|
LOG.debug("Publishing message. CId={}, messageId={}", clientId, messageID);
|
|
m_processor.internalPublish(msg, clientId);
|
|
}
|
|
|
|
public void stopServer() {
|
|
LOG.info("Unbinding server from the configured ports");
|
|
m_acceptor.close();
|
|
LOG.trace("Stopping MQTT protocol processor");
|
|
m_processorBootstrapper.shutdown();
|
|
m_initialized = false;
|
|
if (hazelcastInstance != null) {
|
|
LOG.trace("Stopping embedded Hazelcast instance");
|
|
try {
|
|
hazelcastInstance.shutdown();
|
|
} catch (HazelcastInstanceNotActiveException e) {
|
|
LOG.warn("embedded Hazelcast instance is already shut down.");
|
|
}
|
|
}
|
|
|
|
scheduler.shutdown();
|
|
|
|
LOG.info("Moquette server has been stopped.");
|
|
}
|
|
|
|
/**
|
|
* SPI method used by Broker embedded applications to get list of subscribers. Returns null if
|
|
* the broker is not started.
|
|
*
|
|
* @return list of subscriptions.
|
|
*/
|
|
public List<Subscription> getSubscriptions() {
|
|
if (m_processorBootstrapper == null) {
|
|
return null;
|
|
}
|
|
return m_processorBootstrapper.getSubscriptions();
|
|
}
|
|
|
|
/**
|
|
* SPI method used by Broker embedded applications to add intercept handlers.
|
|
*
|
|
* @param interceptHandler the handler to add.
|
|
*/
|
|
public void addInterceptHandler(InterceptHandler interceptHandler) {
|
|
if (!m_initialized) {
|
|
LOG.error("Moquette is not started, MQTT message interceptor cannot be added. InterceptorId={}",
|
|
interceptHandler.getID());
|
|
throw new IllegalStateException("Can't register interceptors on a server that is not yet started");
|
|
}
|
|
LOG.info("Adding MQTT message interceptor. InterceptorId={}", interceptHandler.getID());
|
|
m_processor.addInterceptHandler(interceptHandler);
|
|
}
|
|
|
|
/**
|
|
* SPI method used by Broker embedded applications to remove intercept handlers.
|
|
*
|
|
* @param interceptHandler the handler to remove.
|
|
*/
|
|
public void removeInterceptHandler(InterceptHandler interceptHandler) {
|
|
if (!m_initialized) {
|
|
LOG.error("Moquette is not started, MQTT message interceptor cannot be removed. InterceptorId={}",
|
|
interceptHandler.getID());
|
|
throw new IllegalStateException("Can't deregister interceptors from a server that is not yet started");
|
|
}
|
|
LOG.info("Removing MQTT message interceptor. InterceptorId={}", interceptHandler.getID());
|
|
m_processor.removeInterceptHandler(interceptHandler);
|
|
}
|
|
|
|
/**
|
|
* Returns the connections manager of this broker.
|
|
*
|
|
* @return IConnectionsManager the instance used bt the broker.
|
|
*/
|
|
public IConnectionsManager getConnectionsManager() {
|
|
return m_processorBootstrapper.getConnectionDescriptors();
|
|
}
|
|
|
|
public ProtocolProcessor getProcessor() {
|
|
return m_processor;
|
|
}
|
|
|
|
public ScheduledExecutorService getScheduler() {
|
|
return scheduler;
|
|
}
|
|
}
|