Moved project package from org.eclipse.moquette to io.moquette

This commit is contained in:
andsel
2015-11-20 21:30:59 +01:00
parent f438ef7f49
commit 371c740321
175 changed files with 629 additions and 669 deletions
@@ -0,0 +1,49 @@
/*
* Copyright (c) 2012-2015 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.interception;
import io.moquette.proto.messages.AbstractMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.interception.messages.InterceptConnectMessage;
import io.moquette.interception.messages.InterceptDisconnectMessage;
import io.moquette.interception.messages.InterceptPublishMessage;
import io.moquette.interception.messages.InterceptSubscribeMessage;
import io.moquette.interception.messages.InterceptUnsubscribeMessage;
/**
* This interface is used to inject code for intercepting broker events.
* <p>
* The events can act only as observers.
* <p>
* Almost every method receives a subclass of {@link AbstractMessage}, except
* <code>onDisconnect</code> that receives the client id string and
* <code>onSubscribe</code> and <code>onUnsubscribe</code> that receive a
* {@link Subscription} object.
*
* @author Wagner Macedo
*/
public interface InterceptHandler {
void onConnect(InterceptConnectMessage msg);
void onDisconnect(InterceptDisconnectMessage msg);
void onPublish(InterceptPublishMessage msg);
void onSubscribe(InterceptSubscribeMessage msg);
void onUnsubscribe(InterceptUnsubscribeMessage msg);
}
@@ -0,0 +1,47 @@
/*
* Copyright (c) 2012-2015 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.interception;
import io.moquette.proto.messages.ConnectMessage;
import io.moquette.proto.messages.PublishMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
/**
* This interface is to be used internally by the broker components.
* <p>
* An interface is used instead of a class to allow more flexibility in changing
* an implementation.
* <p>
* Interceptor implementations forward notifications to a <code>InterceptHandler</code>,
* that is normally a field. So, the implementations should act as a proxy to a custom
* intercept handler.
*
* @see InterceptHandler
* @author Wagner Macedo
*/
public interface Interceptor {
void notifyClientConnected(ConnectMessage msg);
void notifyClientDisconnected(String clientID);
void notifyTopicPublished(PublishMessage msg, String clientID);
void notifyTopicSubscribed(Subscription sub);
void notifyTopicUnsubscribed(String topic, String clientID);
}
@@ -0,0 +1,26 @@
package io.moquette.interception.messages;
import io.moquette.proto.messages.AbstractMessage;
/**
* @author Wagner Macedo
*/
public abstract class InterceptAbstractMessage {
private final AbstractMessage msg;
InterceptAbstractMessage(AbstractMessage msg) {
this.msg = msg;
}
public boolean isRetainFlag() {
return msg.isRetainFlag();
}
public boolean isDupFlag() {
return msg.isDupFlag();
}
public AbstractMessage.QOSType getQos() {
return msg.getQos();
}
}
@@ -0,0 +1,71 @@
package io.moquette.interception.messages;
import io.moquette.proto.messages.ConnectMessage;
/**
* @author Wagner Macedo
*/
public class InterceptConnectMessage extends InterceptAbstractMessage {
private final ConnectMessage msg;
public InterceptConnectMessage(ConnectMessage msg) {
super(msg);
this.msg = msg;
}
public String getClientID() {
return msg.getClientID();
}
public boolean isCleanSession() {
return msg.isCleanSession();
}
public int getKeepAlive() {
return msg.getKeepAlive();
}
public boolean isPasswordFlag() {
return msg.isPasswordFlag();
}
public byte getProtocolVersion() {
return msg.getProtocolVersion();
}
public String getProtocolName() {
return msg.getProtocolName();
}
public boolean isUserFlag() {
return msg.isUserFlag();
}
public boolean isWillFlag() {
return msg.isWillFlag();
}
public byte getWillQos() {
return msg.getWillQos();
}
public boolean isWillRetain() {
return msg.isWillRetain();
}
public String getUsername() {
return msg.getUsername();
}
public byte[] getPassword() {
return msg.getPassword();
}
public String getWillTopic() {
return msg.getWillTopic();
}
public byte[] getWillMessage() {
return msg.getWillMessage();
}
}
@@ -0,0 +1,16 @@
package io.moquette.interception.messages;
/**
* @author Wagner Macedo
*/
public class InterceptDisconnectMessage {
private final String clientID;
public InterceptDisconnectMessage(String clientID) {
this.clientID = clientID;
}
public String getClientID() {
return clientID;
}
}
@@ -0,0 +1,31 @@
package io.moquette.interception.messages;
import io.moquette.proto.messages.PublishMessage;
import java.nio.ByteBuffer;
/**
* @author Wagner Macedo
*/
public class InterceptPublishMessage extends InterceptAbstractMessage {
private final PublishMessage msg;
private final String clientID;
public InterceptPublishMessage(PublishMessage msg, String clientID) {
super(msg);
this.msg = msg;
this.clientID = clientID;
}
public String getTopicName() {
return msg.getTopicName();
}
public ByteBuffer getPayload() {
return msg.getPayload();
}
public String getClientID() {
return clientID;
}
}
@@ -0,0 +1,27 @@
package io.moquette.interception.messages;
import io.moquette.proto.messages.AbstractMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
/**
* @author Wagner Macedo
*/
public class InterceptSubscribeMessage {
private final Subscription subscription;
public InterceptSubscribeMessage(Subscription subscription) {
this.subscription = subscription;
}
public String getClientID() {
return subscription.getClientId();
}
public AbstractMessage.QOSType getRequestedQos() {
return subscription.getRequestedQos();
}
public String getTopicFilter() {
return subscription.getTopicFilter();
}
}
@@ -0,0 +1,22 @@
package io.moquette.interception.messages;
/**
* @author Wagner Macedo
*/
public class InterceptUnsubscribeMessage {
private final String topicFilter;
private final String clientID;
public InterceptUnsubscribeMessage(String topicFilter, String clientID) {
this.topicFilter = topicFilter;
this.clientID = clientID;
}
public String getTopicFilter() {
return topicFilter;
}
public String getClientID() {
return clientID;
}
}
@@ -0,0 +1,41 @@
/*
* Copyright (c) 2012-2015 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;
/**
* Value object to maintain the information of single connection, like ClientID, IoSession,
* and other clean session fla.
*
*
* @author andrea
*/
public class ConnectionDescriptor {
public final String clientID;
public final ServerChannel session;
public final boolean cleanSession;
public ConnectionDescriptor(String clientID, ServerChannel session, boolean cleanSession) {
this.clientID = clientID;
this.session = session;
this.cleanSession = cleanSession;
}
@Override
public String toString() {
return "ConnectionDescriptor{" + "clientID=" + clientID + ", cleanSession=" + cleanSession + '}';
}
}
@@ -0,0 +1,25 @@
/*
* Copyright (c) 2012-2015 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;
/**
* Server constants keeper
*/
public class Constants {
public static final String ATTR_CLIENTID = "ClientID";
public static final String CLEAN_SESSION = "cleanSession";
public static final String KEEP_ALIVE = "keepAlive";
}
@@ -0,0 +1,107 @@
/*
* Copyright (c) 2012-2015 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 io.moquette.server.config.MemoryConfig;
import io.moquette.spi.impl.SimpleMessaging;
import io.moquette.server.config.FilesystemConfig;
import io.moquette.server.config.IConfig;
import io.moquette.server.netty.NettyAcceptor;
import io.moquette.spi.impl.ProtocolProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.Properties;
/**
* Launch a configured version of the server.
* @author andrea
*/
public class Server {
private static final Logger LOG = LoggerFactory.getLogger(Server.class);
private ServerAcceptor m_acceptor;
public static void main(String[] args) throws IOException {
final Server server = new Server();
server.startServer();
System.out.println("Server started, version 0.8-SNAPSHOT");
//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
*/
public void startServer() throws IOException {
final IConfig config = new FilesystemConfig();
startServer(config);
}
/**
* Starts Moquette bringing the configuration from the given file
*/
public void startServer(File configFile) throws IOException {
LOG.info("Using m_config file: " + configFile.getAbsolutePath());
final IConfig config = new FilesystemConfig(configFile);
startServer(config);
}
/**
* Starts the server with the given properties.
*
* Its suggested to at least have the following properties:
* <ul>
* <li>port</li>
* <li>password_file</li>
* </ul>
*/
public void startServer(Properties configProps) throws IOException {
final IConfig config = new MemoryConfig(configProps);
startServer(config);
}
/**
* Starts Moquette bringing the configuration files from the given Config implementation.
*/
public void startServer(IConfig config) throws IOException {
final String handlerProp = System.getProperty("intercept.handler");
if (handlerProp != null) {
config.setProperty("intercept.handler", handlerProp);
}
LOG.info("Persistent store file: " + config.getProperty(io.moquette.commons.Constants.PERSISTENT_STORE_PROPERTY_NAME));
final ProtocolProcessor processor = SimpleMessaging.getInstance().init(config);
m_acceptor = new NettyAcceptor();
m_acceptor.initialize(processor, config);
}
public void stopServer() {
LOG.info("Server stopping...");
m_acceptor.close();
SimpleMessaging.getInstance().shutdown();
LOG.info("Server stopped");
}
}
@@ -0,0 +1,32 @@
/*
* Copyright (c) 2012-2015 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 java.io.IOException;
import io.moquette.server.config.IConfig;
import io.moquette.spi.impl.ProtocolProcessor;
/**
*
* @author andrea
*/
public interface ServerAcceptor {
void initialize(ProtocolProcessor processor, IConfig props) throws IOException;
void close();
}
@@ -0,0 +1,35 @@
/*
* Copyright (c) 2012-2015 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 io.netty.util.AttributeKey;
/**
*
* @author andrea
*/
public interface ServerChannel {
Object getAttribute(AttributeKey<Object> key);
void setAttribute(AttributeKey<Object> key, Object value);
void setIdleTime(int idleTime);
void close(boolean immediately);
void write(Object value);
}
@@ -0,0 +1,68 @@
/*
* Copyright (c) 2012-2015 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.config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.text.ParseException;
import java.util.Properties;
/**
* Configuration that loads file from the classpath
*
* @author andrea
*/
public class ClasspathConfig implements IConfig {
private static final Logger LOG = LoggerFactory.getLogger(ClasspathConfig.class);
private final Properties m_properties;
public ClasspathConfig() {
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
InputStream is = classLoader.getResourceAsStream("config/moquette.conf");
if (is == null) {
throw new RuntimeException("Can't locate the resourse \"config/moquette.conf\"");
}
Reader configReader = new InputStreamReader(is);
ConfigurationParser confParser = new ConfigurationParser();
try {
confParser.parse(configReader);
} catch (ParseException pex) {
LOG.warn("An error occurred in parsing configuration, fallback on default configuration", pex);
}
m_properties = confParser.getProperties();
}
@Override
public void setProperty(String name, String value) {
m_properties.setProperty(name, value);
}
@Override
public String getProperty(String name) {
return m_properties.getProperty(name);
}
@Override
public String getProperty(String name, String defaultValue) {
return m_properties.getProperty(name, defaultValue);
}
}
@@ -0,0 +1,123 @@
/*
* Copyright (c) 2012-2015 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.config;
import io.moquette.commons.Constants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.text.ParseException;
import java.util.Properties;
/**
* Mosquitto configuration parser.
*
* A line that at the very first has # is a comment
* Each line has key value format, where the separator used it the space.
*
* @author andrea
*/
class ConfigurationParser {
private static final Logger LOG = LoggerFactory.getLogger(ConfigurationParser.class);
private Properties m_properties = new Properties();
ConfigurationParser() {
createDefaults();
}
private void createDefaults() {
m_properties.put(Constants.PORT_PROPERTY_NAME, Integer.toString(Constants.PORT));
m_properties.put(Constants.HOST_PROPERTY_NAME, Constants.HOST);
m_properties.put(Constants.WEB_SOCKET_PORT_PROPERTY_NAME, Integer.toString(Constants.WEBSOCKET_PORT));
m_properties.put(Constants.PASSWORD_FILE_PROPERTY_NAME, "");
m_properties.put(Constants.PERSISTENT_STORE_PROPERTY_NAME, Constants.DEFAULT_PERSISTENT_PATH);
m_properties.put(Constants.ALLOW_ANONYMOUS_PROPERTY_NAME, true);
m_properties.put(Constants.AUTHENTICATOR_CLASS_NAME, "");
m_properties.put(Constants.AUTHORIZATOR_CLASS_NAME, "");
}
/**
* Parse the configuration from file.
*/
void parse(File file) throws ParseException {
if (file == null) {
LOG.warn("parsing NULL file, so fallback on default configuration!");
return;
}
if (!file.exists()) {
LOG.warn(String.format("parsing not existing file %s, so fallback on default configuration!", file.getAbsolutePath()));
return;
}
try {
FileReader reader = new FileReader(file);
parse(reader);
} catch (FileNotFoundException fex) {
LOG.warn(String.format("parsing not existing file %s, so fallback on default configuration!", file.getAbsolutePath()), fex);
return;
}
}
/**
* Parse the configuration
*
* @throws ParseException if the format is not compliant.
*/
void parse(Reader reader) throws ParseException {
if (reader == null) {
//just log and return default properties
LOG.warn("parsing NULL reader, so fallback on default configuration!");
return;
}
BufferedReader br = new BufferedReader(reader);
String line;
try {
while ((line = br.readLine()) != null) {
int commentMarker = line.indexOf('#');
if (commentMarker != -1) {
if (commentMarker == 0) {
//skip its a comment
continue;
} else {
//it's a malformed comment
throw new ParseException(line, commentMarker);
}
} else {
if (line.isEmpty() || line.matches("^\\s*$")) {
//skip it's a black line
continue;
}
//split till the first space
int delimiterIdx = line.indexOf(' ');
String key = line.substring(0, delimiterIdx).trim();
String value = line.substring(delimiterIdx).trim();
m_properties.put(key, value);
}
}
} catch (IOException ex) {
throw new ParseException("Failed to read", 1);
}
}
Properties getProperties() {
return m_properties;
}
}
@@ -0,0 +1,68 @@
/*
* Copyright (c) 2012-2015 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.config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.text.ParseException;
import java.util.Properties;
/**
* Configuration that loads file from the file system
*
* @author andrea
*/
public class FilesystemConfig implements IConfig {
private static final Logger LOG = LoggerFactory.getLogger(FilesystemConfig.class);
private final Properties m_properties;
public FilesystemConfig(File file) {
ConfigurationParser confParser = new ConfigurationParser();
try {
confParser.parse(file);
} catch (ParseException pex) {
LOG.warn("An error occurred in parsing configuration, fallback on default configuration", pex);
}
m_properties = confParser.getProperties();
}
public FilesystemConfig() {
this(defaultConfigFile());
}
private static File defaultConfigFile() {
String configPath = System.getProperty("moquette.path", null);
return new File(configPath, "config/moquette.conf");
}
@Override
public void setProperty(String name, String value) {
m_properties.setProperty(name, value);
}
@Override
public String getProperty(String name) {
return m_properties.getProperty(name);
}
@Override
public String getProperty(String name, String defaultValue) {
return m_properties.getProperty(name, defaultValue);
}
}
@@ -0,0 +1,29 @@
/*
* Copyright (c) 2012-2015 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.config;
/**
* Base interface for all configuration implementations (filesystem, memory or classpath)
*
* @author andrea
*/
public interface IConfig {
void setProperty(String name, String value);
String getProperty(String name);
String getProperty(String name, String defaultValue);
}
@@ -0,0 +1,64 @@
/*
* Copyright (c) 2012-2015 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.config;
import io.moquette.commons.Constants;
import java.util.Map;
import java.util.Properties;
/**
* Configuration backed by memory.
*
* @author andrea
*/
public class MemoryConfig implements IConfig {
private final Properties m_properties = new Properties();
public MemoryConfig(Properties properties) {
createDefaults();
for (Map.Entry<Object, Object> entrySet : properties.entrySet()) {
m_properties.put(entrySet.getKey(), entrySet.getValue());
}
}
private void createDefaults() {
m_properties.put(Constants.PORT_PROPERTY_NAME, Integer.toString(Constants.PORT));
m_properties.put(Constants.HOST_PROPERTY_NAME, Constants.HOST);
m_properties.put(Constants.WEB_SOCKET_PORT_PROPERTY_NAME, Integer.toString(Constants.WEBSOCKET_PORT));
m_properties.put(Constants.PASSWORD_FILE_PROPERTY_NAME, "");
m_properties.put(Constants.PERSISTENT_STORE_PROPERTY_NAME, Constants.DEFAULT_PERSISTENT_PATH);
m_properties.put(Constants.ALLOW_ANONYMOUS_PROPERTY_NAME, true);
m_properties.put(Constants.AUTHENTICATOR_CLASS_NAME, "");
m_properties.put(Constants.AUTHORIZATOR_CLASS_NAME, "");
}
@Override
public void setProperty(String name, String value) {
m_properties.setProperty(name, value);
}
@Override
public String getProperty(String name) {
return m_properties.getProperty(name);
}
@Override
public String getProperty(String name, String defaultValue) {
return m_properties.getProperty(name, defaultValue);
}
}
@@ -0,0 +1,39 @@
/*
* Copyright (c) 2012-2015 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.netty;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
@Sharable
class MoquetteIdleTimoutHandler extends ChannelDuplexHandler {
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleState e = ((IdleStateEvent) evt).state();
if (e == IdleState.ALL_IDLE) {
//fire a channelInactive to trigger publish of Will
ctx.fireChannelInactive();
ctx.close();
}
} else {
super.userEventTriggered(ctx, evt);
}
}
}
@@ -0,0 +1,366 @@
/*
* Copyright (c) 2012-2015 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.netty;
import io.moquette.commons.Constants;
import io.moquette.parser.netty.MQTTDecoder;
import io.moquette.server.config.IConfig;
import io.moquette.server.netty.metrics.*;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
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.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.Future;
import io.moquette.parser.netty.MQTTEncoder;
import io.moquette.server.ServerAcceptor;
import io.moquette.spi.impl.ProtocolProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import java.io.*;
import java.net.URL;
import java.security.*;
import java.security.cert.CertificateException;
import java.util.List;
/**
*
* @author andrea
*/
public class NettyAcceptor implements ServerAcceptor {
static class WebSocketFrameToByteBufDecoder extends MessageToMessageDecoder<BinaryWebSocketFrame> {
@Override
protected void decode(ChannelHandlerContext chc, BinaryWebSocketFrame frame, List<Object> out) throws Exception {
//convert the frame to a ByteBuf
ByteBuf bb = frame.content();
//System.out.println("WebSocketFrameToByteBufDecoder decode - " + ByteBufUtil.hexDump(bb));
bb.retain();
out.add(bb);
}
}
static class ByteBufToWebSocketFrameEncoder extends MessageToMessageEncoder<ByteBuf> {
@Override
protected void encode(ChannelHandlerContext chc, ByteBuf bb, List<Object> out) throws Exception {
//convert the ByteBuf to a WebSocketFrame
BinaryWebSocketFrame result = new BinaryWebSocketFrame();
//System.out.println("ByteBufToWebSocketFrameEncoder encode - " + ByteBufUtil.hexDump(bb));
result.content().writeBytes(bb);
out.add(result);
}
}
abstract class PipelineInitializer {
abstract void init(ChannelPipeline pipeline) throws Exception;
}
private static final Logger LOG = LoggerFactory.getLogger(NettyAcceptor.class);
EventLoopGroup m_bossGroup;
EventLoopGroup m_workerGroup;
BytesMetricsCollector m_bytesMetricsCollector = new BytesMetricsCollector();
MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector();
@Override
public void initialize(ProtocolProcessor processor, IConfig props) throws IOException {
m_bossGroup = new NioEventLoopGroup();
m_workerGroup = new NioEventLoopGroup();
final NettyMQTTHandler handler = new NettyMQTTHandler(processor);
initializePlainTCPTransport(handler, props);
initializeWebSocketTransport(handler, props);
String sslTcpPortProp = props.getProperty(Constants.SSL_PORT_PROPERTY_NAME);
String wssPortProp = props.getProperty(Constants.WSS_PORT_PROPERTY_NAME);
if (sslTcpPortProp != null || wssPortProp != null) {
SslHandlerFactory sslHandlerFactory = initSSLHandlerFactory(props);
if (!sslHandlerFactory.canCreate()) {
LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks");
return;
}
initializeSSLTCPTransport(handler, props, sslHandlerFactory);
initializeWSSTransport(handler, props, sslHandlerFactory);
}
}
private void initFactory(String host, int port, final PipelineInitializer pipeliner) {
ServerBootstrap b = new ServerBootstrap();
b.group(m_bossGroup, m_workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
try {
pipeliner.init(pipeline);
} catch (Throwable th) {
LOG.error("Severe error during pipeline creation", th);
throw th;
}
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.option(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.TCP_NODELAY, true)
.childOption(ChannelOption.SO_KEEPALIVE, true);
try {
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(host, port);
LOG.info("Server binded host: {}, port: {}", host, port);
f.sync();
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
private void initializePlainTCPTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
int port = Integer.parseInt(props.getProperty(Constants.PORT_PROPERTY_NAME));
initFactory(host, port, new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) {
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
//pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("handler", handler);
}
});
}
private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
String webSocketPortProp = props.getProperty(Constants.WEB_SOCKET_PORT_PROPERTY_NAME);
if (webSocketPortProp == null) {
//Do nothing no WebSocket configured
LOG.info("WebSocket is disabled");
return;
}
int port = Integer.parseInt(webSocketPortProp);
final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
initFactory(host, port, new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) {
pipeline.addLast("httpEncoder", new HttpResponseEncoder());
pipeline.addLast("httpDecoder", new HttpRequestDecoder());
pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqtt, mqttv3.1, mqttv3.1.1"));
pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("handler", handler);
}
});
}
private void initializeSSLTCPTransport(final NettyMQTTHandler handler, IConfig props, final SslHandlerFactory sslHandlerFactory) throws IOException {
String sslPortProp = props.getProperty(Constants.SSL_PORT_PROPERTY_NAME);
if (sslPortProp == null) {
//Do nothing no SSL configured
LOG.info("SSL is disabled");
return;
}
int sslPort = Integer.parseInt(sslPortProp);
LOG.info("Starting SSL on port {}", sslPort);
final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
initFactory(host, sslPort, new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) throws Exception {
pipeline.addLast("ssl", sslHandlerFactory.create());
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
//pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("handler", handler);
}
});
}
private void initializeWSSTransport(final NettyMQTTHandler handler, IConfig props, final SslHandlerFactory sslHandlerFactory) throws IOException {
String sslPortProp = props.getProperty(Constants.WSS_PORT_PROPERTY_NAME);
if (sslPortProp == null) {
//Do nothing no SSL configured
LOG.info("SSL is disabled");
return;
}
int sslPort = Integer.parseInt(sslPortProp);
final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
initFactory(host, sslPort, new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) throws Exception {
pipeline.addLast("ssl", sslHandlerFactory.create());
pipeline.addLast("httpEncoder", new HttpResponseEncoder());
pipeline.addLast("httpDecoder", new HttpRequestDecoder());
pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqtt mqttv3.1, mqttv3.1.1"));
pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("handler", handler);
}
});
}
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();
try {
workerWaiter.await(100);
} catch (InterruptedException iex) {
throw new IllegalStateException(iex);
}
try {
bossWaiter.await(100);
} catch (InterruptedException iex) {
throw new IllegalStateException(iex);
}
MessageMetrics metrics = m_metricsCollector.computeMetrics();
LOG.info("Msg read: {}, msg wrote: {}", metrics.messagesRead(), metrics.messagesWrote());
BytesMetrics bytesMetrics = m_bytesMetricsCollector.computeMetrics();
LOG.info(String.format("Bytes read: %d, bytes wrote: %d", bytesMetrics.readBytes(), bytesMetrics.wroteBytes()));
}
private SslHandlerFactory initSSLHandlerFactory(IConfig props) {
SslHandlerFactory factory = new SslHandlerFactory(props);
return factory.canCreate() ? factory : null;
}
private static class SslHandlerFactory {
private SSLContext sslContext;
public SslHandlerFactory(IConfig props) {
this.sslContext = initSSLContext(props);
}
public boolean canCreate() {
return this.sslContext != null;
}
private SSLContext initSSLContext(IConfig props) {
final String jksPath = props.getProperty(Constants.JKS_PATH_PROPERTY_NAME);
LOG.info("Starting SSL using keystore at {}", 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");
return null;
}
//if we have the port also the jks then keyStorePassword and keyManagerPassword
//has to be defined
final String keyStorePassword = props.getProperty(Constants.KEY_STORE_PASSWORD_PROPERTY_NAME);
final String keyManagerPassword = props.getProperty(Constants.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");
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");
return null;
}
try {
InputStream jksInputStream = jksDatastore(jksPath);
SSLContext serverContext = SSLContext.getInstance("TLS");
final KeyStore ks = KeyStore.getInstance("JKS");
ks.load(jksInputStream, keyStorePassword.toCharArray());
final KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
kmf.init(ks, keyManagerPassword.toCharArray());
serverContext.init(kmf.getKeyManagers(), null, null);
return serverContext;
} catch (NoSuchAlgorithmException | UnrecoverableKeyException | CertificateException | KeyStoreException
| KeyManagementException | IOException ex) {
LOG.error("Can't start SSL layer!", ex);
return null;
}
}
private InputStream jksDatastore(String jksPath) throws FileNotFoundException {
URL jksUrl = getClass().getClassLoader().getResource(jksPath);
if (jksUrl != null) {
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");
File jksFile = new File(jksPath);
if (jksFile.exists()) {
LOG.info("Using {} ", jksFile.getAbsolutePath());
return new FileInputStream(jksFile);
}
LOG.warn("File {} doesn't exists", jksFile.getAbsolutePath());
return null;
}
public ChannelHandler create() {
SSLEngine sslEngine = sslContext.createSSLEngine();
sslEngine.setUseClientMode(false);
return new SslHandler(sslEngine);
}
}
}
@@ -0,0 +1,76 @@
/*
* Copyright (c) 2012-2015 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.netty;
import io.moquette.server.Constants;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import io.moquette.server.ServerChannel;
/**
*
* @author andrea
*/
public class NettyChannel implements ServerChannel {
private ChannelHandlerContext m_channel;
public static final String ATTR_USERNAME = "username";
public static final String ATTR_SESSION_STOLEN = "sessionStolen";
public static final AttributeKey<Object> ATTR_KEY_KEEPALIVE = AttributeKey.valueOf(Constants.KEEP_ALIVE);
public static final AttributeKey<Object> ATTR_KEY_CLEANSESSION = AttributeKey.valueOf(Constants.CLEAN_SESSION);
public static final AttributeKey<Object> ATTR_KEY_CLIENTID = AttributeKey.valueOf(Constants.ATTR_CLIENTID);
public static final AttributeKey<Object> ATTR_KEY_USERNAME = AttributeKey.valueOf(ATTR_USERNAME);
public static final AttributeKey<Object> ATTR_KEY_SESSION_STOLEN = AttributeKey.valueOf(ATTR_SESSION_STOLEN);
NettyChannel(ChannelHandlerContext ctx) {
m_channel = ctx;
}
public Object getAttribute(AttributeKey<Object> key) {
Attribute<Object> attr = m_channel.attr(key);
return attr.get();
}
public void setAttribute(AttributeKey<Object> key, Object value) {
Attribute<Object> attr = m_channel.attr(key);
attr.set(value);
}
public void setIdleTime(int idleTime) {
if (m_channel.pipeline().names().contains("idleStateHandler")) {
m_channel.pipeline().remove("idleStateHandler");
}
m_channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
}
public void close(boolean immediately) {
m_channel.close();
}
public void write(Object value) {
m_channel.writeAndFlush(value);
}
@Override
public String toString() {
String clientID = (String) getAttribute(ATTR_KEY_CLIENTID);
return "session [clientID: "+ clientID +"]";
}
}
@@ -0,0 +1,114 @@
/*
* Copyright (c) 2012-2015 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.netty;
import io.moquette.proto.Utils;
import io.moquette.proto.messages.*;
import io.moquette.spi.impl.ProtocolProcessor;
import static io.moquette.proto.messages.AbstractMessage.*;
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;
/**
*
* @author andrea
*/
@Sharable
public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);
private final ProtocolProcessor m_processor;
public NettyMQTTHandler(ProtocolProcessor processor) {
m_processor = processor;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
AbstractMessage msg = (AbstractMessage) message;
LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType()));
try {
switch (msg.getMessageType()) {
case CONNECT:
m_processor.processConnect(new NettyChannel(ctx), (ConnectMessage) msg);
break;
case SUBSCRIBE:
m_processor.processSubscribe(new NettyChannel(ctx), (SubscribeMessage) msg);
break;
case UNSUBSCRIBE:
m_processor.processUnsubscribe(new NettyChannel(ctx), (UnsubscribeMessage) msg);
break;
case PUBLISH:
m_processor.processPublish(new NettyChannel(ctx), (PublishMessage) msg);
break;
case PUBREC:
m_processor.processPubRec(new NettyChannel(ctx), (PubRecMessage) msg);
break;
case PUBCOMP:
m_processor.processPubComp(new NettyChannel(ctx), (PubCompMessage) msg);
break;
case PUBREL:
m_processor.processPubRel(new NettyChannel(ctx), (PubRelMessage) msg);
break;
case DISCONNECT:
m_processor.processDisconnect(new NettyChannel(ctx));
break;
case PUBACK:
m_processor.processPubAck(new NettyChannel(ctx), (PubAckMessage) msg);
break;
case PINGREQ:
PingRespMessage pingResp = new PingRespMessage();
ctx.writeAndFlush(pingResp);
break;
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
}
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String clientID = (String) NettyUtils.getAttribute(ctx, NettyChannel.ATTR_KEY_CLIENTID);
if (clientID != null && !clientID.isEmpty()) {
//if the channel was of a correctly connected client, inform messaging
//else it was of a not completed CONNECT message or sessionStolen
boolean stolen = false;
Boolean stolenAttr = (Boolean) NettyUtils.getAttribute(ctx, NettyChannel.ATTR_KEY_SESSION_STOLEN);
if (stolenAttr != null && stolenAttr == Boolean.TRUE) {
stolen = stolenAttr;
}
m_processor.processConnectionLost(clientID, stolen);
}
ctx.close(/*false*/);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
if (cause instanceof CorruptedFrameException) {
//something goes bad with decoding
LOG.warn("Error decoding a packet, probably a bad formatted packet, message: " + cause.getMessage());
} else {
LOG.error("Ugly error on networking", cause);
}
ctx.close();
}
}
@@ -0,0 +1,50 @@
/*
* Copyright (c) 2012-2015 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.netty;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
/**
* Some Netty's channels utilities.
*
* @author andrea
*/
public class NettyUtils {
public static Object getAttribute(ChannelHandlerContext channel, AttributeKey<Object> key) {
Attribute<Object> attr = channel.attr(key);
return attr.get();
}
public static void setAttribute(ChannelHandlerContext channel, AttributeKey<Object> key, Object value) {
Attribute<Object> attr = channel.attr(key);
attr.set(value);
}
public static void setIdleTime(ChannelHandlerContext channel, int idleTime) {
if (channel.pipeline().names().contains("idleStateHandler")) {
channel.pipeline().remove("idleStateHandler");
}
if (channel.pipeline().names().contains("idleEventHandler")) {
channel.pipeline().remove("idleEventHandler");
}
channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
}
}
@@ -0,0 +1,37 @@
/*
* Copyright (c) 2012-2015 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.netty.metrics;
public class BytesMetrics {
private long m_readBytes = 0;
private long m_wroteBytes = 0;
void incrementRead(long numBytes) {
m_readBytes += numBytes;
}
void incrementWrote(long numBytes) {
m_wroteBytes += numBytes;
}
public long readBytes() {
return m_readBytes;
}
public long wroteBytes() {
return m_wroteBytes;
}
}
@@ -0,0 +1,42 @@
/*
* Copyright (c) 2012-2015 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.netty.metrics;
import java.util.concurrent.atomic.AtomicLong;
/**
* Collects all the metrics from the various pipeline.
*/
public class BytesMetricsCollector {
private AtomicLong readBytes = new AtomicLong();
private AtomicLong wroteBytes = new AtomicLong();
public BytesMetrics computeMetrics() {
BytesMetrics allMetrics = new BytesMetrics();
allMetrics.incrementRead(readBytes.get());
allMetrics.incrementWrote(wroteBytes.get());
return allMetrics;
}
public void sumReadBytes(long count) {
readBytes.getAndAdd(count);
}
public void sumWroteBytes(long count) {
wroteBytes.getAndAdd(count);
}
}
@@ -0,0 +1,66 @@
/*
* Copyright (c) 2012-2015 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.netty.metrics;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
public class BytesMetricsHandler extends ChannelDuplexHandler {
private static final AttributeKey<BytesMetrics> ATTR_KEY_METRICS = AttributeKey.valueOf("BytesMetrics");
private BytesMetricsCollector m_collector;
public BytesMetricsHandler(BytesMetricsCollector collector) {
m_collector = collector;
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Attribute<BytesMetrics> attr = ctx.attr(ATTR_KEY_METRICS);
attr.set(new BytesMetrics());
super.channelActive(ctx);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
BytesMetrics metrics = ctx.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();
metrics.incrementWrote(((ByteBuf)msg).writableBytes());
ctx.write(msg, promise);
}
@Override
public void close(ChannelHandlerContext ctx,
ChannelPromise promise) throws Exception {
BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
m_collector.sumReadBytes(metrics.readBytes());
m_collector.sumWroteBytes(metrics.wroteBytes());
super.close(ctx, promise);
}
}
@@ -0,0 +1,37 @@
/*
* Copyright (c) 2012-2015 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.netty.metrics;
public class MessageMetrics {
private long m_messagesRead = 0;
private long m_messageWrote = 0;
void incrementRead(long numMessages) {
m_messagesRead += numMessages;
}
void incrementWrote(long numMessages) {
m_messageWrote += numMessages;
}
public long messagesRead() {
return m_messagesRead;
}
public long messagesWrote() {
return m_messageWrote;
}
}
@@ -0,0 +1,41 @@
/*
* Copyright (c) 2012-2015 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.netty.metrics;
import java.util.concurrent.atomic.AtomicLong;
/**
* Collects all the metrics from the various pipeline.
*/
public class MessageMetricsCollector {
private AtomicLong readMsgs = new AtomicLong();
private AtomicLong wroteMsgs = new AtomicLong();
public MessageMetrics computeMetrics() {
MessageMetrics allMetrics = new MessageMetrics();
allMetrics.incrementRead(readMsgs.get());
allMetrics.incrementWrote(wroteMsgs.get());
return allMetrics;
}
public void sumReadMessages(long count) {
readMsgs.getAndAdd(count);
}
public void sumWroteMessages(long count) {
wroteMsgs.getAndAdd(count);
}
}
@@ -0,0 +1,65 @@
/*
* Copyright (c) 2012-2015 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.netty.metrics;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
public class MessageMetricsHandler extends ChannelDuplexHandler {
private static final AttributeKey<MessageMetrics> ATTR_KEY_METRICS = AttributeKey.valueOf("MessageMetrics");
private MessageMetricsCollector m_collector;
public MessageMetricsHandler(MessageMetricsCollector collector) {
m_collector = collector;
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Attribute<MessageMetrics> attr = ctx.attr(ATTR_KEY_METRICS);
attr.set(new MessageMetrics());
super.channelActive(ctx);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
MessageMetrics metrics = ctx.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();
metrics.incrementWrote(1);
ctx.write(msg, promise);
}
@Override
public void close(ChannelHandlerContext ctx,
ChannelPromise promise) throws Exception {
MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
m_collector.sumReadMessages(metrics.messagesRead());
m_collector.sumWroteMessages(metrics.messagesWrote());
super.close(ctx, promise);
}
}
@@ -0,0 +1,178 @@
/*
* Copyright (c) 2012-2015 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.spi;
import io.moquette.proto.messages.AbstractMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
/**
* Model a Session like describe on page 25 of MQTT 3.1.1 specification:
* The Session state in the Server consists of:
* <ul>
* <li>The existence of a Session, even if the rest of the Session state is empty.</li>
* <li>The Client’s subscriptions.</li>
* <li>QoS 1 and QoS 2 messages which have been sent to the Client, but have not been
* completely acknowledged.</li>
* <li>QoS 1 and QoS 2 messages pending transmission to the Client.</li>
* <li>QoS 2 messages which have been received from the Client, but have not been
* completely acknowledged.</li>
* <li>Optionally, QoS 0 messages pending transmission to the Client.</li>
* </ul>
*
* @author andrea
*/
public class ClientSession {
private final static Logger LOG = LoggerFactory.getLogger(ClientSession.class);
public final String clientID;
private final IMessagesStore messagesStore;
private final ISessionsStore m_sessionsStore;
private Set<Subscription> subscriptions = new HashSet<>();
public final boolean cleanSession;
private boolean active = false;
public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore,
boolean cleanSession) {
this.clientID = clientID;
this.messagesStore = messagesStore;
this.m_sessionsStore = sessionsStore;
this.cleanSession = cleanSession;
}
//List of client's subscriptions
//list of messages not acknowledged by client
//list of in-flight messages
/**
* @return the list of messages to be delivered for client related to the session.
* */
public List<IMessagesStore.StoredMessage> storedMessages() {
//read all messages from enqueued store
Collection<String> guids = this.m_sessionsStore.enqueued(clientID);
return messagesStore.listMessagesInSession(guids);
}
/**
* Remove the messages stored in a cleanSession false.
*/
public void removeEnqueued(String guid) {
this.m_sessionsStore.removeEnqueued(this.clientID, guid);
}
@Override
public String toString() {
return "ClientSession{clientID='" + clientID + '\'' +"}";
}
public boolean subscribe(String topicFilter, Subscription newSubscription) {
LOG.info("<{}> subscribed to topicFilter <{}> with QoS {}",
newSubscription.getClientId(), topicFilter,
AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter());
if (!validTopic) {
//send SUBACK with 0x80 for this topic filter
return false;
}
subscriptions.add(newSubscription);
m_sessionsStore.addNewSubscription(newSubscription);
return true;
}
public void unsubscribeFrom(String topicFilter) {
m_sessionsStore.removeSubscription(topicFilter, clientID);
Set<Subscription> subscriptionsToRemove = new HashSet<>();
for (Subscription sub : this.subscriptions) {
if (sub.getTopicFilter().equals(topicFilter)) {
subscriptionsToRemove.add(sub);
}
}
subscriptions.removeAll(subscriptionsToRemove);
}
public void disconnect() {
if (this.cleanSession) {
//cleanup topic subscriptions
cleanSession();
}
//deactivate the session
deactivate();
}
public void cleanSession() {
LOG.info("cleaning old saved subscriptions for client <{}>", this.clientID);
m_sessionsStore.wipeSubscriptions(this.clientID);
//remove also the messages stored of type QoS1/2
messagesStore.dropMessagesInSession(this.clientID);
}
public void activate() {
this.active = true;
this.m_sessionsStore.activate(this.clientID);
}
public void deactivate() {
this.active = false;
this.m_sessionsStore.deactivate(this.clientID);
}
public boolean isActive() {
return this.active;
}
public int nextPacketId() {
return this.messagesStore.nextPacketID(this.clientID);
}
public void inFlightAcknowledged(int messageID) {
m_sessionsStore.inFlightAck(this.clientID, messageID);
}
public void inFlightAckWaiting(String guid, int messageID) {
m_sessionsStore.inFlight(this.clientID, messageID, guid);
}
public void secondPhaseAcknowledged(int messageID) {
m_sessionsStore.secondPhaseAcknowledged(clientID, messageID);
}
public void secondPhaseAckWaiting(int messageID) {
m_sessionsStore.secondPhaseAckWaiting(clientID, messageID);
}
public void enqueueToDeliver(String guid) {
this.m_sessionsStore.bindToDeliver(guid, this.clientID);
}
public IMessagesStore.StoredMessage storedMessage(int messageID) {
final String guid = m_sessionsStore.mapToGuid(clientID, messageID);
return messagesStore.getMessageByGuid(guid);
}
}
@@ -0,0 +1,23 @@
/*
* Copyright (c) 2012-2015 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.spi;
/**
*/
public interface IMatchingCondition {
boolean match(String key);
}
@@ -0,0 +1,147 @@
/*
* Copyright (c) 2012-2015 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.spi;
import java.io.Serializable;
import java.nio.ByteBuffer;
import io.moquette.proto.messages.AbstractMessage;
import java.util.Collection;
import java.util.List;
/**
* Defines the SPI to be implemented by a StorageService that handle persistence of messages
*/
public interface IMessagesStore {
class StoredMessage implements Serializable {
final AbstractMessage.QOSType m_qos;
final byte[] m_payload;
final String m_topic;
private boolean m_retained;
private String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
private Integer m_msgID;
private String m_guid;
public StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) {
m_qos = qos;
m_payload = message;
m_topic = topic;
}
public AbstractMessage.QOSType getQos() {
return m_qos;
}
public ByteBuffer getPayload() {
return (ByteBuffer) ByteBuffer.allocate(m_payload.length).put(m_payload).flip();
}
public String getTopic() {
return m_topic;
}
public void setGuid(String guid) {
this.m_guid = guid;
}
public String getGuid() {
return m_guid;
}
public String getClientID() {
return m_clientID;
}
public void setClientID(String m_clientID) {
this.m_clientID = m_clientID;
}
public void setMessageID(Integer messageID) {
this.m_msgID = messageID;
}
public Integer getMessageID() {
return m_msgID;
}
public ByteBuffer getMessage() {
return ByteBuffer.wrap(m_payload);
}
public void setRetained(boolean retained) {
this.m_retained = retained;
}
public boolean isRetained() {
return m_retained;
}
@Override
public String toString() {
return "PublishEvent{" +
"m_msgID=" + m_msgID +
", clientID='" + m_clientID + '\'' +
", m_retain=" + m_retained +
", m_qos=" + m_qos +
", m_topic='" + m_topic + '\'' +
'}';
}
}
/**
* Used to initialize all persistent store structures
* */
void initStore();
/**
* Persist the message.
* If the message is empty then the topic is cleaned, else it's stored.
*/
void storeRetained(String topic, String guid);
/**
* Return a list of retained messages that satisfy the condition.
*/
Collection<StoredMessage> searchMatching(IMatchingCondition condition);
/**
* Persist the message.
* @return the unique id in the storage (guid).
* */
String storePublishForFuture(StoredMessage evt);
/**
* Return the list of persisted publishes for the given clientID.
* For QoS1 and QoS2 with clean session flag, this method return the list of
* missed publish events while the client was disconnected.
*/
List<StoredMessage> listMessagesInSession(Collection<String> guids);
void dropMessagesInSession(String clientID);
StoredMessage getMessageByGuid(String guid);
/**
* Return the next valid packetIdentifier for the given client session.
* */
int nextPacketID(String clientID);
// void close();
void cleanRetained(String topic);
}
@@ -0,0 +1,27 @@
/*
* Copyright (c) 2012-2015 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.spi;
/**
* Callback interface used to be notified of some events from the input event queue.
*
* It's the abstraction of the messaging stuff attached in after the front protocol
* parsing stuff.
*
* @author andrea
*/
public interface IMessaging {
}
@@ -0,0 +1,98 @@
/*
* Copyright (c) 2012-2015 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.spi;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import io.moquette.spi.impl.subscriptions.Subscription;
/**
* Store used to handle the persistence of the subscriptions tree.
*
* @author andrea
*/
public interface ISessionsStore {
void initStore();
/**
* Add a new subscription to the session
*/
void addNewSubscription(Subscription newSubscription);
/**
* Removed a specific subscription
* */
void removeSubscription(String topic, String clientID);
/**
* Remove all the subscriptions of the session
*/
void wipeSubscriptions(String sessionID);
/**
* Updates the subscriptions set for the clientID
* */
void updateSubscriptions(String clientID, Set<Subscription> subscriptions);
List<Subscription> listAllSubscriptions();
/**
* @return true iff there are subscriptions persisted with clientID
*/
boolean contains(String clientID);
ClientSession createNewSession(String clientID, boolean cleanSession);
/**
* @return the session for the given clientID, null if not found.
* */
ClientSession sessionForClient(String clientID);
void activate(String clientID);
void deactivate(String clientID);
void inFlightAck(String clientID, int messageID);
/**
* Save the binding messageID, clientID <-> guid
* */
void inFlight(String clientID, int messageID, String guid);
/**
* Store the guid to be later published.
* */
void bindToDeliver(String guid, String clientID);
/**
* List the guids for retained messages for the session
* */
Collection<String> enqueued(String clientID);
/**
* Remove form the queue of stored messages for session.
* */
void removeEnqueued(String clientID, String guid);
void secondPhaseAcknowledged(String clientID, int messageID);
void secondPhaseAckWaiting(String clientID, int messageID);
String mapToGuid(String clientID, int messageID);
}
@@ -0,0 +1,109 @@
/*
* Copyright (c) 2012-2015 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.spi.impl;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.Interceptor;
import io.moquette.interception.messages.*;
import io.moquette.proto.messages.ConnectMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.proto.messages.PublishMessage;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* An interceptor that execute the interception tasks asynchronously.
*
* @author Wagner Macedo
*/
final class BrokerInterceptor implements Interceptor {
private final List<InterceptHandler> handlers;
private final ExecutorService executor;
BrokerInterceptor(List<InterceptHandler> handlers) {
this.handlers = handlers;
executor = Executors.newFixedThreadPool(1);
}
/**
* Shutdown graciously the executor service
*/
void stop() {
executor.shutdown();
}
@Override
public void notifyClientConnected(final ConnectMessage msg) {
for (final InterceptHandler handler : this.handlers) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onConnect(new InterceptConnectMessage(msg));
}
});
}
}
@Override
public void notifyClientDisconnected(final String clientID) {
for (final InterceptHandler handler : this.handlers) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onDisconnect(new InterceptDisconnectMessage(clientID));
}
});
}
}
@Override
public void notifyTopicPublished(final PublishMessage msg, final String clientID) {
for (final InterceptHandler handler : this.handlers) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onPublish(new InterceptPublishMessage(msg, clientID));
}
});
}
}
@Override
public void notifyTopicSubscribed(final Subscription sub) {
for (final InterceptHandler handler : this.handlers) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onSubscribe(new InterceptSubscribeMessage(sub));
}
});
}
}
@Override
public void notifyTopicUnsubscribed(final String topic, final String clientID) {
for (final InterceptHandler handler : this.handlers) {
executor.execute(new Runnable() {
@Override
public void run() {
handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID));
}
});
}
}
}
@@ -0,0 +1,32 @@
/*
* Copyright (c) 2012-2015 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.spi.impl;
import java.nio.ByteBuffer;
/**
*
* @author andrea
*/
class DebugUtils {
static String payload2Str(ByteBuffer content) {
byte[] b = new byte[content.remaining()];
content.mark();
content.get(b);
content.reset();
return new String(b);
}
}
@@ -0,0 +1,651 @@
/*
* Copyright (c) 2012-2015 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.spi.impl;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.ServerChannel;
import io.moquette.server.netty.NettyChannel;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.security.IAuthenticator;
import io.moquette.spi.impl.security.IAuthorizator;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Subscription;
import static io.moquette.parser.netty.Utils.VERSION_3_1;
import static io.moquette.parser.netty.Utils.VERSION_3_1_1;
import io.moquette.proto.messages.AbstractMessage;
import io.moquette.proto.messages.AbstractMessage.QOSType;
import io.moquette.proto.messages.ConnAckMessage;
import io.moquette.proto.messages.ConnectMessage;
import io.moquette.proto.messages.PubAckMessage;
import io.moquette.proto.messages.PubCompMessage;
import io.moquette.proto.messages.PubRecMessage;
import io.moquette.proto.messages.PubRelMessage;
import io.moquette.proto.messages.PublishMessage;
import io.moquette.proto.messages.SubAckMessage;
import io.moquette.proto.messages.SubscribeMessage;
import io.moquette.proto.messages.UnsubAckMessage;
import io.moquette.proto.messages.UnsubscribeMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Class responsible to handle the logic of MQTT protocol it's the director of
* the protocol execution.
*
* Used by the front facing class SimpleMessaging.
*
* @author andrea
*/
public class ProtocolProcessor {
static final class WillMessage {
private final String topic;
private final ByteBuffer payload;
private final boolean retained;
private final QOSType qos;
public WillMessage(String topic, ByteBuffer payload, boolean retained, QOSType qos) {
this.topic = topic;
this.payload = payload;
this.retained = retained;
this.qos = qos;
}
public String getTopic() {
return topic;
}
public ByteBuffer getPayload() {
return payload;
}
public boolean isRetained() {
return retained;
}
public QOSType getQos() {
return qos;
}
}
private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
private Map<String, ConnectionDescriptor> m_clientIDs;
private SubscriptionsStore subscriptions;
private boolean allowAnonymous;
private IAuthorizator m_authorizator;
private IMessagesStore m_messagesStore;
private ISessionsStore m_sessionsStore;
private IAuthenticator m_authenticator;
private BrokerInterceptor m_interceptor;
//maps clientID to Will testament, if specified on CONNECT
private Map<String, WillMessage> m_willStore = new ConcurrentHashMap<>();
ProtocolProcessor() {}
/**
* @param subscriptions the subscription store where are stored all the existing
* clients subscriptions.
* @param storageService the persistent store to use for save/load of messages
* for QoS1 and QoS2 handling.
* @param sessionsStore the clients sessions store, used to persist subscriptions.
* @param authenticator the authenticator used in connect messages.
* @param allowAnonymous true connection to clients without credentials.
* @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,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) {
this.m_clientIDs = new ConcurrentHashMap<>();
this.m_interceptor = interceptor;
this.subscriptions = subscriptions;
this.allowAnonymous = allowAnonymous;
m_authorizator = authorizator;
LOG.debug("subscription tree on init {}", subscriptions.dumpTree());
m_authenticator = authenticator;
m_messagesStore = storageService;
m_sessionsStore = sessionsStore;
}
public void processConnect(ServerChannel session, ConnectMessage msg) {
LOG.debug("CONNECT for client <{}>", msg.getClientID());
if (msg.getProtocolVersion() != VERSION_3_1 && msg.getProtocolVersion() != VERSION_3_1_1) {
ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
LOG.warn("processConnect sent bad proto ConnAck");
session.write(badProto);
session.close(false);
return;
}
if (msg.getClientID() == null || msg.getClientID().length() == 0) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
session.write(okResp);
m_interceptor.notifyClientConnected(msg);
return;
}
//handle user authentication
if (msg.isUserFlag()) {
byte[] pwd = null;
if (msg.isPasswordFlag()) {
pwd = msg.getPassword();
} else if (!this.allowAnonymous) {
failedCredentials(session);
return;
}
if (!m_authenticator.checkValid(msg.getUsername(), pwd)) {
failedCredentials(session);
return;
}
session.setAttribute(NettyChannel.ATTR_KEY_USERNAME, msg.getUsername());
} else if (!this.allowAnonymous) {
failedCredentials(session);
return;
}
//if an old client with the same ID already exists close its session.
if (m_clientIDs.containsKey(msg.getClientID())) {
LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID());
//clean the subscriptions if the old used a cleanSession = true
ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).session;
ClientSession oldClientSession = m_sessionsStore.sessionForClient(msg.getClientID());
oldClientSession.disconnect();
oldSession.setAttribute(NettyChannel.ATTR_KEY_SESSION_STOLEN, true);
oldSession.close(false);
LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID());
}
ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession());
m_clientIDs.put(msg.getClientID(), connDescr);
int keepAlive = msg.getKeepAlive();
LOG.debug("Connect with keepAlive {} s", keepAlive);
session.setAttribute(NettyChannel.ATTR_KEY_KEEPALIVE, keepAlive);
session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, msg.isCleanSession());
//used to track the client in the subscription and publishing phases.
session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, msg.getClientID());
LOG.debug("Connect create session <{}>", session);
session.setIdleTime(Math.round(keepAlive * 1.5f));
//Handle will flag
if (msg.isWillFlag()) {
AbstractMessage.QOSType willQos = AbstractMessage.QOSType.valueOf(msg.getWillQos());
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);
}
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
ClientSession clientSession = m_sessionsStore.sessionForClient(msg.getClientID());
boolean isSessionAlreadyStored = clientSession != null;
if (!msg.isCleanSession() && isSessionAlreadyStored) {
okResp.setSessionPresent(true);
}
session.write(okResp);
m_interceptor.notifyClientConnected(msg);
if (!isSessionAlreadyStored) {
LOG.info("Create persistent session for clientID <{}>", msg.getClientID());
clientSession = m_sessionsStore.createNewSession(msg.getClientID(), msg.isCleanSession());
}
clientSession.activate();
if (msg.isCleanSession()) {
clientSession.cleanSession();
}
LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
if (!msg.isCleanSession()) {
//force the republish of stored QoS1 and QoS2
republishStoredInSession(clientSession);
}
LOG.info("CONNECT processed");
}
private void failedCredentials(ServerChannel session) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
session.write(okResp);
session.close(false);
}
/**
* Republish QoS1 and QoS2 messages stored into the session for the clientID.
* */
private void republishStoredInSession(ClientSession clientSession) {
LOG.trace("republishStoredInSession for client <{}>", clientSession);
List<IMessagesStore.StoredMessage> publishedEvents = clientSession.storedMessages();
if (publishedEvents.isEmpty()) {
LOG.info("No stored messages for client <{}>", clientSession.clientID);
return;
}
LOG.info("republishing stored messages to client <{}>", clientSession.clientID);
for (IMessagesStore.StoredMessage pubEvt : publishedEvents) {
//TODO put in flight zone
directSend(clientSession, pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID());
clientSession.removeEnqueued(pubEvt.getGuid());
}
}
public void processPubAck(ServerChannel session, PubAckMessage msg) {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
int messageID = msg.getMessageID();
//Remove the message from message store
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
targetSession.inFlightAcknowledged(messageID);
}
private static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) {
IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(msg.getPayload().array(), msg.getQos(), msg.getTopicName());
stored.setRetained(msg.isRetainFlag());
stored.setMessageID(msg.getMessageID());
return stored;
}
private static IMessagesStore.StoredMessage asStoredMessage(WillMessage will) {
IMessagesStore.StoredMessage pub = new IMessagesStore.StoredMessage(will.getPayload().array(), will.getQos(), will.getTopic());
pub.setRetained(will.isRetained());
return pub;
}
public void processPublish(ServerChannel session, PublishMessage msg) {
LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg);
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
final String topic = msg.getTopicName();
//check if the topic can be wrote
String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME);
if (!m_authorizator.canWrite(topic, user, clientID)) {
LOG.debug("topic {} doesn't have write credentials", topic);
return;
}
final AbstractMessage.QOSType qos = msg.getQos();
boolean retain = msg.isRetainFlag();
final Integer messageID = msg.getMessageID();
LOG.info("PUBLISH from clientID <{}> on topic <{}> with QoS {}", clientID, topic, qos);
String guid = null;
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
toStoreMsg.setClientID(clientID);
if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0
forward2Subscribers(toStoreMsg);
} else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1
forward2Subscribers(toStoreMsg);
sendPubAck(clientID, messageID);
LOG.debug("replying with PubAck to MSG ID {}", messageID);
} else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
sendPubRec(clientID, messageID);
//Next the client will send us a pub rel
//NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
}
if (retain) {
if (qos == AbstractMessage.QOSType.MOST_ONE) {
//QoS == 0 && retain => clean old retained
m_messagesStore.cleanRetained(topic);
} else {
if (!msg.getPayload().hasRemaining()) {
m_messagesStore.cleanRetained(topic);
} else {
if (guid == null) {
//before wasn't stored
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
}
m_messagesStore.storeRetained(topic, guid);
}
}
}
m_interceptor.notifyTopicPublished(msg, clientID);
}
/**
* Specialized version to publish will testament message.
*/
private void forwardPublishWill(WillMessage will, String clientID) {
//it has just to publish the message downstream to the subscribers
//NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1
Integer messageId = null;
if (will.getQos() != AbstractMessage.QOSType.MOST_ONE) {
messageId = m_messagesStore.nextPacketID(clientID);
}
IMessagesStore.StoredMessage tobeStored = asStoredMessage(will);
tobeStored.setClientID(clientID);
tobeStored.setMessageID(messageId);
forward2Subscribers(tobeStored);
}
/**
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */
//TODO rename to route2Subscribers
void forward2Subscribers(IMessagesStore.StoredMessage pubEvt) {
final String topic = pubEvt.getTopic();
final AbstractMessage.QOSType publishingQos = pubEvt.getQos();
final ByteBuffer origMessage = pubEvt.getMessage();
LOG.debug("forward2Subscribers republishing to existing subscribers that matches the topic {}", topic);
if (LOG.isDebugEnabled()) {
LOG.debug("content <{}>", DebugUtils.payload2Str(origMessage));
LOG.debug("subscription tree {}", subscriptions.dumpTree());
}
//if QoS 1 or 2 store the message
String guid = null;
if (publishingQos == QOSType.EXACTLY_ONCE || publishingQos == QOSType.LEAST_ONE) {
guid = m_messagesStore.storePublishForFuture(pubEvt);
}
for (final Subscription sub : subscriptions.matches(topic)) {
AbstractMessage.QOSType qos = publishingQos;
if (qos.byteValue() > sub.getRequestedQos().byteValue()) {
qos = sub.getRequestedQos();
}
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
sub.getClientId(), sub.getTopicFilter(), qos, targetSession.isActive());
ByteBuffer message = origMessage.duplicate();
if (qos == AbstractMessage.QOSType.MOST_ONE && targetSession.isActive()) {
//QoS 0
directSend(targetSession, topic, qos, message, false, null);
} else {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
if (!sub.isCleanSession() && !targetSession.isActive()) {
//store the message in targetSession queue to deliver
targetSession.enqueueToDeliver(guid);
} else {
//publish
if (targetSession.isActive()) {
int messageId = targetSession.nextPacketId();
targetSession.inFlightAckWaiting(guid, messageId);
directSend(targetSession, topic, qos, message, false, messageId);
}
}
}
}
}
protected void directSend(ClientSession clientsession, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) {
String clientId = clientsession.clientID;
LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", clientId, topic, qos, retained, messageID);
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(retained);
pubMessage.setTopicName(topic);
pubMessage.setQos(qos);
pubMessage.setPayload(message);
LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
if (LOG.isDebugEnabled()) {
LOG.debug("content <{}>", DebugUtils.payload2Str(message));
}
//set the PacketIdentifier only for QoS > 0
if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) {
pubMessage.setMessageID(messageID);
} else {
if (messageID != null) {
throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message with PacketIdentifier: " + messageID);
}
}
if (m_clientIDs == null) {
throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
}
LOG.debug("clientIDs are {}", m_clientIDs);
if (m_clientIDs.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, m_clientIDs));
}
ServerChannel session = m_clientIDs.get(clientId).session;
LOG.debug("Session for clientId {} is {}", clientId, session);
String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME);
if (!m_authorizator.canRead(topic, user, clientId)) {
LOG.debug("topic {} doesn't have read credentials", topic);
return;
}
session.write(pubMessage);
}
private void sendPubRec(String clientID, int messageID) {
LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID);
PubRecMessage pubRecMessage = new PubRecMessage();
pubRecMessage.setMessageID(messageID);
m_clientIDs.get(clientID).session.write(pubRecMessage);
}
private void sendPubAck(String clientId, int messageID) {
LOG.trace("sendPubAck invoked");
PubAckMessage pubAckMessage = new PubAckMessage();
pubAckMessage.setMessageID(messageID);
try {
if (m_clientIDs == null) {
throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!");
}
LOG.debug("clientIDs are {}", m_clientIDs);
if (m_clientIDs.get(clientId) == null) {
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs));
}
m_clientIDs.get(clientId).session.write(pubAckMessage);
} catch(Throwable t) {
LOG.error(null, t);
}
}
/**
* Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish
* to all interested subscribers.
* */
public void processPubRel(ServerChannel session, PubRelMessage msg) {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
int messageID = msg.getMessageID();
LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID);
forward2Subscribers(evt);
if (evt.isRetained()) {
final String topic = evt.getTopic();
if (!evt.getMessage().hasRemaining()) {
m_messagesStore.cleanRetained(topic);
} else {
m_messagesStore.storeRetained(topic, evt.getGuid());
}
}
sendPubComp(clientID, messageID);
}
private void sendPubComp(String clientID, int messageID) {
LOG.debug("PUB <--PUBCOMP-- SRV sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
PubCompMessage pubCompMessage = new PubCompMessage();
pubCompMessage.setMessageID(messageID);
m_clientIDs.get(clientID).session.write(pubCompMessage);
}
public void processPubRec(ServerChannel session, PubRecMessage msg) {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
int messageID = msg.getMessageID();
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
//remove from the inflight and move to the QoS2 second phase queue
targetSession.inFlightAcknowledged(messageID);
targetSession.secondPhaseAckWaiting(messageID);
//once received a PUBREC reply with a PUBREL(messageID)
LOG.debug("\t\tSRV <--PUBREC-- SUB processPubRec invoked for clientID {} ad messageID {}", clientID, messageID);
PubRelMessage pubRelMessage = new PubRelMessage();
pubRelMessage.setMessageID(messageID);
pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE);
session.write(pubRelMessage);
}
public void processPubComp(ServerChannel session, PubCompMessage msg) {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
int messageID = msg.getMessageID();
LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
//once received the PUBCOMP then remove the message from the temp memory
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
targetSession.secondPhaseAcknowledged(messageID);
}
public void processDisconnect(ServerChannel session) throws InterruptedException {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION);
LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
clientSession.disconnect();
m_clientIDs.remove(clientID);
session.close(true);
//cleanup the will store
m_willStore.remove(clientID);
m_interceptor.notifyClientDisconnected(clientID);
LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession);
}
public void processConnectionLost(String clientID, boolean sessionStolen) {
//If already removed a disconnect message was already processed for this clientID
if (sessionStolen && m_clientIDs.remove(clientID) != null) {
//de-activate the subscriptions for this ClientID
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
clientSession.deactivate();
LOG.info("Lost connection with client <{}>", clientID);
}
//publish the Will message (if any) for the clientID
if (!sessionStolen && m_willStore.containsKey(clientID)) {
WillMessage will = m_willStore.get(clientID);
forwardPublishWill(will, clientID);
m_willStore.remove(clientID);
}
}
/**
* Remove the clientID from topic subscription, if not previously subscribed,
* doesn't reply any error
*/
public void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) {
List<String> topics = msg.topicFilters();
int messageID = msg.getMessageID();
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
for (String topic : topics) {
boolean validTopic = SubscriptionsStore.validate(topic);
if (!validTopic) {
//close the connection, not valid topicFilter is a protocol violation
session.close(true);
LOG.warn("UNSUBSCRIBE found an invalid topic filter <{}> for clientID <{}>", topic, clientID);
return;
}
subscriptions.removeSubscription(topic, clientID);
clientSession.unsubscribeFrom(topic);
m_interceptor.notifyTopicUnsubscribed(topic, clientID);
}
//ack the client
UnsubAckMessage ackMessage = new UnsubAckMessage();
ackMessage.setMessageID(messageID);
LOG.info("replying with UnsubAck to MSG ID {}", messageID);
session.write(ackMessage);
}
public void processSubscribe(ServerChannel session, SubscribeMessage msg) {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION);
LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID());
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
//ack the client
SubAckMessage ackMessage = new SubAckMessage();
ackMessage.setMessageID(msg.getMessageID());
List<Subscription> newSubscriptions = new ArrayList<>();
for (SubscribeMessage.Couple req : msg.subscriptions()) {
AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.getQos());
Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos, cleanSession);
//boolean valid = subscribeSingleTopic(newSubscription, req.getTopicFilter());
boolean valid = clientSession.subscribe(req.getTopicFilter(), newSubscription);
ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE);
if (valid) {
newSubscriptions.add(newSubscription);
}
}
//save session, persist subscriptions from session
LOG.debug("SUBACK for packetID {}", msg.getMessageID());
session.write(ackMessage);
//fire the publish
for(Subscription subscription : newSubscriptions) {
subscribeSingleTopic(subscription, subscription.getTopicFilter());
}
}
private boolean subscribeSingleTopic(Subscription newSubscription, final String topic) {
subscriptions.add(newSubscription);
//scans retained messages to be published to the new subscription
//TODO this is ugly, it does a linear scan on potential big dataset
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
public boolean match(String key) {
return SubscriptionsStore.matchTopics(key, topic);
}
});
ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId());
for (IMessagesStore.StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("send publish message for topic {}", topic);
//forwardPublishQoS0(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null :
m_messagesStore.nextPacketID(newSubscription.getClientId());
directSend(targetSession, storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID);
}
//notify the Observables
m_interceptor.notifyTopicSubscribed(newSubscription);
return true;
}
}
@@ -0,0 +1,178 @@
/*
* Copyright (c) 2012-2015 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.spi.impl;
import io.moquette.spi.IMessagesStore;
import io.moquette.commons.Constants;
import io.moquette.interception.InterceptHandler;
import io.moquette.server.config.IConfig;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.security.*;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.persistence.MapDBPersistentStore;
import static io.moquette.commons.Constants.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
/**
*
* Singleton class that orchestrate the execution of the protocol.
*
* It's main responsibility is instantiate the ProtocolProcessor.
*
* @author andrea
*/
public class SimpleMessaging {
private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class);
private SubscriptionsStore subscriptions;
private MapDBPersistentStore m_mapStorage;
private BrokerInterceptor m_interceptor;
private static SimpleMessaging INSTANCE;
private final ProtocolProcessor m_processor = new ProtocolProcessor();
private SimpleMessaging() {
}
public static SimpleMessaging getInstance() {
if (INSTANCE == null) {
INSTANCE = new SimpleMessaging();
}
return INSTANCE;
}
public ProtocolProcessor init(IConfig props) {
subscriptions = new SubscriptionsStore();
//TODO use a property to select the storage path
m_mapStorage = new MapDBPersistentStore(props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, ""));
m_mapStorage.initStore();
IMessagesStore messagesStore = m_mapStorage.messagesStore();
ISessionsStore sessionsStore = m_mapStorage.sessionsStore(messagesStore);
List<InterceptHandler> observers = new ArrayList<>();
String interceptorClassName = props.getProperty("intercept.handler");
if (interceptorClassName != null && !interceptorClassName.isEmpty()) {
try {
InterceptHandler handler = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).newInstance();
observers.add(handler);
} catch (Throwable ex) {
LOG.error("Can't load the intercept handler {}", ex);
}
}
m_interceptor = new BrokerInterceptor(observers);
subscriptions.init(sessionsStore);
String configPath = System.getProperty("moquette.path", null);
String authenticatorClassName = props.getProperty(Constants.AUTHENTICATOR_CLASS_NAME, "");
IAuthenticator authenticator = null;
if (!authenticatorClassName.isEmpty()) {
authenticator = (IAuthenticator)loadClass(authenticatorClassName, IAuthenticator.class);
LOG.info("Loaded custom authenticator {}", authenticatorClassName);
}
if (authenticator == null) {
String passwdPath = props.getProperty(PASSWORD_FILE_PROPERTY_NAME, "");
if (passwdPath.isEmpty()) {
authenticator = new AcceptAllAuthenticator();
} else {
authenticator = new FileAuthenticator(configPath, passwdPath);
}
}
IAuthorizator authorizator = null;
String authorizatorClassName = props.getProperty(Constants.AUTHORIZATOR_CLASS_NAME, "");
if (!authorizatorClassName.isEmpty()) {
authorizator = (IAuthorizator)loadClass(authorizatorClassName, IAuthorizator.class);
LOG.info("Loaded custom authorizator {}", authorizatorClassName);
}
if (authorizator == null) {
String aclFilePath = props.getProperty(ACL_FILE_PROPERTY_NAME, "");
if (aclFilePath != null && !aclFilePath.isEmpty()) {
authorizator = new DenyAllAuthorizator();
File aclFile = new File(configPath, aclFilePath);
try {
authorizator = ACLFileParser.parse(aclFile);
} catch (ParseException pex) {
LOG.error(String.format("Format error in parsing acl file %s", aclFile), pex);
}
LOG.info("Using acl file defined at path {}", aclFilePath);
} else {
authorizator = new PermitAllAuthorizator();
LOG.info("Starting without ACL definition");
}
}
boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor);
return m_processor;
}
private Object loadClass(String className, Class<?> cls) {
Object instance = null;
try {
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 {
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);
}
return instance;
}
public void shutdown() {
this.m_mapStorage.close();
}
}
@@ -0,0 +1,16 @@
package io.moquette.spi.impl;
import java.util.Map;
/**
* Utility static methods, like Map get with default value, or elvis operator.
*/
public class Utils {
public static <T, K> T defaultGet(Map<K, T> map, K key, T defaultValue) {
T value = map.get(key);
if (value != null) {
return value;
}
return defaultValue;
}
}
@@ -0,0 +1,96 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.security;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.text.ParseException;
/**
* Parses the acl configuration file.
* If a line starts with # it's comment.
* Blank lines are skipped. The format is "topic [read|write|readwrite] <topic name>"
*
* @author andrea
* */
public class ACLFileParser {
private static final Logger LOG = LoggerFactory.getLogger(ACLFileParser.class);
/**
* Parse the configuration from file.
*/
public static AuthorizationsCollector parse(File file) throws ParseException {
if (file == null) {
LOG.warn("parsing NULL file, so fallback on default configuration!");
return AuthorizationsCollector.emptyImmutableCollector();
}
if (!file.exists()) {
LOG.warn(String.format("parsing not existing file %s, so fallback on default configuration!", file.getAbsolutePath()));
return AuthorizationsCollector.emptyImmutableCollector();
}
try {
FileReader reader = new FileReader(file);
return parse(reader);
} catch (FileNotFoundException fex) {
LOG.warn(String.format("parsing not existing file %s, so fallback on default configuration!", file.getAbsolutePath()), fex);
return AuthorizationsCollector.emptyImmutableCollector();
}
}
/**
* Parse the ACL configuration file
*
* @throws java.text.ParseException if the format is not compliant.
*/
public static AuthorizationsCollector parse(Reader reader) throws ParseException {
if (reader == null) {
//just log and return default properties
LOG.warn("parsing NULL reader, so fallback on default configuration!");
return AuthorizationsCollector.emptyImmutableCollector();
}
BufferedReader br = new BufferedReader(reader);
String line;
AuthorizationsCollector collector = new AuthorizationsCollector();
try {
while ((line = br.readLine()) != null) {
int commentMarker = line.indexOf('#');
if (commentMarker != -1) {
if (commentMarker == 0) {
//skip its a comment
continue;
} else {
//it's a malformed comment
throw new ParseException(line, commentMarker);
}
} else {
if (line.isEmpty() || line.matches("^\\s*$")) {
//skip it's a black line
continue;
}
collector.parse(line);
}
}
} catch (IOException ex) {
throw new ParseException("Failed to read", 1);
}
return collector;
}
}
@@ -0,0 +1,10 @@
package io.moquette.spi.impl.security;
/**
* Created by andrea on 8/23/14.
*/
public class AcceptAllAuthenticator implements IAuthenticator {
public boolean checkValid(String username, byte[] password) {
return true;
}
}
@@ -0,0 +1,68 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.security;
import static io.moquette.spi.impl.security.Authorization.Permission.READWRITE;
/**
* Carries the read/write authorization to topics for the users.
*
* @author andrea
*/
public class Authorization {
protected final String topic;
protected final Permission permission;
/**
* Access rights
* */
enum Permission {
READ, WRITE, READWRITE
}
Authorization(String topic) {
this(topic, Permission.READWRITE);
}
Authorization(String topic, Permission permission) {
this.topic = topic;
this.permission = permission;
}
public boolean grant(Permission desiredPermission) {
return permission == desiredPermission || permission == READWRITE;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Authorization that = (Authorization) o;
if (permission != that.permission) return false;
if (!topic.equals(that.topic)) return false;
return true;
}
@Override
public int hashCode() {
int result = topic.hashCode();
result = 31 * result + permission.hashCode();
return result;
}
}
@@ -0,0 +1,165 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.security;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.text.ParseException;
import java.util.*;
/**
* Used by the ACLFileParser to push all authorizations it finds.
* ACLAuthorizator uses it in read mode to check it topics matches the ACLs.
*
* Not thread safe.
*
* @author andrea
*/
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 boolean m_parsingUsersSpecificSection = false;
private boolean m_parsingPatternSpecificSection = false;
private String m_currentUser = "";
static final AuthorizationsCollector emptyImmutableCollector() {
AuthorizationsCollector coll = new AuthorizationsCollector();
coll.m_globalAuthorizations = Collections.emptyList();
coll.m_patternAuthorizations = Collections.emptyList();
coll.m_userAuthorizations = Collections.emptyMap();
return coll;
}
void parse(String line) throws ParseException {
Authorization acl = parseAuthLine(line);
if (acl == null) {
//skip it's a user
return;
}
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());
}
List<Authorization> userAuths = m_userAuthorizations.get(m_currentUser);
userAuths.add(acl);
} else if (m_parsingPatternSpecificSection) {
m_patternAuthorizations.add(acl);
} else {
m_globalAuthorizations.add(acl);
}
}
protected Authorization parseAuthLine(String line) throws ParseException {
String[] tokens = line.split("\\s+");
String keyword = tokens[0].toLowerCase();
switch (keyword) {
case "topic":
return createAuthorization(line, tokens);
case "user":
m_parsingUsersSpecificSection = true;
m_currentUser = tokens[1];
m_parsingPatternSpecificSection = false;
return null;
case "pattern":
m_parsingUsersSpecificSection = false;
m_currentUser = "";
m_parsingPatternSpecificSection = true;
return createAuthorization(line, tokens);
default:
throw new ParseException(String.format("invalid line definition found %s", line), 1);
}
}
private Authorization createAuthorization(String line, String[] tokens) throws ParseException {
if (tokens.length > 2) {
//if the tokenized lines has 3 token the second must be the permission
try {
Authorization.Permission permission = Authorization.Permission.valueOf(tokens[1].toUpperCase());
//bring topic with all original spacing
String topic = line.substring(line.indexOf(tokens[2]));
return new Authorization(topic, permission);
} catch (IllegalArgumentException iaex) {
throw new ParseException("invalid permission token", 1);
}
}
String topic = tokens[1];
return new Authorization(topic);
}
@Override
public boolean canWrite(String topic, String user, String client) {
return canDoOperation(topic, Authorization.Permission.WRITE, user, client);
}
@Override
public boolean canRead(String topic, String user, String client) {
return canDoOperation(topic, Authorization.Permission.READ, user, client);
}
private boolean canDoOperation(String topic, Authorization.Permission permission, String username, String client) {
if (matchACL(m_globalAuthorizations, topic, permission)) {
return true;
}
if (isNotEmpty(client) || isNotEmpty(username)) {
for (Authorization auth : m_patternAuthorizations) {
String substitutedTopic = auth.topic.replace("%c", client).replace("%u", username);
if (auth.grant(permission)) {
if (SubscriptionsStore.matchTopics(topic, substitutedTopic)) {
return true;
}
}
}
}
if (isNotEmpty(username)) {
if (m_userAuthorizations.containsKey(username)) {
List<Authorization> auths = m_userAuthorizations.get(username);
if (matchACL(auths, topic, permission)) {
return true;
}
}
}
return false;
}
private boolean matchACL(List<Authorization> auths, String topic, Authorization.Permission permission) {
for (Authorization auth : auths) {
if (auth.grant(permission)) {
if (SubscriptionsStore.matchTopics(topic, auth.topic)) {
return true;
}
}
}
return false;
}
private boolean isNotEmpty(String client) {
return client != null && !client.isEmpty();
}
public boolean isEmpty() {
return m_globalAuthorizations.isEmpty();
}
}
@@ -0,0 +1,31 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.security;
/**
* @author andrea
*/
public class DenyAllAuthorizator implements IAuthorizator {
@Override
public boolean canWrite(String topic, String user, String client) {
return true;
}
@Override
public boolean canRead(String topic, String user, String client) {
return true;
}
}
@@ -0,0 +1,124 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.security;
import org.apache.commons.codec.binary.Hex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.text.ParseException;
import java.util.HashMap;
import java.util.Map;
/**
* Load user credentials from a text file.
* Each line of the file is formatted as "<username>:<sha256(password)>". The username mustn't contains : char.
*
* To encode your password from command line on Linux systems, you could use:
* <pre>
* echo -n "yourpassword" | sha256sum
* </pre>
* NB -n is important because echo append a newline by default at the of string. -n avoid this behaviour.
*
* @author andrea
*/
public class FileAuthenticator implements IAuthenticator {
private static final Logger LOG = LoggerFactory.getLogger(FileAuthenticator.class);
private Map<String, String> m_identities = new HashMap<>();
private MessageDigest m_digest;
public FileAuthenticator(String parent, String filePath) {
File file = new File(parent, filePath);
LOG.info("Loading password file: " + file);
if (file.isDirectory()) {
LOG.warn(String.format("Bad file reference %s is a directory", file));
return;
}
try {
this.m_digest = MessageDigest.getInstance("SHA-256");
} catch (NoSuchAlgorithmException nsaex) {
LOG.error("Can't find SHA-256 for password encoding", nsaex);
throw new RuntimeException(nsaex);
}
try {
FileReader reader = new FileReader(file);
parse(reader);
} catch (FileNotFoundException fex) {
LOG.warn(String.format("Parsing not existing file %s", file), fex);
} catch (ParseException pex) {
LOG.warn(String.format("Format error in parsing password file %s", file), pex);
}
}
private void parse(Reader reader) throws ParseException {
if (reader == null) {
return;
}
BufferedReader br = new BufferedReader(reader);
String line;
try {
while ((line = br.readLine()) != null) {
int commentMarker = line.indexOf('#');
if (commentMarker != -1) {
if (commentMarker == 0) {
//skip its a comment
continue;
} else {
//it's a malformed comment
throw new ParseException(line, commentMarker);
}
} else {
if (line.isEmpty() || line.matches("^\\s*$")) {
//skip it's a black line
continue;
}
//split till the first space
int delimiterIdx = line.indexOf(':');
String username = line.substring(0, delimiterIdx).trim();
String password = line.substring(delimiterIdx + 1).trim();
m_identities.put(username, password);
}
}
} catch (IOException ex) {
throw new ParseException("Failed to read", 1);
}
}
public boolean checkValid(String username, byte[] password) {
if (username == null || password == null) {
LOG.info("username or password was null");
return false;
}
String foundPwq = m_identities.get(username);
if (foundPwq == null) {
return false;
}
m_digest.update(password);
byte[] digest = m_digest.digest();
String encodedPasswd = new String(Hex.encodeHex(digest));
return foundPwq.equals(encodedPasswd);
}
}
@@ -0,0 +1,26 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.security;
/**
* username and password checker
*
* @author andrea
*/
public interface IAuthenticator {
boolean checkValid(String username, byte[] password);
}
@@ -0,0 +1,35 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.security;
/**
* ACL checker.
*
* Create an authorizator that matches topic names with same grammar of subscriptions.
* The # is always a terminator and its the multilevel matcher.
* The + sign is the single level matcher.
*
* @author andrea
*/
public interface IAuthorizator {
/**
* Ask the implementation of the authorizator if the topic can be used in a publish.
* */
boolean canWrite(String topic, String user, String client);
boolean canRead(String topic, String user, String client);
}
@@ -0,0 +1,31 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.security;
/**
* @author andrea
*/
public class PermitAllAuthorizator implements IAuthorizator {
@Override
public boolean canWrite(String topic, String user, String client) {
return true;
}
@Override
public boolean canRead(String topic, String user, String client) {
return true;
}
}
@@ -0,0 +1,104 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.subscriptions;
import java.io.Serializable;
import io.moquette.proto.messages.AbstractMessage.QOSType;
/**
* Maintain the information about which Topic a certain ClientID is subscribed
* and at which QoS
*
*
* @author andrea
*/
public class Subscription implements Serializable {
QOSType requestedQos; //max QoS acceptable
String clientId;
String topicFilter;
boolean cleanSession;
boolean active = true;
public Subscription(String clientId, String topicFilter, QOSType requestedQos, boolean cleanSession) {
this.requestedQos = requestedQos;
this.clientId = clientId;
this.topicFilter = topicFilter;
this.cleanSession = cleanSession;
}
public Subscription(Subscription orig) {
this.requestedQos = orig.requestedQos;
this.clientId = orig.clientId;
this.topicFilter = orig.topicFilter;
this.cleanSession = orig.cleanSession;
this.active = orig.active;
}
public String getClientId() {
return clientId;
}
public QOSType getRequestedQos() {
return requestedQos;
}
public String getTopicFilter() {
return topicFilter;
}
public boolean isCleanSession() {
return this.cleanSession;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Subscription that = (Subscription) o;
if (cleanSession != that.cleanSession) return false;
if (clientId != null ? !clientId.equals(that.clientId) : that.clientId != null) return false;
if (requestedQos != that.requestedQos) return false;
if (topicFilter != null ? !topicFilter.equals(that.topicFilter) : that.topicFilter != null) return false;
return true;
}
@Override
public int hashCode() {
int result = requestedQos.hashCode();
result = 31 * result + (clientId != null ? clientId.hashCode() : 0);
result = 31 * result + (topicFilter != null ? topicFilter.hashCode() : 0);
result = 31 * result + (cleanSession ? 1 : 0);
return result;
}
@Override
public String toString() {
return String.format("[filter:%s, cliID: %s, qos: %s, active: %s]", this.topicFilter, this.clientId, this.requestedQos, this.active);
}
@Override
public Subscription clone() {
try {
return (Subscription) super.clone();
} catch (CloneNotSupportedException e) {
return null;
}
}
}
@@ -0,0 +1,339 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.subscriptions;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicReference;
import io.moquette.spi.ISessionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Represents a tree of topics subscriptions.
*
* @author andrea
*/
public class SubscriptionsStore {
public static class NodeCouple {
final TreeNode root;
final TreeNode createdNode;
public NodeCouple(TreeNode root, TreeNode createdNode) {
this.root = root;
this.createdNode = createdNode;
}
}
/**
* Check if the topic filter of the subscription is well formed
* */
public static boolean validate(String topicFilter) {
try {
parseTopic(topicFilter);
return true;
} catch (ParseException pex) {
LOG.info("Bad matching topic filter <{}>", topicFilter);
return false;
}
}
public interface IVisitor<T> {
void visit(TreeNode node, int deep);
T getResult();
}
private class DumpTreeVisitor implements IVisitor<String> {
String s = "";
@Override
public void visit(TreeNode node, int deep) {
String subScriptionsStr = "";
String indentTabs = indentTabs(deep);
for (Subscription sub : node.m_subscriptions) {
subScriptionsStr += indentTabs + sub.toString() + "\n";
}
s += node.getToken() == null ? "" : node.getToken().toString();
s += "\n" + (node.m_subscriptions.isEmpty() ? indentTabs : "") + subScriptionsStr /*+ "\n"*/;
}
private String indentTabs(int deep) {
String s = "";
for (int i=0; i < deep; i++) {
s += "\t";
// s += "--";
}
return s;
}
@Override
public String getResult() {
return s;
}
}
private AtomicReference<TreeNode> subscriptions = new AtomicReference<>(new TreeNode(null));
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
/**
* Initialize the subscription tree with the list of subscriptions.
* Maintained for compatibility reasons.
*/
public void init(ISessionsStore sessionsStore) {
LOG.debug("init invoked");
List<Subscription> subscriptions = sessionsStore.listAllSubscriptions();
//reload any subscriptions persisted
if (LOG.isDebugEnabled()) {
LOG.debug("Reloading all stored subscriptions...subscription tree before {}", dumpTree());
}
for (Subscription subscription : subscriptions) {
LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopicFilter());
add(subscription);
}
if (LOG.isDebugEnabled()) {
LOG.debug("Finished loading. Subscription tree after {}", dumpTree());
}
}
public void add(Subscription newSubscription) {
TreeNode oldRoot;
NodeCouple couple;
do {
oldRoot = subscriptions.get();
couple = recreatePath(newSubscription.topicFilter, oldRoot);
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);
}
protected NodeCouple recreatePath(String topic, final TreeNode oldRoot) {
List<Token> tokens = new ArrayList<>();
try {
tokens = parseTopic(topic);
} catch (ParseException ex) {
//TODO handle the parse exception
LOG.error(null, ex);
}
final TreeNode newRoot = oldRoot.copy();
TreeNode parent = newRoot;
TreeNode current = newRoot;
for (Token token : tokens) {
TreeNode matchingChildren;
//check if a children with the same token already exists
if ((matchingChildren = current.childWithToken(token)) != null) {
//copy the traversed node
current = matchingChildren.copy();
current.m_parent = parent;
//update the child just added in the children list
parent.updateChild(matchingChildren, current);
parent = current;
} else {
//create a new node for the newly inserted token
matchingChildren = new TreeNode(current);
matchingChildren.setToken(token);
current.addChild(matchingChildren);
current = matchingChildren;
}
}
return new NodeCouple(newRoot, current);
}
public void removeSubscription(String topic, String clientID) {
TreeNode oldRoot;
NodeCouple couple;
do {
oldRoot = subscriptions.get();
couple = recreatePath(topic, oldRoot);
//do the job
//search for the subscription to remove
Subscription toBeRemoved = null;
for (Subscription sub : couple.createdNode.subscriptions()) {
if (sub.topicFilter.equals(topic) && sub.getClientId().equals(clientID)) {
toBeRemoved = sub;
break;
}
}
if (toBeRemoved != null) {
couple.createdNode.subscriptions().remove(toBeRemoved);
}
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, couple.root));
}
/**
* Visit the topics tree to remove matching subscriptions with clientID.
* It's a mutating structure operation so create a new subscription tree (partial or total).
*/
public void removeForClient(String clientID) {
TreeNode oldRoot;
TreeNode newRoot;
do {
oldRoot = subscriptions.get();
newRoot = oldRoot.removeClientSubscriptions(clientID);
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, newRoot));
}
/**
* Given a topic string return the clients subscriptions that matches it.
* Topic string can't contain character # and + because they are reserved to
* listeners subscriptions, and not topic publishing.
*/
public List<Subscription> matches(String topic) {
List<Token> tokens;
try {
tokens = parseTopic(topic);
} catch (ParseException ex) {
//TODO handle the parse exception
LOG.error(null, ex);
return Collections.emptyList();
}
Queue<Token> tokenQueue = new LinkedBlockingDeque<>(tokens);
List<Subscription> matchingSubs = new ArrayList<>();
subscriptions.get().matches(tokenQueue, matchingSubs);
//remove the overlapping subscriptions, selecting ones with greatest qos
Map<String, Subscription> subsForClient = new HashMap<>();
for (Subscription sub : matchingSubs) {
Subscription existingSub = subsForClient.get(sub.getClientId());
//update the selected subscriptions if not present or if has a greater qos
if (existingSub == null || existingSub.getRequestedQos().byteValue() < sub.getRequestedQos().byteValue()) {
subsForClient.put(sub.getClientId(), sub);
}
}
return new ArrayList<>(subsForClient.values());
}
public boolean contains(Subscription sub) {
return !matches(sub.topicFilter).isEmpty();
}
public int size() {
return subscriptions.get().size();
}
public String dumpTree() {
DumpTreeVisitor visitor = new DumpTreeVisitor();
bfsVisit(subscriptions.get(), visitor, 0);
return visitor.getResult();
}
private void bfsVisit(TreeNode node, IVisitor visitor, int deep) {
if (node == null) {
return;
}
visitor.visit(node, deep);
for (TreeNode child : node.m_children) {
bfsVisit(child, visitor, ++deep);
}
}
/**
* Verify if the 2 topics matching respecting the rules of MQTT Appendix A
*/
//TODO reimplement with iterators or with queues
public static boolean matchTopics(String msgTopic, String subscriptionTopic) {
try {
List<Token> msgTokens = SubscriptionsStore.parseTopic(msgTopic);
List<Token> subscriptionTokens = SubscriptionsStore.parseTopic(subscriptionTopic);
int i = 0;
for (; i< subscriptionTokens.size(); i++) {
Token subToken = subscriptionTokens.get(i);
if (subToken != Token.MULTI && subToken != Token.SINGLE) {
if (i >= msgTokens.size()) {
return false;
}
Token msgToken = msgTokens.get(i);
if (!msgToken.equals(subToken)) {
return false;
}
} else {
if (subToken == Token.MULTI) {
return true;
}
if (subToken == Token.SINGLE) {
//skip a step forward
}
}
}
//if last token was a SINGLE then treat it as an empty
// if (subToken == Token.SINGLE && (i - msgTokens.size() == 1)) {
// i--;
// }
return i == msgTokens.size();
} catch (ParseException ex) {
LOG.error(null, ex);
throw new RuntimeException(ex);
}
}
protected static List<Token> parseTopic(String topic) throws ParseException {
List<Token> res = new ArrayList<>();
String[] splitted = topic.split("/");
if (splitted.length == 0) {
res.add(Token.EMPTY);
}
if (topic.endsWith("/")) {
//Add a fictious space
String[] newSplitted = new String[splitted.length + 1];
System.arraycopy(splitted, 0, newSplitted, 0, splitted.length);
newSplitted[splitted.length] = "";
splitted = newSplitted;
}
for (int i = 0; i < splitted.length; i++) {
String s = splitted[i];
if (s.isEmpty()) {
// if (i != 0) {
// throw new ParseException("Bad format of topic, expetec topic name between separators", i);
// }
res.add(Token.EMPTY);
} else if (s.equals("#")) {
//check that multi is the last symbol
if (i != splitted.length - 1) {
throw new ParseException("Bad format of topic, the multi symbol (#) has to be the last one after a separator", i);
}
res.add(Token.MULTI);
} else if (s.contains("#")) {
throw new ParseException("Bad format of topic, invalid subtopic name: " + s, i);
} else if (s.equals("+")) {
res.add(Token.SINGLE);
} else if (s.contains("+")) {
throw new ParseException("Bad format of topic, invalid subtopic name: " + s, i);
} else {
res.add(new Token(s));
}
}
return res;
}
}
@@ -0,0 +1,74 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.subscriptions;
/**
* Internal use only class.
* */
public class Token {
static final Token EMPTY = new Token("");
static final Token MULTI = new Token("#");
static final Token SINGLE = new Token("+");
final String name;
protected Token(String s) {
name = s;
}
protected String name() {
return name;
}
protected boolean match(Token t) {
if (t == MULTI || t == SINGLE) {
return false;
}
if (this == MULTI || this == SINGLE) {
return true;
}
return equals(t);
}
@Override
public int hashCode() {
int hash = 7;
hash = 29 * hash + (this.name != null ? this.name.hashCode() : 0);
return hash;
}
@Override
public boolean equals(Object obj) {
if (obj == null) {
return false;
}
if (getClass() != obj.getClass()) {
return false;
}
final Token other = (Token) obj;
if ((this.name == null) ? (other.name != null) : !this.name.equals(other.name)) {
return false;
}
return true;
}
@Override
public String toString() {
return name;
}
}
@@ -0,0 +1,192 @@
/*
* Copyright (c) 2012-2015 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.spi.impl.subscriptions;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
class TreeNode {
private class ClientIDComparator implements Comparator<Subscription> {
public int compare(Subscription o1, Subscription o2) {
return o1.getClientId().compareTo(o2.getClientId());
}
}
TreeNode m_parent;
Token m_token;
List<TreeNode> m_children = new ArrayList<>();
List<Subscription> m_subscriptions = new ArrayList<>();
TreeNode(TreeNode parent) {
this.m_parent = parent;
}
Token getToken() {
return m_token;
}
void setToken(Token topic) {
this.m_token = topic;
}
void addSubscription(Subscription s) {
//avoid double registering for same clientID, topic and QoS
if (m_subscriptions.contains(s)) {
return;
}
//remove existing subscription for same client and topic but different QoS
Comparator<Subscription> comparator = new ClientIDComparator();
Collections.sort(m_subscriptions, comparator);
int existingSubIdx = Collections.binarySearch(m_subscriptions, s, comparator);
if (existingSubIdx >= 0) {
m_subscriptions.remove(existingSubIdx);
}
m_subscriptions.add(s);
}
void addChild(TreeNode child) {
m_children.add(child);
}
/**
* Creates a shallow copy of the current node.
* Copy the token and the children.
* */
TreeNode copy() {
final TreeNode copy = new TreeNode(this);
copy.m_parent = m_parent;
copy.m_children = new ArrayList<>(m_children);
copy.m_subscriptions = new ArrayList<>(m_subscriptions.size());
for (Subscription sub : m_subscriptions) {
copy.m_subscriptions.add(new Subscription(sub));
}
copy.m_token = m_token;
return copy;
}
/**
* Search for children that has the specified token, if not found return
* null;
*/
TreeNode childWithToken(Token token) {
for (TreeNode child : m_children) {
if (child.getToken().equals(token)) {
return child;
}
}
return null;
}
void updateChild(TreeNode oldChild, TreeNode newChild) {
m_children.remove(oldChild);
m_children.add(newChild);
}
List<Subscription> subscriptions() {
return m_subscriptions;
}
void matches(Queue<Token> tokens, List<Subscription> matchingSubs) {
Token t = tokens.poll();
//check if t is null <=> tokens finished
if (t == null) {
matchingSubs.addAll(m_subscriptions);
//check if it has got a MULTI child and add its subscriptions
for (TreeNode n : m_children) {
if (n.getToken() == Token.MULTI || n.getToken() == Token.SINGLE) {
matchingSubs.addAll(n.subscriptions());
}
}
return;
}
//we are on MULTI, than add subscriptions and return
if (m_token == Token.MULTI) {
matchingSubs.addAll(m_subscriptions);
return;
}
for (TreeNode n : m_children) {
if (n.getToken().match(t)) {
//Create a copy of token, else if navigate 2 sibling it
//consumes 2 elements on the queue instead of one
n.matches(new LinkedBlockingQueue<>(tokens), matchingSubs);
//TODO don't create a copy n.matches(tokens, matchingSubs);
}
}
}
/**
* Return the number of registered subscriptions
*/
int size() {
int res = m_subscriptions.size();
for (TreeNode child : m_children) {
res += child.size();
}
return res;
}
/**
* Create a copied subtree rooted on this node but purged of clientID's subscriptions.
* */
TreeNode removeClientSubscriptions(String clientID) {
//collect what to delete and then delete to avoid ConcurrentModification
TreeNode newSubRoot = this.copy();
List<Subscription> subsToRemove = new ArrayList<>();
for (Subscription s : newSubRoot.m_subscriptions) {
if (s.clientId.equals(clientID)) {
subsToRemove.add(s);
}
}
for (Subscription s : subsToRemove) {
newSubRoot.m_subscriptions.remove(s);
}
//go deep
List<TreeNode> newChildren = new ArrayList<>(newSubRoot.m_children.size());
for (TreeNode child : newSubRoot.m_children) {
newChildren.add(child.removeClientSubscriptions(clientID));
}
newSubRoot.m_children = newChildren;
return newSubRoot;
}
/**
* @return the set of subscriptions for the given client.
* */
Set<Subscription> findAllByClientID(String clientID) {
Set<Subscription> subs = new HashSet<>();
for (Subscription s : m_subscriptions) {
if (s.clientId.equals(clientID)) {
subs.add(s);
}
}
//go deep
for (TreeNode child : m_children) {
subs.addAll(child.findAllByClientID(clientID));
}
return subs;
}
}
@@ -0,0 +1,136 @@
/*
* Copyright (c) 2012-2015 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.spi.persistence;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import org.mapdb.DB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.ConcurrentMap;
/**
* IMessagesStore implementation backed by MapDB.
*
* @author andrea
*/
class MapDBMessagesStore implements IMessagesStore {
private static final Logger LOG = LoggerFactory.getLogger(MapDBMessagesStore.class);
private DB m_db;
//map clientID <-> set of currently in flight packet identifiers
private Map<String, Set<Integer>> m_inFlightIds;
//maps clientID -> guid
private ConcurrentMap<String, String> m_retainedStore;
//maps guid to message, it's message store
private ConcurrentMap<String, IMessagesStore.StoredMessage> m_persistentMessageStore;
MapDBMessagesStore(DB db) {
m_db = db;
}
@Override
public void initStore() {
m_retainedStore = m_db.getHashMap("retained");
m_persistentMessageStore = m_db.getHashMap("persistedMessages");
m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
}
@Override
public void storeRetained(String topic, String 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());
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, String> entry : m_retainedStore.entrySet()) {
final String guid = entry.getValue();
StoredMessage storedMsg = m_persistentMessageStore.get(guid);
if (condition.match(entry.getKey())) {
results.add(storedMsg);
}
}
return results;
}
@Override
public String storePublishForFuture(StoredMessage evt) {
LOG.debug("storePublishForFuture store evt {}", evt);
if (evt.getClientID() == null) {
LOG.error("persisting a message without a clientID, bad programming error msg: {}", evt);
throw new IllegalArgumentException("\"persisting a message without a clientID, bad programming error");
}
String guid = UUID.randomUUID().toString();
evt.setGuid(guid);
m_persistentMessageStore.put(guid, evt);
ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(evt.getClientID()));
messageIdToGuid.put(evt.getMessageID(), guid);
return guid;
}
@Override
public List<StoredMessage> listMessagesInSession(Collection<String> guids) {
List<StoredMessage> ret = new ArrayList<>();
for (String guid : guids) {
ret.add(m_persistentMessageStore.get(guid));
}
return ret;
}
@Override
public void dropMessagesInSession(String clientID) {
m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID)).clear();
m_persistentMessageStore.remove(clientID);
}
@Override
public StoredMessage getMessageByGuid(String guid) {
return m_persistentMessageStore.get(guid);
}
/**
* Return the next valid packetIdentifier for the given client session.
* */
@Override
public int nextPacketID(String clientID) {
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient == null) {
int nextPacketId = 1;
inFlightForClient = new HashSet<>();
inFlightForClient.add(nextPacketId);
this.m_inFlightIds.put(clientID, inFlightForClient);
return nextPacketId;
}
int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
int nextPacketId = (maxId + 1) % 0xFFFF;
inFlightForClient.add(nextPacketId);
return nextPacketId;
}
@Override
public void cleanRetained(String topic) {
m_retainedStore.remove(topic);
}
}
@@ -0,0 +1,115 @@
/*
* Copyright (c) 2012-2015 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.spi.persistence;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.proto.MQTTException;
import org.mapdb.DB;
import org.mapdb.DBMaker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* MapDB main persistence implementation
*/
public class MapDBPersistentStore {
/**
* This is a DTO used to persist minimal status (clean session and activation status) of
* a session.
* */
public static class PersistentSession implements Serializable {
public final boolean cleanSession;
public final boolean active;
public PersistentSession(boolean cleanSession, boolean active) {
this.cleanSession = cleanSession;
this.active = active;
}
}
private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class);
private DB m_db;
private String m_storePath;
protected final ScheduledExecutorService m_scheduler = Executors.newScheduledThreadPool(1);
public MapDBPersistentStore(String storePath) {
this.m_storePath = storePath;
}
/**
* Factory method to create message store backed by MapDB
* */
public IMessagesStore messagesStore() {
//TODO check m_db is valid and
IMessagesStore msgStore = new MapDBMessagesStore(m_db);
msgStore.initStore();
return msgStore;
}
public ISessionsStore sessionsStore(IMessagesStore msgStore) {
ISessionsStore sessionsStore = new MapDBSessionsStore(m_db, msgStore);
sessionsStore.initStore();
return sessionsStore;
}
public void initStore() {
if (m_storePath == null || m_storePath.isEmpty()) {
m_db = DBMaker.newMemoryDB().make();
} else {
File tmpFile;
try {
tmpFile = new File(m_storePath);
boolean fileNewlyCreated = tmpFile.createNewFile();
LOG.info("Starting with {} [{}] db file", fileNewlyCreated ? "fresh" : "existing", m_storePath);
} catch (IOException ex) {
LOG.error(null, ex);
throw new MQTTException("Can't create temp file for subscriptions storage [" + m_storePath + "]", ex);
}
m_db = DBMaker.newFileDB(tmpFile).make();
}
m_scheduler.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
m_db.commit();
}
}, 30, 30, TimeUnit.SECONDS);
}
public void close() {
if (this.m_db.isClosed()) {
LOG.debug("already closed");
return;
}
this.m_db.commit();
//LOG.debug("persisted subscriptions {}", m_persistentSubscriptions);
this.m_db.close();
LOG.debug("closed disk storage");
this.m_scheduler.shutdown();
LOG.debug("Persistence commit scheduler is shutdown");
}
}
@@ -0,0 +1,263 @@
/*
* Copyright (c) 2012-2015 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.spi.persistence;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.Utils;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
import org.mapdb.DB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.ConcurrentMap;
/**
* ISessionsStore implementation backed by MapDB.
*
* @author andrea
*/
class MapDBSessionsStore implements ISessionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class);
//maps clientID->[MessageId -> guid]
private ConcurrentMap<String, Map<Integer, String>> m_inflightStore;
//map clientID <-> set of currently in flight packet identifiers
private Map<String, Set<Integer>> m_inFlightIds;
//persistent Map of clientID, set of Subscriptions
private ConcurrentMap<String, Set<Subscription>> m_persistentSubscriptions;
private ConcurrentMap<String, PersistentSession> m_persistentSessions;
//maps clientID->[guid*], insertion order cares, it's queue
private ConcurrentMap<String, List<String>> m_enqueuedStore;
//maps clientID->[messageID*]
private ConcurrentMap<String, Set<Integer>> m_secondPhaseStore;
private final DB m_db;
private final IMessagesStore m_messagesStore;
MapDBSessionsStore(DB db, IMessagesStore messagesStore) {
m_db = db;
m_messagesStore = messagesStore;
}
@Override
public void initStore() {
m_inflightStore = m_db.getHashMap("inflight");
m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
m_persistentSubscriptions = m_db.getHashMap("subscriptions");
m_persistentSessions = m_db.getHashMap("sessions");
m_enqueuedStore = m_db.getHashMap("sessionQueue");
m_secondPhaseStore = m_db.getHashMap("secondPhase");
}
@Override
public void addNewSubscription(Subscription newSubscription) {
LOG.debug("addNewSubscription invoked with subscription {}", newSubscription);
final String clientID = newSubscription.getClientId();
if (!m_persistentSubscriptions.containsKey(clientID)) {
LOG.debug("subscriptions for clientID <{}> not present, empty subscriptions set", clientID);
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
}
Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
if (!subs.contains(newSubscription)) {
LOG.debug("updating <{}> subscriptions set with new subscription", clientID);
//TODO check the subs doesn't contain another subscription to the same topic with different
Subscription existingSubscription = null;
for (Subscription scanSub : subs) {
if (newSubscription.getTopicFilter().equals(scanSub.getTopicFilter())) {
existingSubscription = scanSub;
break;
}
}
if (existingSubscription != null) {
subs.remove(existingSubscription);
}
subs.add(newSubscription);
m_persistentSubscriptions.put(clientID, subs);
LOG.debug("clientID {} subscriptions set now is {}", clientID, subs);
}
}
@Override
public void removeSubscription(String topic, String clientID) {
LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
if (!m_persistentSubscriptions.containsKey(clientID)) {
return;
}
Set<Subscription> clientSubscriptions = m_persistentSubscriptions.get(clientID);
//search for the subscription to remove
Subscription toBeRemoved = null;
for (Subscription sub : clientSubscriptions) {
if (sub.getTopicFilter().equals(topic)) {
toBeRemoved = sub;
break;
}
}
if (toBeRemoved != null) {
clientSubscriptions.remove(toBeRemoved);
}
m_persistentSubscriptions.put(clientID, clientSubscriptions);
if (LOG.isTraceEnabled()) {
LOG.trace("persisted subscriptions {}", m_persistentSubscriptions);
}
}
@Override
public void wipeSubscriptions(String clientID) {
m_persistentSubscriptions.remove(clientID);
}
@Override
public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
m_persistentSubscriptions.put(clientID, subscriptions);
}
@Override
public List<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<>();
for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
allSubscriptions.addAll(entry.getValue());
}
LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions);
return allSubscriptions;
}
@Override
public boolean contains(String clientID) {
return m_persistentSubscriptions.containsKey(clientID);
}
@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);
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);
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession, false));
return new ClientSession(clientID, m_messagesStore, this, cleanSession);
}
@Override
public ClientSession sessionForClient(String clientID) {
if (!m_persistentSessions.containsKey(clientID)) {
return null;
}
PersistentSession storedSession = m_persistentSessions.get(clientID);
ClientSession clientSession = new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
if (storedSession.active) {
clientSession.activate();
}
return clientSession;
}
@Override
public void activate(String clientID) {
activationHelper(clientID, true);
}
@Override
public void deactivate(String clientID) {
activationHelper(clientID, false);
}
private void activationHelper(String clientID, boolean activation) {
PersistentSession storedSession = m_persistentSessions.get(clientID);
if (storedSession == null) {
throw new IllegalStateException((activation ? "activating" : "deactivating") + " a session never stored/created, clientID <"+ clientID + ">", null);
}
PersistentSession newStoredSession = new PersistentSession(storedSession.cleanSession, activation);
m_persistentSessions.put(clientID, newStoredSession);
}
@Override
public void inFlightAck(String clientID, int messageID) {
Map<Integer, String> m = this.m_inflightStore.get(clientID);
if (m == null) {
LOG.error("Can't find the inFlight record for client <{}>", clientID);
return;
}
m.remove(messageID);
//remove from the ids store
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient != null) {
inFlightForClient.remove(messageID);
}
}
@Override
public void inFlight(String clientID, int messageID, String guid) {
Map<Integer, String> m = this.m_inflightStore.get(clientID);
if (m == null) {
m = new HashMap<>();
}
m.put(messageID, guid);
this.m_inflightStore.put(clientID, m);
}
@Override
public void bindToDeliver(String guid, String clientID) {
List<String> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
guids.add(guid);
m_enqueuedStore.put(clientID, guids);
}
@Override
public Collection<String> enqueued(String clientID) {
return Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
}
@Override
public void removeEnqueued(String clientID, String guid) {
List<String> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
guids.remove(guid);
m_enqueuedStore.put(clientID, guids);
}
@Override
public void secondPhaseAcknowledged(String clientID, int messageID) {
Set<Integer> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
messageIDs.remove(messageID);
m_secondPhaseStore.put(clientID, messageIDs);
}
@Override
public void secondPhaseAckWaiting(String clientID, int messageID) {
Set<Integer> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
messageIDs.add(messageID);
m_secondPhaseStore.put(clientID, messageIDs);
}
@Override
public String mapToGuid(String clientID, int messageID) {
ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
return messageIdToGuid.get(messageID);
}
static String messageId2GuidsMapName(String clientID) {
return "guidsMapping_" + clientID;
}
}