diff --git a/broker/src/main/java/io/moquette/broker/Authorizator.java b/broker/src/main/java/io/moquette/broker/Authorizator.java
index 621edfdc..a676d9de 100644
--- a/broker/src/main/java/io/moquette/broker/Authorizator.java
+++ b/broker/src/main/java/io/moquette/broker/Authorizator.java
@@ -1,6 +1,5 @@
package io.moquette.broker;
-import io.moquette.spi.ClientSession;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizatorPolicy;
import io.netty.handler.codec.mqtt.MqttQoS;
diff --git a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
index 9c996559..93ac682c 100644
--- a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
+++ b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
@@ -33,14 +33,22 @@ import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.MessageToMessageEncoder;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+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.WebSocketServerProtocolHandler;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.net.ssl.SSLEngine;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
@@ -144,18 +152,18 @@ public class NewNettyAcceptor {
this.errorsCather = Optional.empty();
}
initializePlainTCPTransport(mqttHandler, props);
-// initializeWebSocketTransport(mqttHandler, props);
-// String sslTcpPortProp = props.getProperty(BrokerConstants.SSL_PORT_PROPERTY_NAME);
-// String wssPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME);
-// if (sslTcpPortProp != null || wssPortProp != null) {
-// SslContext sslContext = sslCtxCreator.initSSLContext();
-// if (sslContext == null) {
-// LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks");
-// return;
-// }
-// initializeSSLTCPTransport(mqttHandler, props, sslContext);
-// initializeWSSTransport(mqttHandler, props, sslContext);
-// }
+ initializeWebSocketTransport(mqttHandler, props);
+ String sslTcpPortProp = props.getProperty(BrokerConstants.SSL_PORT_PROPERTY_NAME);
+ String wssPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME);
+ if (sslTcpPortProp != null || wssPortProp != null) {
+ SslContext sslContext = sslCtxCreator.initSSLContext();
+ if (sslContext == null) {
+ LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks");
+ return;
+ }
+ initializeSSLTCPTransport(mqttHandler, props, sslContext);
+ initializeWSSTransport(mqttHandler, props, sslContext);
+ }
}
private void initFactory(String host, int port, String protocol, final PipelineInitializer pipeliner) {
@@ -219,117 +227,117 @@ public class NewNettyAcceptor {
});
}
-// private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) {
-// LOG.debug("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("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);
-//
-// final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
-//
-// String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
-// initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
-//
-// @Override
-// void init(SocketChannel channel) {
-// ChannelPipeline pipeline = channel.pipeline();
-// pipeline.addLast(new HttpServerCodec());
-// pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
-// 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(nettyChannelTimeoutSeconds, 0, 0));
-// pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
-// pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
-// pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
-// pipeline.addLast("encoder", MqttEncoder.INSTANCE);
-// pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
-// pipeline.addLast("messageLogger", new MQTTMessageLogger());
-// pipeline.addLast("handler", handler);
-// }
-// });
-// }
-//
-// private void initializeSSLTCPTransport(NettyMQTTHandler handler, IConfig props, SslContext sslContext) {
-// LOG.debug("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("Property {} has been set to {}. SSL MQTT will be disabled",
-// BrokerConstants.SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
-// return;
-// }
-//
-// int sslPort = Integer.parseInt(sslPortProp);
-// LOG.debug("Starting SSL on port {}", sslPort);
-//
-// final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
-// 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, "SSL MQTT", new PipelineInitializer() {
-//
-// @Override
-// void init(SocketChannel channel) throws Exception {
-// ChannelPipeline pipeline = channel.pipeline();
-// pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
-// pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
-// pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
-// // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
-// pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
-// pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
-// pipeline.addLast("encoder", MqttEncoder.INSTANCE);
-// pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
-// pipeline.addLast("messageLogger", new MQTTMessageLogger());
-// pipeline.addLast("handler", handler);
-// }
-// });
-// }
-//
-// private void initializeWSSTransport(NettyMQTTHandler handler, IConfig props, SslContext sslContext) {
-// LOG.debug("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("Property {} has been set to {}. Secure websocket MQTT will be disabled",
-// BrokerConstants.WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
-// return;
-// }
-// int sslPort = Integer.parseInt(sslPortProp);
-// final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
-// 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, "Secure websocket", new PipelineInitializer() {
-//
-// @Override
-// void init(SocketChannel channel) throws Exception {
-// ChannelPipeline pipeline = channel.pipeline();
-// pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
-// pipeline.addLast("httpEncoder", new HttpResponseEncoder());
-// pipeline.addLast("httpDecoder", new HttpRequestDecoder());
-// pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
-// 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(nettyChannelTimeoutSeconds, 0, 0));
-// pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
-// pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
-// pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
-// pipeline.addLast("encoder", MqttEncoder.INSTANCE);
-// pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
-// pipeline.addLast("messageLogger", new MQTTMessageLogger());
-// pipeline.addLast("handler", handler);
-// }
-// });
-// }
+ private void initializeWebSocketTransport(final NewNettyMQTTHandler handler, IConfig props) {
+ LOG.debug("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("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);
+
+ final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+
+ String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+ initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
+
+ @Override
+ void init(SocketChannel channel) {
+ ChannelPipeline pipeline = channel.pipeline();
+ pipeline.addLast(new HttpServerCodec());
+ pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
+ 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(nettyChannelTimeoutSeconds, 0, 0));
+ pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
+ pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
+ pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
+ pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+ pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+ pipeline.addLast("messageLogger", new MQTTMessageLogger());
+ pipeline.addLast("handler", handler);
+ }
+ });
+ }
+
+ private void initializeSSLTCPTransport(NewNettyMQTTHandler handler, IConfig props, SslContext sslContext) {
+ LOG.debug("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("Property {} has been set to {}. SSL MQTT will be disabled",
+ BrokerConstants.SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+ return;
+ }
+
+ int sslPort = Integer.parseInt(sslPortProp);
+ LOG.debug("Starting SSL on port {}", sslPort);
+
+ final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+ 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, "SSL MQTT", new PipelineInitializer() {
+
+ @Override
+ void init(SocketChannel channel) throws Exception {
+ ChannelPipeline pipeline = channel.pipeline();
+ pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
+ pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
+ pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
+ // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
+ pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
+ pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
+ pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+ pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+ pipeline.addLast("messageLogger", new MQTTMessageLogger());
+ pipeline.addLast("handler", handler);
+ }
+ });
+ }
+
+ private void initializeWSSTransport(NewNettyMQTTHandler handler, IConfig props, SslContext sslContext) {
+ LOG.debug("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("Property {} has been set to {}. Secure websocket MQTT will be disabled",
+ BrokerConstants.WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+ return;
+ }
+ int sslPort = Integer.parseInt(sslPortProp);
+ final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+ 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, "Secure websocket", new PipelineInitializer() {
+
+ @Override
+ void init(SocketChannel channel) throws Exception {
+ ChannelPipeline pipeline = channel.pipeline();
+ pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
+ pipeline.addLast("httpEncoder", new HttpResponseEncoder());
+ pipeline.addLast("httpDecoder", new HttpRequestDecoder());
+ pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
+ 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(nettyChannelTimeoutSeconds, 0, 0));
+ pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
+ pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
+ pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
+ pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+ pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+ pipeline.addLast("messageLogger", new MQTTMessageLogger());
+ pipeline.addLast("handler", handler);
+ }
+ });
+ }
@SuppressWarnings("FutureReturnValueIgnored")
public void close() {
@@ -369,15 +377,15 @@ public class NewNettyAcceptor {
metrics.messagesWrote(), bytesMetrics.readBytes(), bytesMetrics.wroteBytes());
}
-// private ChannelHandler createSslHandler(SocketChannel channel, SslContext sslContext, boolean needsClientAuth) {
-// SSLEngine sslEngine = sslContext.newEngine(
-// channel.alloc(),
-// channel.remoteAddress().getHostString(),
-// channel.remoteAddress().getPort());
-// sslEngine.setUseClientMode(false);
-// if (needsClientAuth) {
-// sslEngine.setNeedClientAuth(true);
-// }
-// return new SslHandler(sslEngine);
-// }
+ private ChannelHandler createSslHandler(SocketChannel channel, SslContext sslContext, boolean needsClientAuth) {
+ SSLEngine sslEngine = sslContext.newEngine(
+ channel.alloc(),
+ channel.remoteAddress().getHostString(),
+ channel.remoteAddress().getPort());
+ sslEngine.setUseClientMode(false);
+ if (needsClientAuth) {
+ sslEngine.setNeedClientAuth(true);
+ }
+ return new SslHandler(sslEngine);
+ }
}
diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java
index 9c358642..210029bc 100644
--- a/broker/src/main/java/io/moquette/broker/Server.java
+++ b/broker/src/main/java/io/moquette/broker/Server.java
@@ -3,12 +3,12 @@ package io.moquette.broker;
import io.moquette.BrokerConstants;
import io.moquette.interception.InterceptHandler;
import io.moquette.persistence.MemoryStorageService;
+import io.moquette.server.DefaultMoquetteSslContextCreator;
import io.moquette.server.config.*;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.BrokerInterceptor;
import io.moquette.spi.impl.SessionsRepository;
-import io.moquette.spi.impl.security.AcceptAllAuthenticator;
-import io.moquette.spi.impl.security.PermitAllAuthorizatorPolicy;
+import io.moquette.spi.impl.security.*;
import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory;
import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
import io.moquette.spi.security.IAuthenticator;
@@ -21,6 +21,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
+import java.text.ParseException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -136,24 +137,17 @@ public class Server {
config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp);
}
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("Configuring Using persistent store file, path: {}", persistencePath);
initInterceptors(config, handlers);
LOG.debug("Initialized MQTT protocol processor");
-// if (sslCtxCreator == null) {
-// LOG.info("Using default SSL context creator");
-// sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
-// }
- // DBG
- if (authenticator == null) {
- authenticator = new AcceptAllAuthenticator();
- }
- if (authorizatorPolicy == null) {
- authorizatorPolicy = new PermitAllAuthorizatorPolicy();
+ if (sslCtxCreator == null) {
+ LOG.info("Using default SSL context creator");
+ sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
}
+ authenticator = initializeAuthenticator(authenticator, config);
+ authorizatorPolicy = initializeAuthorizatorPolicy(authorizatorPolicy, config);
+ // TODO user real implementation DBG
MemoryStorageService memStorage = new MemoryStorageService(null, null);
ISessionsStore sessionStore = memStorage.sessionsStore();
SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null);
@@ -177,6 +171,63 @@ public class Server {
initialized = true;
}
+ private IAuthorizatorPolicy initializeAuthorizatorPolicy(IAuthorizatorPolicy authorizatorPolicy, IConfig props) {
+// if (authorizatorPolicy == null) {
+// authorizatorPolicy = new PermitAllAuthorizatorPolicy();
+// }
+// return authorizatorPolicy;
+
+ LOG.debug("Configuring MQTT authorizator policy");
+ String authorizatorClassName = props.getProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, "");
+ if (authorizatorPolicy == null && !authorizatorClassName.isEmpty()) {
+ authorizatorPolicy = loadClass(authorizatorClassName, IAuthorizatorPolicy.class, IConfig.class, props);
+ }
+
+ if (authorizatorPolicy == null) {
+ String aclFilePath = props.getProperty(BrokerConstants.ACL_FILE_PROPERTY_NAME, "");
+ if (aclFilePath != null && !aclFilePath.isEmpty()) {
+ authorizatorPolicy = new DenyAllAuthorizatorPolicy();
+ try {
+ LOG.info("Parsing ACL file. Path = {}", aclFilePath);
+ IResourceLoader resourceLoader = props.getResourceLoader();
+ authorizatorPolicy = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath));
+ } catch (ParseException pex) {
+ LOG.error("Unable to parse ACL file. path=" + aclFilePath, pex);
+ }
+ } else {
+ authorizatorPolicy = new PermitAllAuthorizatorPolicy();
+ }
+ LOG.info("Authorizator policy {} instance will be used", authorizatorPolicy.getClass().getName());
+ }
+ return authorizatorPolicy;
+ }
+
+ private IAuthenticator initializeAuthenticator(IAuthenticator authenticator, IConfig props) {
+// if (authenticator == null) {
+// authenticator = new AcceptAllAuthenticator();
+// }
+// return authenticator;
+
+ LOG.debug("Configuring MQTT authenticator");
+ String authenticatorClassName = props.getProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, "");
+
+ if (authenticator == null && !authenticatorClassName.isEmpty()) {
+ authenticator = loadClass(authenticatorClassName, IAuthenticator.class, IConfig.class, props);
+ }
+
+ IResourceLoader resourceLoader = props.getResourceLoader();
+ if (authenticator == null) {
+ String passwdPath = props.getProperty(BrokerConstants.PASSWORD_FILE_PROPERTY_NAME, "");
+ if (passwdPath.isEmpty()) {
+ authenticator = new AcceptAllAuthenticator();
+ } else {
+ authenticator = new ResourceAuthenticator(resourceLoader, passwdPath);
+ }
+ LOG.info("An {} authenticator instance will be used", authenticator.getClass().getName());
+ }
+ return authenticator;
+ }
+
private void initInterceptors(IConfig props, List embeddedObservers) {
LOG.info("Configuring message interceptors...");
diff --git a/broker/src/main/java/io/moquette/server/DefaultMoquetteSslContextCreator.java b/broker/src/main/java/io/moquette/server/DefaultMoquetteSslContextCreator.java
index df981f47..1e1c6963 100644
--- a/broker/src/main/java/io/moquette/server/DefaultMoquetteSslContextCreator.java
+++ b/broker/src/main/java/io/moquette/server/DefaultMoquetteSslContextCreator.java
@@ -49,13 +49,13 @@ import org.slf4j.LoggerFactory;
* Moquette server implementation to load SSL certificate from local filesystem path configured in
* config file.
*/
-class DefaultMoquetteSslContextCreator implements ISslContextCreator {
+public class DefaultMoquetteSslContextCreator implements ISslContextCreator {
private static final Logger LOG = LoggerFactory.getLogger(DefaultMoquetteSslContextCreator.class);
private final IConfig props;
- DefaultMoquetteSslContextCreator(IConfig props) {
+ public DefaultMoquetteSslContextCreator(IConfig props) {
this.props = Objects.requireNonNull(props);
}
diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java
index 11cc8851..7d58a559 100644
--- a/broker/src/main/java/io/moquette/spi/ClientSession.java
+++ b/broker/src/main/java/io/moquette/spi/ClientSession.java
@@ -38,6 +38,7 @@ import java.util.Set;
* Optionally, QoS 0 messages pending transmission to the Client.
*
*/
+@Deprecated
public abstract class ClientSession {
public final String clientID;
diff --git a/broker/src/main/java/io/moquette/spi/DurableSession.java b/broker/src/main/java/io/moquette/spi/DurableSession.java
index 467c4877..ae80fd97 100644
--- a/broker/src/main/java/io/moquette/spi/DurableSession.java
+++ b/broker/src/main/java/io/moquette/spi/DurableSession.java
@@ -30,6 +30,7 @@ import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
+@Deprecated
public class DurableSession extends ClientSession {
class OutboundFlightZone {
diff --git a/broker/src/main/java/io/moquette/spi/EnqueuedMessage.java b/broker/src/main/java/io/moquette/spi/EnqueuedMessage.java
index c4c3b905..1403ade5 100644
--- a/broker/src/main/java/io/moquette/spi/EnqueuedMessage.java
+++ b/broker/src/main/java/io/moquette/spi/EnqueuedMessage.java
@@ -16,6 +16,7 @@
package io.moquette.spi;
+@Deprecated
public class EnqueuedMessage {
public final IMessagesStore.StoredMessage msg;
diff --git a/broker/src/main/java/io/moquette/spi/InboundFlightZone.java b/broker/src/main/java/io/moquette/spi/InboundFlightZone.java
index 6b82ce24..79f47161 100644
--- a/broker/src/main/java/io/moquette/spi/InboundFlightZone.java
+++ b/broker/src/main/java/io/moquette/spi/InboundFlightZone.java
@@ -18,6 +18,7 @@ package io.moquette.spi;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+@Deprecated
class InboundFlightZone {
private final Map inboundFlightMessages = new ConcurrentHashMap<>();
diff --git a/broker/src/main/java/io/moquette/spi/TransientSession.java b/broker/src/main/java/io/moquette/spi/TransientSession.java
index b2f1a435..2efebf46 100644
--- a/broker/src/main/java/io/moquette/spi/TransientSession.java
+++ b/broker/src/main/java/io/moquette/spi/TransientSession.java
@@ -32,6 +32,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
+@Deprecated
public class TransientSession extends ClientSession {
private static final Logger LOG = LoggerFactory.getLogger(TransientSession.class);