Changed package from org.dna.mqtt to orga.eclipse.moquette

This commit is contained in:
andrea
2014-10-05 13:22:30 +02:00
parent a599c38537
commit 7bcd437664
148 changed files with 399 additions and 400 deletions
@@ -0,0 +1,119 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.Reader;
import java.text.ParseException;
import java.util.Properties;
import org.eclipse.moquette.commons.Constants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* 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() {
m_properties.put("port", Integer.toString(Constants.PORT));
m_properties.put("host", Constants.HOST);
m_properties.put("websocket_port", Integer.toString(Constants.WEBSOCKET_PORT));
m_properties.put("password_file", "");
}
/**
* 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 deilimiterIdx = line.indexOf(' ');
String key = line.substring(0, deilimiterIdx).trim();
String value = line.substring(deilimiterIdx).trim();
m_properties.put(key, value);
}
}
} catch (IOException ex) {
throw new ParseException("Failed to read", 1);
}
}
Properties getProperties() {
return m_properties;
}
}
@@ -0,0 +1,53 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server;
/**
* Maintains the information of single connection, like ClientID, IoSession,
* and other connection related flags.
*
*
* @author andrea
*/
public class ConnectionDescriptor {
private String m_clientID;
private ServerChannel m_session;
private boolean m_cleanSession;
public ConnectionDescriptor(String clientID, ServerChannel session, boolean cleanSession) {
this.m_clientID = clientID;
this.m_session = session;
this.m_cleanSession = cleanSession;
}
public boolean isCleanSession() {
return m_cleanSession;
}
public String getClientID() {
return m_clientID;
}
public ServerChannel getSession() {
return m_session;
}
@Override
public String toString() {
return "ConnectionDescriptor{" + "m_clientID=" + m_clientID + ", m_cleanSession=" + m_cleanSession + '}';
}
}
@@ -0,0 +1,25 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.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,26 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server;
/**
* username and password checker
*
* @author andrea
*/
public interface IAuthenticator {
boolean checkValid(String username, String password);
}
@@ -0,0 +1,99 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server;
import java.io.File;
import java.io.IOException;
import java.text.ParseException;
import java.util.Properties;
import org.eclipse.moquette.spi.impl.SimpleMessaging;
import org.eclipse.moquette.server.netty.NettyAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Launch a configured version of the server.
* @author andrea
*/
public class Server {
private static final Logger LOG = LoggerFactory.getLogger(Server.class);
public static final String STORAGE_FILE_PATH = System.getProperty("user.home") +
File.separator + "moquette_store.mapdb";
private ServerAcceptor m_acceptor;
SimpleMessaging messaging;
public static void main(String[] args) throws IOException {
final Server server = new Server();
server.startServer();
System.out.println("Server started, version 0.7-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 config/moquette.conf
*/
public void startServer() throws IOException {
String configPath = System.getProperty("moquette.path", null);
startServer(new File(configPath, "config/moquette.conf"));
}
/**
* Starts Moquette bringing the configuration from the given file
*/
public void startServer(File configFile) throws IOException {
ConfigurationParser confParser = new ConfigurationParser();
try {
confParser.parse(configFile);
} catch (ParseException pex) {
LOG.warn("An error occurred in parsing configuration, fallback on default configuration", pex);
}
Properties configProps = confParser.getProperties();
startServer(configProps);
}
/**
* Starts the server with the given properties.
*
* Its need at least the following properties:
* <ul>
* <li>port</li>
* <li>password_file</li>
* </ul>
*/
public void startServer(Properties configProps) throws IOException {
messaging = SimpleMessaging.getInstance();
messaging.init(configProps);
m_acceptor = new NettyAcceptor();
m_acceptor.initialize(messaging, configProps);
}
public void stopServer() {
System.out.println("Server stopping...");
messaging.stop();
m_acceptor.close();
System.out.println("Server stopped");
}
}
@@ -0,0 +1,31 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server;
import java.io.IOException;
import java.util.Properties;
import org.eclipse.moquette.spi.IMessaging;
/**
*
* @author andrea
*/
public interface ServerAcceptor {
void initialize(IMessaging messaging, Properties props) throws IOException;
void close();
}
@@ -0,0 +1,33 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server;
/**
*
* @author andrea
*/
public interface ServerChannel {
Object getAttribute(Object key);
void setAttribute(Object key, Object value);
void setIdleTime(int idleTime);
void close(boolean immediately);
void write(Object value);
}
@@ -0,0 +1,37 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server.netty;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
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 if (e.getState() == IdleState.WRITER_IDLE) {
ctx.writeAndFlush(new PingMessage());
}*/
}
}
}
@@ -0,0 +1,256 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server.netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
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 java.io.IOException;
import java.io.InputStream;
import java.security.KeyStore;
import java.util.List;
import java.util.Properties;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import org.eclipse.moquette.commons.Constants;
import org.eclipse.moquette.spi.IMessaging;
import org.eclipse.moquette.parser.netty.MQTTDecoder;
import org.eclipse.moquette.parser.netty.MQTTEncoder;
import org.eclipse.moquette.server.ServerAcceptor;
import org.eclipse.moquette.server.netty.metrics.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @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();
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(bb);
out.add(bb);
}
}
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_metricsCollector = new BytesMetricsCollector();
MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector();
@Override
public void initialize(IMessaging messaging, Properties props) throws IOException {
m_bossGroup = new NioEventLoopGroup();
m_workerGroup = new NioEventLoopGroup();
initializePlainTCPTransport(messaging, props);
initializeWebSocketTransport(messaging, props);
initializeSSLTCPTransport(messaging, props);
}
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();
pipeliner.init(pipeline);
}
})
.option(ChannelOption.SO_BACKLOG, 128)
.option(ChannelOption.SO_REUSEADDR, true)
.childOption(ChannelOption.SO_KEEPALIVE, true);
try {
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(host, port);
LOG.info("Server binded");
f.sync();
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
private void initializePlainTCPTransport(IMessaging messaging, Properties props) throws IOException {
final NettyMQTTHandler handler = new NettyMQTTHandler();
handler.setMessaging(messaging);
String host = props.getProperty("host");
int port = Integer.parseInt(props.getProperty("port"));
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", new MoquetteIdleTimoutHandler());
//pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("handler", handler);
}
});
}
private void initializeWebSocketTransport(IMessaging messaging, Properties props) throws IOException {
String webSocketPortProp = props.getProperty("websocket_port");
if (webSocketPortProp == null) {
//Do nothing no WebSocket configured
LOG.info("WebSocket is disabled");
return;
}
int port = Integer.parseInt(webSocketPortProp);
final NettyMQTTHandler handler = new NettyMQTTHandler();
handler.setMessaging(messaging);
String host = props.getProperty("host");
initFactory(host, port, new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) {
pipeline.addLast("httpEncoder", new HttpResponseEncoder());
pipeline.addLast("httpDdecoder", new HttpRequestDecoder());
pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqttv3.1"));
pipeline.addLast("bytebuf2wsDecoderr", new ByteBufToWebSocketFrameEncoder());
pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
pipeline.addLast("decoder", new MQTTDecoder());
pipeline.addLast("encoder", new MQTTEncoder());
pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
pipeline.addLast("handler", handler);
}
});
}
private void initializeSSLTCPTransport(IMessaging messaging, Properties props) throws IOException {
String sslPortProp = props.getProperty("ssl_port");
if (sslPortProp == null) {
//Do nothing no SSL configured
LOG.info("SSL is disabled");
return;
}
final String jksPath = props.getProperty("jks_path");
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;
}
//if we have the port also the jks then keyStorePassword and keyManagerPassword
//has to be defined
final String keyStorePassword = props.getProperty("key_store_password");
final String keyManagerPassword = props.getProperty("key_manager_password");
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;
}
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;
}
int sslPort = Integer.parseInt(sslPortProp);
String host = props.getProperty("host");
final NettyMQTTHandler handler = new NettyMQTTHandler();
handler.setMessaging(messaging);
initFactory(host, sslPort, new PipelineInitializer() {
@Override
void init(ChannelPipeline pipeline) throws Exception {
InputStream jksInputStream = getClass().getClassLoader().getResourceAsStream(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);
SSLEngine engine = serverContext.createSSLEngine();
engine.setUseClientMode(false);
final SslHandler sslHandler = new SslHandler(engine);
pipeline.addLast("ssl", sslHandler);
//pipeline.addFirst("metrics", new BytesMetricsHandler(m_metricsCollector));
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
//pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
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");
}
m_workerGroup.shutdownGracefully();
m_bossGroup.shutdownGracefully();
MessageMetrics metrics = m_metricsCollector.computeMetrics();
//LOG.info(String.format("Bytes read: %d, bytes wrote: %d", metrics.readBytes(), metrics.wroteBytes()));
LOG.info("Msg read: {}, msg wrote: {}", metrics.messagesRead(), metrics.messagesWrote());
}
}
@@ -0,0 +1,84 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server.netty;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import java.util.HashMap;
import java.util.Map;
import org.eclipse.moquette.server.Constants;
import org.eclipse.moquette.server.ServerChannel;
/**
*
* @author andrea
*/
public class NettyChannel implements ServerChannel {
private ChannelHandlerContext m_channel;
private Map<Object, AttributeKey<Object>> m_attributesKeys = new HashMap<Object, AttributeKey<Object>>();
private static final AttributeKey<Object> ATTR_KEY_KEEPALIVE = new AttributeKey<Object>(Constants.KEEP_ALIVE);
private static final AttributeKey<Object> ATTR_KEY_CLEANSESSION = new AttributeKey<Object>(Constants.CLEAN_SESSION);
private static final AttributeKey<Object> ATTR_KEY_CLIENTID = new AttributeKey<Object>(Constants.ATTR_CLIENTID);
NettyChannel(ChannelHandlerContext ctx) {
m_channel = ctx;
m_attributesKeys.put(Constants.KEEP_ALIVE, ATTR_KEY_KEEPALIVE);
m_attributesKeys.put(Constants.CLEAN_SESSION, ATTR_KEY_CLEANSESSION);
m_attributesKeys.put(Constants.ATTR_CLIENTID, ATTR_KEY_CLIENTID);
}
public Object getAttribute(Object key) {
Attribute<Object> attr = m_channel.attr(mapKey(key));
return attr.get();
}
public void setAttribute(Object key, Object value) {
Attribute<Object> attr = m_channel.attr(mapKey(key));
attr.set(value);
}
private synchronized AttributeKey<Object> mapKey(Object key) {
if (!m_attributesKeys.containsKey(key)) {
throw new IllegalArgumentException("mapKey can't find a matching AttributeKey for " + key);
}
return m_attributesKeys.get(key);
}
public void setIdleTime(int idleTime) {
if (m_channel.pipeline().names().contains("idleStateHandler")) {
m_channel.pipeline().remove("idleStateHandler");
}
if (m_channel.pipeline().names().contains("idleEventHandler")) {
m_channel.pipeline().remove("idleEventHandler");
}
m_channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
m_channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
}
public void close(boolean immediately) {
m_channel.close();
}
public void write(Object value) {
m_channel.writeAndFlush(value);
}
}
@@ -0,0 +1,92 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server.netty;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.util.HashMap;
import java.util.Map;
import org.eclipse.moquette.spi.IMessaging;
import org.eclipse.moquette.proto.Utils;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import static org.eclipse.moquette.proto.messages.AbstractMessage.*;
import org.eclipse.moquette.proto.messages.PingRespMessage;
import org.eclipse.moquette.server.Constants;
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 IMessaging m_messaging;
private final Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<ChannelHandlerContext, NettyChannel>();
@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:
case SUBSCRIBE:
case UNSUBSCRIBE:
case PUBLISH:
case PUBREC:
case PUBCOMP:
case PUBREL:
case DISCONNECT:
case PUBACK:
NettyChannel channel;
synchronized(m_channelMapper) {
if (!m_channelMapper.containsKey(ctx)) {
m_channelMapper.put(ctx, new NettyChannel(ctx));
}
channel = m_channelMapper.get(ctx);
}
m_messaging.handleProtocolMessage(channel, 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 {
NettyChannel channel = m_channelMapper.get(ctx);
String clientID = (String) channel.getAttribute(Constants.ATTR_CLIENTID);
m_messaging.lostConnection(clientID);
ctx.close(/*false*/);
synchronized(m_channelMapper) {
m_channelMapper.remove(ctx);
}
}
public void setMessaging(IMessaging messaging) {
m_messaging = messaging;
}
}
@@ -0,0 +1,37 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.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,39 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server.netty.metrics;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
* Collects all the metrics from the various pipeline.
*/
public class BytesMetricsCollector {
private Queue<BytesMetrics> m_allMetrics = new ConcurrentLinkedQueue<BytesMetrics>();
void addMetrics(BytesMetrics metrics) {
m_allMetrics.add(metrics);
}
public BytesMetrics computeMetrics() {
BytesMetrics allMetrics = new BytesMetrics();
for (BytesMetrics m : m_allMetrics) {
allMetrics.incrementRead(m.readBytes());
allMetrics.incrementWrote(m.wroteBytes());
}
return allMetrics;
}
}
@@ -0,0 +1,65 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.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 = new AttributeKey<BytesMetrics>("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.addMetrics(metrics);
super.close(ctx, promise);
}
}
@@ -0,0 +1,37 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.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,39 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.server.netty.metrics;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
* Collects all the metrics from the various pipeline.
*/
public class MessageMetricsCollector {
private Queue<MessageMetrics> m_allMetrics = new ConcurrentLinkedQueue<MessageMetrics>();
void addMetrics(MessageMetrics metrics) {
m_allMetrics.add(metrics);
}
public MessageMetrics computeMetrics() {
MessageMetrics allMetrics = new MessageMetrics();
for (MessageMetrics m : m_allMetrics) {
allMetrics.incrementRead(m.messagesRead());
allMetrics.incrementWrote(m.messagesWrote());
}
return allMetrics;
}
}
@@ -0,0 +1,64 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.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 = new AttributeKey<MessageMetrics>("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.addMetrics(metrics);
super.close(ctx, promise);
}
}
@@ -0,0 +1,23 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi;
/**
*/
public interface IMatchingCondition {
boolean match(String key);
}
@@ -0,0 +1,97 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi;
import java.io.Serializable;
import java.nio.ByteBuffer;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.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 {
public static class StoredMessage implements Serializable {
final AbstractMessage.QOSType m_qos;
final byte[] m_payload;
final String m_topic;
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;
}
}
/**
* 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, ByteBuffer message, AbstractMessage.QOSType qos);
/**
* Return a list of retained messages that satisfy the condition.
*/
Collection<StoredMessage> searchMatching(IMatchingCondition condition);
void storePublishForFuture(PublishEvent 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<PublishEvent> retrievePersistedPublishes(String clientID);
void cleanPersistedPublishMessage(String clientID, int messageID);
void cleanPersistedPublishes(String clientID);
void cleanInFlight(String msgID);
void addInFlight(PublishEvent evt, String publishKey);
void close();
void persistQoS2Message(String publishKey, PublishEvent evt);
void removeQoS2Message(String publishKey);
PublishEvent retrieveQoS2Message(String publishKey);
void cleanRetained(String topic);
}
@@ -0,0 +1,36 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.server.ServerChannel;
/**
* 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 {
void stop();
void lostConnection(String clientID);
void handleProtocolMessage(ServerChannel session, AbstractMessage msg);
}
@@ -0,0 +1,51 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi;
import java.util.List;
import java.util.Set;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
/**
* Store used to handle the persistence of the subscriptions tree.
*
* @author andrea
*/
public interface ISessionsStore {
/**
* Add a new subscription to the session
*/
void addNewSubscription(Subscription newSubscription, 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);
}
@@ -0,0 +1,12 @@
package org.eclipse.moquette.spi.impl;
import org.eclipse.moquette.server.IAuthenticator;
/**
* Created by andrea on 8/23/14.
*/
class AcceptAllAuthenticator implements IAuthenticator {
public boolean checkValid(String username, String password) {
return true;
}
}
@@ -0,0 +1,69 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.server.ServerChannel;
/**
* Annotation @MQTTMessage helper class.
*
* It has the responsibility to identify the @MQTTMessage method of a given instance
* or class and define an easy access to invoke them.
*
* @author andrea
*/
class AnnotationSupport {
private final Map<Class, Method> messageClassToMethod = new HashMap<Class, Method>();
private Object targetInstance;
void processAnnotations(Object instance) {
targetInstance = instance;
Class instanceCls = instance.getClass();
Method[] allMethods = instanceCls.getDeclaredMethods();
for (Method m : allMethods) {
MQTTMessage annotation = m.getAnnotation(MQTTMessage.class);
if (annotation == null) {
continue;
}
Class mqttMessageClass = annotation.message();
messageClassToMethod.put(mqttMessageClass, m);
}
}
/**
* Dispatch a call to the wrapped target invoking the method that match the
* msg class.
*
* @throws RuntimeException if the message can't be routed to any annotated method,
* or if there are any errors during invocation.
*/
void dispatch(ServerChannel session, AbstractMessage msg) {
Method targetMethod = this.messageClassToMethod.get(msg.getClass());
if (targetMethod == null) {
throw new RuntimeException("Can't dispatch to any @MQTTMessage marked the message: " + msg);
}
try {
targetMethod.invoke(targetInstance, session, msg);
} catch (Exception ex) {
throw new RuntimeException(ex);
}
}
}
@@ -0,0 +1,32 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.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,105 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.Reader;
import java.text.ParseException;
import java.util.HashMap;
import java.util.Map;
import org.eclipse.moquette.server.IAuthenticator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class FileAuthenticator implements IAuthenticator {
private static final Logger LOG = LoggerFactory.getLogger(FileAuthenticator.class);
private Map<String, String> m_identities = new HashMap<String, String>();
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 {
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("Fromat ero 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 deilimiterIdx = line.indexOf(':');
String username = line.substring(0, deilimiterIdx).trim();
String password = line.substring(deilimiterIdx + 1).trim();
m_identities.put(username, password);
}
}
} catch (IOException ex) {
throw new ParseException("Failed to read", 1);
}
}
public boolean checkValid(String username, String password) {
String foundPwq = m_identities.get(username);
if (foundPwq == null) {
return false;
}
return foundPwq.equals(password);
}
}
@@ -0,0 +1,33 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl;
import java.lang.annotation.Documented;
import java.lang.annotation.Retention;
import static java.lang.annotation.ElementType.METHOD;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
import java.lang.annotation.Target;
/**
*
* @author andrea
*/
@Documented
@Target(METHOD)
@Retention(RUNTIME)
@interface MQTTMessage {
Class message();
}
@@ -0,0 +1,185 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl;
import java.nio.ByteBuffer;
import java.util.*;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.spi.ISessionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*/
public class MemoryStorageService implements IMessagesStore, ISessionsStore {
private Map<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<String, Set<Subscription>>();
private Map<String, StoredMessage> m_retainedStore = new HashMap<String, StoredMessage>();
//TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map)
private Map<String, List<PublishEvent>> m_persistentMessageStore = new HashMap<String, List<PublishEvent>>();
private Map<String, PublishEvent> m_inflightStore = new HashMap<String, PublishEvent>();
private Map<String, PublishEvent> m_qos2Store = new HashMap<String, PublishEvent>();
private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class);
public void initStore() {
}
@Override
public void cleanRetained(String topic) {
m_retainedStore.remove(topic);
}
@Override
public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) {
if (!message.hasRemaining()) {
//clean the message from topic
m_retainedStore.remove(topic);
} else {
//store the message to the topic
byte[] raw = new byte[message.remaining()];
message.get(raw);
m_retainedStore.put(topic, new StoredMessage(raw, qos, topic));
}
}
@Override
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<StoredMessage> results = new ArrayList<StoredMessage>();
for (Map.Entry<String, StoredMessage> entry : m_retainedStore.entrySet()) {
StoredMessage storedMsg = entry.getValue();
if (condition.match(entry.getKey())) {
results.add(storedMsg);
}
}
return results;
}
@Override
public void storePublishForFuture(PublishEvent evt) {
LOG.debug("storePublishForFuture store evt {}", evt);
List<PublishEvent> storedEvents;
String clientID = evt.getClientID();
if (!m_persistentMessageStore.containsKey(clientID)) {
storedEvents = new ArrayList<PublishEvent>();
} else {
storedEvents = m_persistentMessageStore.get(clientID);
}
storedEvents.add(evt);
m_persistentMessageStore.put(clientID, storedEvents);
}
@Override
public List<PublishEvent> retrievePersistedPublishes(String clientID) {
return m_persistentMessageStore.get(clientID);
}
@Override
public void cleanPersistedPublishMessage(String clientID, int messageID) {
List<PublishEvent> events = m_persistentMessageStore.get(clientID);
PublishEvent toRemoveEvt = null;
for (PublishEvent evt : events) {
if (evt.getMessageID() == messageID) {
toRemoveEvt = evt;
}
}
events.remove(toRemoveEvt);
m_persistentMessageStore.put(clientID, events);
}
@Override
public void cleanPersistedPublishes(String clientID) {
m_persistentMessageStore.remove(clientID);
}
@Override
public void cleanInFlight(String msgID) {
m_inflightStore.remove(msgID);
}
@Override
public void addInFlight(PublishEvent evt, String publishKey) {
m_inflightStore.put(publishKey, evt);
}
@Override
public void addNewSubscription(Subscription newSubscription, String clientID) {
if (!m_persistentSubscriptions.containsKey(clientID)) {
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
}
Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
if (!subs.contains(newSubscription)) {
subs.add(newSubscription);
m_persistentSubscriptions.put(clientID, subs);
}
}
@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 boolean contains(String clientID) {
return m_persistentSubscriptions.containsKey(clientID);
}
@Override
public List<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
allSubscriptions.addAll(entry.getValue());
}
return allSubscriptions;
}
@Override
public void close() {
//To change body of implemented methods use File | Settings | File Templates.
}
@Override
public void persistQoS2Message(String publishKey, PublishEvent evt) {
LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt);
m_qos2Store.put(publishKey, evt);
}
@Override
public void removeQoS2Message(String publishKey) {
m_qos2Store.remove(publishKey);
}
@Override
public PublishEvent retrieveQoS2Message(String publishKey) {
return m_qos2Store.get(publishKey);
}
}
@@ -0,0 +1,632 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.impl.events.MessagingEvent;
import org.eclipse.moquette.spi.impl.events.OutputMessagingEvent;
import org.eclipse.moquette.spi.impl.events.PubAckEvent;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
import static org.eclipse.moquette.parser.netty.Utils.VERSION_3_1;
import static org.eclipse.moquette.parser.netty.Utils.VERSION_3_1_1;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType;
import org.eclipse.moquette.proto.messages.ConnAckMessage;
import org.eclipse.moquette.proto.messages.ConnectMessage;
import org.eclipse.moquette.proto.messages.DisconnectMessage;
import org.eclipse.moquette.proto.messages.PubAckMessage;
import org.eclipse.moquette.proto.messages.PubCompMessage;
import org.eclipse.moquette.proto.messages.PubRecMessage;
import org.eclipse.moquette.proto.messages.PubRelMessage;
import org.eclipse.moquette.proto.messages.PublishMessage;
import org.eclipse.moquette.proto.messages.SubAckMessage;
import org.eclipse.moquette.proto.messages.SubscribeMessage;
import org.eclipse.moquette.proto.messages.UnsubAckMessage;
import org.eclipse.moquette.proto.messages.UnsubscribeMessage;
import org.eclipse.moquette.server.ConnectionDescriptor;
import org.eclipse.moquette.server.Constants;
import org.eclipse.moquette.server.IAuthenticator;
import org.eclipse.moquette.server.ServerChannel;
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
*/
class ProtocolProcessor implements EventHandler<ValueEvent> {
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 = new HashMap<String, ConnectionDescriptor>();
private SubscriptionsStore subscriptions;
private IMessagesStore m_messagesStore;
private ISessionsStore m_sessionsStore;
private IAuthenticator m_authenticator;
//maps clientID to Will testament, if specified on CONNECT
private Map<String, WillMessage> m_willStore = new HashMap<String, WillMessage>();
private ExecutorService m_executor;
private RingBuffer<ValueEvent> m_ringBuffer;
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
*/
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator) {
//m_clientIDs = clientIDs;
this.subscriptions = subscriptions;
LOG.debug("subscription tree on init {}", subscriptions.dumpTree());
m_authenticator = authenticator;
m_messagesStore = storageService;
m_sessionsStore = sessionsStore;
//init the output ringbuffer
m_executor = Executors.newFixedThreadPool(1);
Disruptor<ValueEvent> disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
disruptor.handleEventsWith(this);
disruptor.start();
// Get the ring buffer from the Disruptor to be used for publishing.
m_ringBuffer = disruptor.getRingBuffer();
}
@MQTTMessage(message = ConnectMessage.class)
void processConnect(ServerChannel session, ConnectMessage msg) {
LOG.debug("processConnect for client {}", msg.getClientID());
if (msg.getProcotolVersion() != VERSION_3_1 && msg.getProcotolVersion() != 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() > 23 || msg.getClientID().length() == 0) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
session.write(okResp);
return;
}
//if an old client with the same ID already exists close its session.
if (m_clientIDs.containsKey(msg.getClientID())) {
//clean the subscriptions if the old used a cleanSession = true
ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).getSession();
boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
cleanSession(msg.getClientID());
}
m_clientIDs.get(msg.getClientID()).getSession().close(false);
}
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(Constants.KEEP_ALIVE, keepAlive);
session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
//used to track the client in the subscription and publishing phases.
session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());
session.setIdleTime(Math.round(keepAlive * 1.5f));
//Handle will flag
if (msg.isWillFlag()) {
AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()];
byte[] willPayload = msg.getWillMessage().getBytes();
ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip();
//save the will testment in the clientID store
WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos );
m_willStore.put(msg.getClientID(), will);
}
//handle user authentication
if (msg.isUserFlag()) {
String pwd = null;
if (msg.isPasswordFlag()) {
pwd = msg.getPassword();
}
if (!m_authenticator.checkValid(msg.getUsername(), pwd)) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
session.write(okResp);
return;
}
}
subscriptions.activate(msg.getClientID());
//handle clean session flag
if (msg.isCleanSession()) {
//remove all prev subscriptions
//cleanup topic subscriptions
cleanSession(msg.getClientID());
}
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
if (!msg.isCleanSession() && m_sessionsStore.contains(msg.getClientID())) {
okResp.setSessionPresent(true);
}
LOG.debug("processConnect sent OK ConnAck");
session.write(okResp);
LOG.warn("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession());
LOG.info("Create persistent session for clientID {}", msg.getClientID());
m_sessionsStore.addNewSubscription(Subscription.createEmptySubscription(msg.getClientID(), true), msg.getClientID()); //null means EmptySubscription
if (!msg.isCleanSession()) {
//force the republish of stored QoS1 and QoS2
republishStored(msg.getClientID());
}
}
private void republishStored(String clientID) {
LOG.trace("republishStored invoked");
List<PublishEvent> publishedEvents = m_messagesStore.retrievePersistedPublishes(clientID);
if (publishedEvents == null) {
LOG.info("No stored messages for client <{}>", clientID);
return;
}
LOG.info("republishing stored messages to client <{}>", clientID);
for (PublishEvent pubEvt : publishedEvents) {
sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID());
}
}
@MQTTMessage(message = PubAckMessage.class)
void processPubAck(ServerChannel session, PubAckMessage msg) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = msg.getMessageID();
//Remove the message from message store
m_messagesStore.cleanPersistedPublishMessage(clientID, messageID);
}
private void cleanSession(String clientID) {
LOG.info("cleaning old saved subscriptions for client <{}>", clientID);
//remove from log all subscriptions
m_sessionsStore.wipeSubscriptions(clientID);
subscriptions.removeForClient(clientID);
//remove also the messages stored of type QoS1/2
m_messagesStore.cleanPersistedPublishes(clientID);
}
@MQTTMessage(message = PublishMessage.class)
void processPublish(ServerChannel session, PublishMessage msg) {
LOG.trace("PUB --PUBLISH--> SRV processPublish invoked with {}", msg);
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
final String topic = msg.getTopicName();
final AbstractMessage.QOSType qos = msg.getQos();
final ByteBuffer message = msg.getPayload();
boolean retain = msg.isRetainFlag();
processPublish(clientID, topic, qos, message, retain, msg.getMessageID());
}
private void processPublish(String clientID, String topic, QOSType qos, ByteBuffer message, boolean retain, Integer messageID) {
LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}",
clientID, topic, qos);
String publishKey = null;
if (qos == AbstractMessage.QOSType.LEAST_ONE) {
//store the temporary message
publishKey = String.format("%s%d", clientID, messageID);
PublishEvent inFlightEvt = new PublishEvent(topic, qos, message, retain,
clientID, messageID);
m_messagesStore.addInFlight(inFlightEvt, publishKey);
} else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) {
publishKey = String.format("%s%d", clientID, messageID);
//store the message in temp store
PublishEvent qos2Persistent = new PublishEvent(topic, qos, message, retain,
clientID, messageID);
m_messagesStore.persistQoS2Message(publishKey, qos2Persistent);
sendPubRec(clientID, messageID);
}
//NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
if (qos != AbstractMessage.QOSType.EXACTLY_ONCE) {
publish2Subscribers(topic, qos, message, retain, messageID);
}
if (qos == AbstractMessage.QOSType.LEAST_ONE) {
if (publishKey == null) {
throw new RuntimeException("Found a publish key null for QoS " + qos + " for message " + messageID);
}
m_messagesStore.cleanInFlight(publishKey);
sendPubAck(new PubAckEvent(messageID, clientID));
LOG.debug("replying with PubAck to MSG ID {}", messageID);
}
if (retain) {
if (qos == AbstractMessage.QOSType.MOST_ONE) {
//QoS == 0 && retain => clean old retained
m_messagesStore.cleanRetained(topic);
} else {
m_messagesStore.storeRetained(topic, message, qos);
}
}
}
/**
* Specialized version to publish will testament message.
*/
private void processPublish(WillMessage will, String clientID) {
final String topic = will.getTopic();
final AbstractMessage.QOSType qos = will.getQos();
final ByteBuffer message = will.getPayload();
boolean retain = will.isRetained();
processPublish(clientID, topic, qos, message, retain, null);
}
/**
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */
private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer origMessage, boolean retain, Integer messageID) {
LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic);
if (LOG.isDebugEnabled()) {
LOG.debug("content <{}>", DebugUtils.payload2Str(origMessage));
LOG.debug("subscription tree {}", subscriptions.dumpTree());
}
for (final Subscription sub : subscriptions.matches(topic)) {
if (qos.ordinal() > sub.getRequestedQos().ordinal()) {
qos = sub.getRequestedQos();
}
ByteBuffer message = origMessage.duplicate();
LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
sub.getClientId(), sub.getTopicFilter(), qos, sub.isActive());
if (qos == AbstractMessage.QOSType.MOST_ONE && sub.isActive()) {
//QoS 0
sendPublish(sub.getClientId(), topic, qos, message, false);
} else {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
if (!sub.isCleanSession() && !sub.isActive()) {
//clone the event with matching clientID
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0);
m_messagesStore.storePublishForFuture(newPublishEvt);
} else {
//if QoS 2 then store it in temp memory
if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) {
String publishKey = String.format("%s%d", sub.getClientId(), messageID);
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0);
m_messagesStore.addInFlight(newPublishEvt, publishKey);
}
//publish
if (sub.isActive()) {
sendPublish(sub.getClientId(), topic, qos, message, false);
}
}
}
}
}
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
//TODO pay attention to the message ID can't be 0 and it's the message sent to subscriber
int messageID = 1;
sendPublish(clientId, topic, qos, message, retained, messageID);
}
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, int messageID) {
LOG.debug("sendPublish 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));
}
if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) {
pubMessage.setMessageID(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) {
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs));
}
LOG.debug("Session for clientId {} is {}", clientId, m_clientIDs.get(clientId).getSession());
// m_clientIDs.get(clientId).getSession().write(pubMessage);
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), 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).getSession().write(pubRecMessage);
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRecMessage));
}
private void sendPubAck(PubAckEvent evt) {
LOG.trace("sendPubAck invoked");
String clientId = evt.getClientID();
PubAckMessage pubAckMessage = new PubAckMessage();
pubAckMessage.setMessageID(evt.getMessageId());
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));
}
// LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
// m_clientIDs.get(clientId).getSession().write(pubAckMessage);
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), 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.
* */
@MQTTMessage(message = PubRelMessage.class)
void processPubRel(ServerChannel session, PubRelMessage msg) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = msg.getMessageID();
LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
String publishKey = String.format("%s%d", clientID, messageID);
PublishEvent evt = m_messagesStore.retrieveQoS2Message(publishKey);
final String topic = evt.getTopic();
final AbstractMessage.QOSType qos = evt.getQos();
publish2Subscribers(topic, qos, evt.getMessage(), evt.isRetain(), evt.getMessageID());
m_messagesStore.removeQoS2Message(publishKey);
if (evt.isRetain()) {
m_messagesStore.storeRetained(topic, evt.getMessage(), qos);
}
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).getSession().write(pubCompMessage);
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubCompMessage));
}
@MQTTMessage(message = PubRecMessage.class)
void processPubRec(ServerChannel session, PubRecMessage msg) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = msg.getMessageID();
//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);
// m_clientIDs.get(clientID).getSession().write(pubRelMessage);
disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage));
}
@MQTTMessage(message = PubCompMessage.class)
void processPubComp(ServerChannel session, PubCompMessage msg) {
String clientID = (String) session.getAttribute(Constants.ATTR_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
String publishKey = String.format("%s%d", clientID, messageID);
m_messagesStore.cleanInFlight(publishKey);
}
@MQTTMessage(message = DisconnectMessage.class)
void processDisconnect(ServerChannel session, DisconnectMessage msg) throws InterruptedException {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
cleanSession(clientID);
}
// m_notifier.disconnect(evt.getSession());
m_clientIDs.remove(clientID);
session.close(true);
//de-activate the subscriptions for this ClientID
subscriptions.deactivate(clientID);
//cleanup the will store
m_willStore.remove(clientID);
LOG.warn("Disconnected client <{}> with clean session {}", clientID, cleanSession);
}
void processConnectionLost(String clientID) {
//If already removed a disconnect message was already processed for this clientID
if (m_clientIDs.remove(clientID) != null) {
//de-activate the subscriptions for this ClientID
subscriptions.deactivate(clientID);
LOG.info("Lost connection with client <{}>", clientID);
}
//publish the Will message (if any) for the clientID
if (m_willStore.containsKey(clientID)) {
WillMessage will = m_willStore.get(clientID);
processPublish(will, clientID);
m_willStore.remove(clientID);
}
}
/**
* Remove the clientID from topic subscription, if not previously subscribed,
* doesn't reply any error
*/
@MQTTMessage(message = UnsubscribeMessage.class)
void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) {
List<String> topics = msg.topicFilters();
int messageID = msg.getMessageID();
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
LOG.debug("processUnsubscribe invoked, removing subscription on topics {}, for clientID <{}>", topics, clientID);
for (String topic : topics) {
subscriptions.removeSubscription(topic, clientID);
}
//ack the client
UnsubAckMessage ackMessage = new UnsubAckMessage();
ackMessage.setMessageID(messageID);
LOG.info("replying with UnsubAck to MSG ID {}", messageID);
session.write(ackMessage);
}
@MQTTMessage(message = SubscribeMessage.class)
void processSubscribe(ServerChannel session, SubscribeMessage msg) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
LOG.debug("processSubscribe invoked from client {} with msgID {}", clientID, msg.getMessageID());
for (SubscribeMessage.Couple req : msg.subscriptions()) {
AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos, cleanSession);
subscribeSingleTopic(newSubscription, req.getTopicFilter());
}
//ack the client
SubAckMessage ackMessage = new SubAckMessage();
ackMessage.setMessageID(msg.getMessageID());
//reply with requested qos
for(SubscribeMessage.Couple req : msg.subscriptions()) {
AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
ackMessage.addType(qos);
}
LOG.debug("replying with SubAck to MSG ID {}", msg.getMessageID());
session.write(ackMessage);
}
private void subscribeSingleTopic(Subscription newSubscription, final String topic) {
LOG.info("<{}> subscribed to topic <{}> with QoS {}",
newSubscription.getClientId(), topic,
AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
String clientID = newSubscription.getClientId();
m_sessionsStore.addNewSubscription(newSubscription, clientID);
subscriptions.add(newSubscription);
//scans retained messages to be published to the new subscription
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
public boolean match(String key) {
return SubscriptionsStore.matchTopics(key, topic);
}
});
for (IMessagesStore.StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("send publish message for topic {}", topic);
sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
}
}
private void disruptorPublish(OutputMessagingEvent msgEvent) {
LOG.debug("disruptorPublish publishing event on output {}", msgEvent);
long sequence = m_ringBuffer.next();
ValueEvent event = m_ringBuffer.get(sequence);
event.setEvent(msgEvent);
m_ringBuffer.publish(sequence);
}
public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
MessagingEvent evt = t.getEvent();
//It's always of type OutputMessagingEvent
OutputMessagingEvent outEvent = (OutputMessagingEvent) evt;
outEvent.getChannel().write(outEvent.getMessage());
}
}
@@ -0,0 +1,194 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.eclipse.moquette.spi.IMessaging;
import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.impl.events.*;
import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
import org.eclipse.moquette.spi.persistence.MapDBPersistentStore;
import org.eclipse.moquette.proto.messages.*;
import org.eclipse.moquette.server.IAuthenticator;
import org.eclipse.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* Singleton class that orchestrate the execution of the protocol.
*
* Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion;
* the requests income from front Netty connectors and are dispatched to the
* ProtocolProcessor.
*
* @author andrea
*/
public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class);
private SubscriptionsStore subscriptions;
private RingBuffer<ValueEvent> m_ringBuffer;
private IMessagesStore m_storageService;
private ISessionsStore m_sessionsStore;
private ExecutorService m_executor;
private Disruptor<ValueEvent> m_disruptor;
private static SimpleMessaging INSTANCE;
private final ProtocolProcessor m_processor = new ProtocolProcessor();
private final AnnotationSupport annotationSupport = new AnnotationSupport();
CountDownLatch m_stopLatch;
private SimpleMessaging() {
}
public static SimpleMessaging getInstance() {
if (INSTANCE == null) {
INSTANCE = new SimpleMessaging();
}
return INSTANCE;
}
public void init(Properties configProps) {
subscriptions = new SubscriptionsStore();
m_executor = Executors.newFixedThreadPool(1);
m_disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
/*Disruptor<ValueEvent> m_disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor,
ProducerType.MULTI, new BusySpinWaitStrategy());*/
m_disruptor.handleEventsWith(this);
m_disruptor.start();
// Get the ring buffer from the Disruptor to be used for publishing.
m_ringBuffer = m_disruptor.getRingBuffer();
annotationSupport.processAnnotations(m_processor);
processInit(configProps);
// disruptorPublish(new InitEvent(configProps));
}
private void disruptorPublish(MessagingEvent msgEvent) {
LOG.debug("disruptorPublish publishing event {}", msgEvent);
long sequence = m_ringBuffer.next();
ValueEvent event = m_ringBuffer.get(sequence);
event.setEvent(msgEvent);
m_ringBuffer.publish(sequence);
}
@Override
public void lostConnection(String clientID) {
disruptorPublish(new LostConnectionEvent(clientID));
}
@Override
public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) {
disruptorPublish(new ProtocolEvent(session, msg));
}
@Override
public void stop() {
m_stopLatch = new CountDownLatch(1);
disruptorPublish(new StopEvent());
try {
//wait the callback notification from the protocol processor thread
LOG.debug("waiting 10 sec to m_stopLatch");
boolean elapsed = !m_stopLatch.await(10, TimeUnit.SECONDS);
LOG.debug("after m_stopLatch");
m_executor.shutdown();
m_disruptor.shutdown();
if (elapsed) {
LOG.error("Can't stop the server in 10 seconds");
}
} catch (InterruptedException ex) {
LOG.error(null, ex);
}
}
@Override
public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
MessagingEvent evt = t.getEvent();
LOG.info("onEvent processing messaging event from input ringbuffer {}", evt);
if (evt instanceof StopEvent) {
processStop();
return;
}
if (evt instanceof LostConnectionEvent) {
LostConnectionEvent lostEvt = (LostConnectionEvent) evt;
m_processor.processConnectionLost(lostEvt.getClientID());
return;
}
if (evt instanceof ProtocolEvent) {
ServerChannel session = ((ProtocolEvent) evt).getSession();
AbstractMessage message = ((ProtocolEvent) evt).getMessage();
annotationSupport.dispatch(session, message);
}
}
private void processInit(Properties props) {
//TODO use a property to select the storage path
MapDBPersistentStore mapStorage = new MapDBPersistentStore();
m_storageService = mapStorage;
m_sessionsStore = mapStorage;
m_storageService.initStore();
//List<Subscription> storedSubscriptions = m_sessionsStore.listAllSubscriptions();
//subscriptions.init(storedSubscriptions);
subscriptions.init(m_sessionsStore);
String passwdPath = props.getProperty("password_file", "");
String configPath = System.getProperty("moquette.path", null);
IAuthenticator authenticator;
if (passwdPath.isEmpty()) {
authenticator = new AcceptAllAuthenticator();
} else {
authenticator = new FileAuthenticator(configPath, passwdPath);
}
m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator);
}
private void processStop() {
LOG.debug("processStop invoked");
m_storageService.close();
LOG.debug("subscription tree {}", subscriptions.dumpTree());
// m_eventProcessor.halt();
// m_executor.shutdown();
subscriptions = null;
m_stopLatch.countDown();
}
}
@@ -0,0 +1,44 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl;
import com.lmax.disruptor.EventFactory;
import org.eclipse.moquette.spi.impl.events.MessagingEvent;
/**
* Carrier value object for the RingBuffer.
*
* @author andrea
*/
public final class ValueEvent {
private MessagingEvent m_event;
public MessagingEvent getEvent() {
return m_event;
}
public void setEvent(MessagingEvent event) {
m_event = event;
}
public final static EventFactory<ValueEvent> EVENT_FACTORY = new EventFactory<ValueEvent>() {
public ValueEvent newInstance() {
return new ValueEvent();
}
};
}
@@ -0,0 +1,34 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.events;
/**
* Used to model the connection lost event
*
* @author andrea
*/
public class LostConnectionEvent extends MessagingEvent{
private String m_clientID;
public LostConnectionEvent(String clienID) {
m_clientID = clienID;
}
public String getClientID() {
return m_clientID;
}
}
@@ -0,0 +1,25 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.events;
/**
*Event used to push work request on the SimpleMessaging.
*
* @author andrea
*/
public abstract class MessagingEvent {
}
@@ -0,0 +1,43 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.events;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.server.ServerChannel;
/**
*
* @author andrea
*/
public class OutputMessagingEvent extends MessagingEvent {
private ServerChannel m_channel;
private AbstractMessage m_message;
public OutputMessagingEvent(ServerChannel channel, AbstractMessage message) {
m_channel = channel;
m_message = message;
}
public ServerChannel getChannel() {
return m_channel;
}
public AbstractMessage getMessage() {
return m_message;
}
}
@@ -0,0 +1,46 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.events;
import org.eclipse.moquette.proto.Utils;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.server.ServerChannel;
/**
* Event used to carry ProtocolMessages from front handler to event processor
*/
public class ProtocolEvent extends MessagingEvent {
ServerChannel m_session;
AbstractMessage message;
public ProtocolEvent(ServerChannel session, AbstractMessage message) {
this.m_session = session;
this.message = message;
}
public ServerChannel getSession() {
return m_session;
}
public AbstractMessage getMessage() {
return message;
}
@Override
public String toString() {
return "ProtocolEvent wrapping " + Utils.msgType2String(message.getMessageType());
}
}
@@ -0,0 +1,39 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.events;
/**
* Used to send the ack message back to the client after a publish
*/
public class PubAckEvent extends MessagingEvent {
int m_messageId;
String m_clientID;
public PubAckEvent(int messageID, String clientID) {
m_messageId = messageID ;
m_clientID = clientID;
}
public int getMessageId() {
return m_messageId;
}
public String getClientID() {
return m_clientID;
}
}
@@ -0,0 +1,82 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.events;
import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType;
import java.nio.ByteBuffer;
/**
*
* @author andrea
*/
public class PublishEvent extends MessagingEvent {
String m_topic;
QOSType m_qos;
//byte[] m_message;
ByteBuffer m_message;
boolean m_retain;
String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
int m_msgID;
public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain,
String clientID, Integer msgID) {
m_topic = topic;
m_qos = qos;
m_message = message;
m_retain = retain;
m_clientID = clientID;
if (qos != QOSType.MOST_ONE) {
m_msgID = msgID;
}
}
public String getTopic() {
return m_topic;
}
public QOSType getQos() {
return m_qos;
}
public ByteBuffer getMessage() {
return m_message;
}
public boolean isRetain() {
return m_retain;
}
public String getClientID() {
return m_clientID;
}
public int getMessageID() {
return m_msgID;
}
@Override
public String toString() {
return "PublishEvent{" +
"m_msgID=" + m_msgID +
", m_clientID='" + m_clientID + '\'' +
", m_retain=" + m_retain +
", m_qos=" + m_qos +
", m_topic='" + m_topic + '\'' +
'}';
}
}
@@ -0,0 +1,31 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.events;
/**
* Used to re-fire stored QoS1 QoS2 events once a client reconnects.
*/
public class RepublishEvent extends MessagingEvent {
private String m_clientID;
public RepublishEvent(String clientID) {
this.m_clientID = clientID;
}
public String getClientID() {
return m_clientID;
}
}
@@ -0,0 +1,25 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.events;
/**
* Poison pill to shutdown the event loop
*
* @author andrea
*/
public class StopEvent extends MessagingEvent {
}
@@ -0,0 +1,42 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.events;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
/**
*
* @author andrea
*/
public class SubscribeEvent extends MessagingEvent {
Subscription m_subscription;
int m_messageID;
public SubscribeEvent(Subscription subscription, int messageID) {
m_subscription = subscription;
m_messageID = messageID;
}
public Subscription getSubscription() {
return m_subscription;
}
public int getMessageID() {
return m_messageID;
}
}
@@ -0,0 +1,73 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.storage;
import java.io.Serializable;
import java.nio.ByteBuffer;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType;
/**
* Publish event serialized to the DB.
*
* @author andrea
*/
public class StoredPublishEvent implements Serializable {
String m_topic;
QOSType m_qos;
byte[] m_message;
boolean m_retain;
String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
int m_msgID;
public StoredPublishEvent(PublishEvent wrapped) {
m_topic = wrapped.getTopic();
m_qos = wrapped.getQos();
m_retain = wrapped.isRetain();
m_clientID = wrapped.getClientID();
m_msgID = wrapped.getMessageID();
ByteBuffer buffer = wrapped.getMessage();
m_message = new byte[buffer.remaining()];
buffer.get(m_message);
buffer.rewind();
}
public String getTopic() {
return m_topic;
}
public QOSType getQos() {
return m_qos;
}
public byte[] getMessage() {
return m_message;
}
public boolean isRetain() {
return m_retain;
}
public String getClientID() {
return m_clientID;
}
public int getMessageID() {
return m_msgID;
}
}
@@ -0,0 +1,115 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.subscriptions;
import java.io.Serializable;
import org.eclipse.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;
}
/**
* Factory method for empty subscriptions
*/
public static final Subscription createEmptySubscription(String clientId, boolean cleanSession) {
return new Subscription(clientId, "", QOSType.MOST_ONE, cleanSession);
}
public String getClientId() {
return clientId;
}
public QOSType getRequestedQos() {
return requestedQos;
}
public String getTopicFilter() {
return topicFilter;
}
public boolean isCleanSession() {
return this.cleanSession;
}
public boolean isActive() {
return active;
}
public void setActive(boolean active) {
this.active = active;
}
@Override
public boolean equals(Object obj) {
if (obj == null) {
return false;
}
if (getClass() != obj.getClass()) {
return false;
}
final Subscription other = (Subscription) obj;
if (this.requestedQos != other.requestedQos) {
return false;
}
if ((this.clientId == null) ? (other.clientId != null) : !this.clientId.equals(other.clientId)) {
return false;
}
if ((this.topicFilter == null) ? (other.topicFilter != null) : !this.topicFilter.equals(other.topicFilter)) {
return false;
}
return true;
}
@Override
public int hashCode() {
int hash = 3;
hash = 37 * hash + (this.requestedQos != null ? this.requestedQos.hashCode() : 0);
hash = 37 * hash + (this.clientId != null ? this.clientId.hashCode() : 0);
hash = 37 * hash + (this.topicFilter != null ? this.topicFilter.hashCode() : 0);
return hash;
}
/**
* Trivial match method
*/
boolean match(String topic) {
return this.topicFilter.equals(topic);
}
@Override
public String toString() {
return String.format("[filter:%s, cliID: %s, qos: %s, active: %s]", this.topicFilter, this.clientId, this.requestedQos, this.active);
}
}
@@ -0,0 +1,318 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.subscriptions;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import org.eclipse.moquette.spi.ISessionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Represents a tree of topics subscriptions.
*
* @author andrea
*/
public class SubscriptionsStore {
public static interface IVisitor<T> {
void visit(TreeNode node);
T getResult();
}
private class DumpTreeVisitor implements IVisitor<String> {
String s = "";
public void visit(TreeNode node) {
String subScriptionsStr = "";
for (Subscription sub : node.m_subscriptions) {
subScriptionsStr += sub.toString();
}
s += node.getToken() == null ? "" : node.getToken().toString();
s += subScriptionsStr + "\n";
}
public String getResult() {
return s;
}
}
private class SubscriptionTreeCollector implements IVisitor<List<Subscription>> {
private List<Subscription> m_allSubscriptions = new ArrayList<Subscription>();
public void visit(TreeNode node) {
m_allSubscriptions.addAll(node.subscriptions());
}
public List<Subscription> getResult() {
return m_allSubscriptions;
}
}
private TreeNode subscriptions = new TreeNode(null);
private ISessionsStore m_sessionsStore;
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
/**
* Initialize the subscription tree with the list of subscriptions.
*/
public void init(ISessionsStore sessionsStore) {
LOG.debug("init invoked");
m_sessionsStore = sessionsStore;
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());
addDirect(subscription);
}
if (LOG.isDebugEnabled()) {
LOG.debug("Finished loading. Subscription tree after {}", dumpTree());
}
}
protected void addDirect(Subscription newSubscription) {
TreeNode current = findMatchingNode(newSubscription.topicFilter);
current.addSubscription(newSubscription);
}
private TreeNode findMatchingNode(String topic) {
List<Token> tokens = new ArrayList<Token>();
try {
tokens = parseTopic(topic);
} catch (ParseException ex) {
//TODO handle the parse exception
LOG.error(null, ex);
// return;
}
TreeNode current = subscriptions;
for (Token token : tokens) {
TreeNode matchingChildren;
//check if a children with the same token already exists
if ((matchingChildren = current.childWithToken(token)) != null) {
current = matchingChildren;
} else {
//create a new node for the newly inserted token
matchingChildren = new TreeNode(current);
matchingChildren.setToken(token);
current.addChild(matchingChildren);
current = matchingChildren;
}
}
return current;
}
public void add(Subscription newSubscription) {
addDirect(newSubscription);
//log the subscription
// String clientID = newSubscription.getClientId();
// m_storageService.addNewSubscription(newSubscription, clientID);
}
public void removeSubscription(String topic, String clientID) {
TreeNode matchNode = findMatchingNode(topic);
//search for the subscription to remove
Subscription toBeRemoved = null;
for (Subscription sub : matchNode.subscriptions()) {
if (sub.topicFilter.equals(topic) && sub.getClientId().equals(clientID)) {
toBeRemoved = sub;
break;
}
}
if (toBeRemoved != null) {
matchNode.subscriptions().remove(toBeRemoved);
}
}
/**
* TODO implement testing
*/
public void clearAllSubscriptions() {
SubscriptionTreeCollector subsCollector = new SubscriptionTreeCollector();
bfsVisit(subscriptions, subsCollector);
List<Subscription> allSubscriptions = subsCollector.getResult();
for (Subscription subscription : allSubscriptions) {
removeSubscription(subscription.getTopicFilter(), subscription.getClientId());
}
}
/**
* Visit the topics tree to remove matching subscriptions with clientID
*/
public void removeForClient(String clientID) {
subscriptions.removeClientSubscriptions(clientID);
//persist the update
m_sessionsStore.wipeSubscriptions(clientID);
}
public void deactivate(String clientID) {
subscriptions.deactivate(clientID);
//persist the update
Set<Subscription> subs = subscriptions.findAllByClientID(clientID);
m_sessionsStore.updateSubscriptions(clientID, subs);
}
public void activate(String clientID) {
LOG.debug("Activating subscriptions for clientID <{}>", clientID);
subscriptions.activate(clientID);
//persist the update
Set<Subscription> subs = subscriptions.findAllByClientID(clientID);
m_sessionsStore.updateSubscriptions(clientID, subs);
}
/**
* 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.EMPTY_LIST;
}
Queue<Token> tokenQueue = new LinkedBlockingDeque<Token>(tokens);
List<Subscription> matchingSubs = new ArrayList<Subscription>();
subscriptions.matches(tokenQueue, matchingSubs);
return matchingSubs;
}
public boolean contains(Subscription sub) {
return !matches(sub.topicFilter).isEmpty();
}
public int size() {
return subscriptions.size();
}
public String dumpTree() {
DumpTreeVisitor visitor = new DumpTreeVisitor();
bfsVisit(subscriptions, visitor);
return visitor.getResult();
}
private void bfsVisit(TreeNode node, IVisitor visitor) {
if (node == null) {
return;
}
visitor.visit(node);
for (TreeNode child : node.m_children) {
bfsVisit(child, visitor);
}
}
/**
* 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;
Token subToken = null;
for (; i< subscriptionTokens.size(); i++) {
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 res = new ArrayList<Token>();
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,71 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.impl.subscriptions;
class Token {
static final Token EMPTY = new Token("");
static final Token MULTI = new Token("#");
static final Token SINGLE = new Token("+");
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,198 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.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<TreeNode>();
List<Subscription> m_subscriptions = new ArrayList<Subscription>();
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
int existingSubIdx = Collections.binarySearch(m_subscriptions, s, new ClientIDComparator());
if (existingSubIdx >= 0) {
m_subscriptions.remove(existingSubIdx);
}
m_subscriptions.add(s);
}
void addChild(TreeNode child) {
m_children.add(child);
}
boolean isLeaf() {
return m_children.isEmpty();
}
/**
* 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;
}
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<Token>(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;
}
void removeClientSubscriptions(String clientID) {
//collect what to delete and then delete to avoid ConcurrentModification
List<Subscription> subsToRemove = new ArrayList<Subscription>();
for (Subscription s : m_subscriptions) {
if (s.clientId.equals(clientID)) {
subsToRemove.add(s);
}
}
for (Subscription s : subsToRemove) {
m_subscriptions.remove(s);
}
//go deep
for (TreeNode child : m_children) {
child.removeClientSubscriptions(clientID);
}
}
/**
* Deactivate all topic subscriptions for the given clientID.
* */
void deactivate(String clientID) {
for (Subscription s : m_subscriptions) {
if (s.clientId.equals(clientID)) {
s.setActive(false);
}
}
//go deep
for (TreeNode child : m_children) {
child.deactivate(clientID);
}
}
/**
* Activate all topic subscriptions for the given clientID.
* */
public void activate(String clientID) {
for (Subscription s : m_subscriptions) {
if (s.clientId.equals(clientID)) {
s.setActive(true);
}
}
//go deep
for (TreeNode child : m_children) {
child.activate(clientID);
}
}
/**
* @return the set of subscriptions for the given client.
* */
Set<Subscription> findAllByClientID(String clientID) {
Set<Subscription> subs = new HashSet<Subscription>();
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,261 @@
/*
* Copyright (c) 2012-2014 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 org.eclipse.moquette.spi.persistence;
import org.eclipse.moquette.proto.MQTTException;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.spi.impl.storage.StoredPublishEvent;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import static org.eclipse.moquette.server.Server.STORAGE_FILE_PATH;
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.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentMap;
/**
* MapDB main persistence implementation
*/
public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class);
private ConcurrentMap<String, StoredMessage> m_retainedStore;
//maps clientID to the list of pending messages stored
private ConcurrentMap<String, List<StoredPublishEvent>> m_persistentMessageStore;
//bind clientID+MsgID -> evt message published
private ConcurrentMap<String, StoredPublishEvent> m_inflightStore;
//bind clientID+MsgID -> evt message published
private ConcurrentMap<String, StoredPublishEvent> m_qos2Store;
//persistent Map of clientID, set of Subscriptions
private ConcurrentMap<String, Set<Subscription>> m_persistentSubscriptions;
private DB m_db;
@Override
public void initStore() {
File tmpFile;
try {
tmpFile = new File(STORAGE_FILE_PATH);
tmpFile.createNewFile();
} catch (IOException ex) {
LOG.error(null, ex);
throw new MQTTException("Can't create temp file for subscriptions storage [" + STORAGE_FILE_PATH + "]", ex);
}
m_db = DBMaker.newFileDB(tmpFile)
.make();
m_retainedStore = m_db.getHashMap("retained");
m_persistentMessageStore = m_db.getHashMap("persistedMessages");
m_inflightStore = m_db.getHashMap("inflight");
m_persistentSubscriptions = m_db.getHashMap("subscriptions");
m_qos2Store = m_db.getHashMap("qos2Store");
}
@Override
public void cleanRetained(String topic) {
m_retainedStore.remove(topic);
}
@Override
public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) {
if (!message.hasRemaining()) {
//clean the message from topic
m_retainedStore.remove(topic);
} else {
//store the message to the topic
byte[] raw = new byte[message.remaining()];
message.get(raw);
m_retainedStore.put(topic, new StoredMessage(raw, qos, topic));
}
m_db.commit();
}
@Override
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
List<StoredMessage> results = new ArrayList<StoredMessage>();
for (Map.Entry<String, StoredMessage> entry : m_retainedStore.entrySet()) {
StoredMessage storedMsg = entry.getValue();
if (condition.match(entry.getKey())) {
results.add(storedMsg);
}
}
return results;
}
@Override
public void storePublishForFuture(PublishEvent evt) {
List<StoredPublishEvent> storedEvents;
String clientID = evt.getClientID();
if (!m_persistentMessageStore.containsKey(clientID)) {
storedEvents = new ArrayList<StoredPublishEvent>();
} else {
storedEvents = m_persistentMessageStore.get(clientID);
}
storedEvents.add(convertToStored(evt));
m_persistentMessageStore.put(clientID, storedEvents);
m_db.commit();
//NB rewind the evt message content
LOG.debug("Stored published message for client <{}> on topic <{}>", clientID, evt.getTopic());
}
@Override
public List<PublishEvent> retrievePersistedPublishes(String clientID) {
List<StoredPublishEvent> storedEvts = m_persistentMessageStore.get(clientID);
if (storedEvts == null) {
return null;
}
List<PublishEvent> liveEvts = new ArrayList<PublishEvent>();
for (StoredPublishEvent storedEvt : storedEvts) {
liveEvts.add(convertFromStored(storedEvt));
}
return liveEvts;
}
@Override
public void cleanPersistedPublishMessage(String clientID, int messageID) {
List<StoredPublishEvent> events = m_persistentMessageStore.get(clientID);
if (events == null) {
return;
}
StoredPublishEvent toRemoveEvt = null;
for (StoredPublishEvent evt : events) {
if (evt.getMessageID() == messageID) {
toRemoveEvt = evt;
}
}
events.remove(toRemoveEvt);
m_persistentMessageStore.put(clientID, events);
m_db.commit();
}
public void cleanPersistedPublishes(String clientID) {
m_persistentMessageStore.remove(clientID);
m_db.commit();
}
public void cleanInFlight(String msgID) {
m_inflightStore.remove(msgID);
m_db.commit();
}
public void addInFlight(PublishEvent evt, String publishKey) {
StoredPublishEvent storedEvt = convertToStored(evt);
m_inflightStore.put(publishKey, storedEvt);
m_db.commit();
}
public void addNewSubscription(Subscription newSubscription, String clientID) {
LOG.debug("addNewSubscription invoked with subscription {} for client {}", newSubscription, clientID);
if (!m_persistentSubscriptions.containsKey(clientID)) {
LOG.debug("clientID {} is a newcome, creating it's subscriptions set", clientID);
m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
}
Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
if (!subs.contains(newSubscription)) {
LOG.debug("updating clientID {} 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);
}
m_db.commit();
}
public void wipeSubscriptions(String clientID) {
m_persistentSubscriptions.remove(clientID);
m_db.commit();
}
@Override
public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
m_persistentSubscriptions.put(clientID, subscriptions);
m_db.commit();
}
public List<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<Subscription>();
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);
}
public void close() {
this.m_db.commit();
LOG.debug("persisted subscriptions {}", m_persistentSubscriptions);
this.m_db.close();
LOG.debug("closed disk storage");
}
/*-------- QoS 2 storage management --------------*/
public void persistQoS2Message(String publishKey, PublishEvent evt) {
LOG.debug("persistQoS2Message store pubKey: {}, evt: {}", publishKey, evt);
m_qos2Store.put(publishKey, convertToStored(evt));
}
public void removeQoS2Message(String publishKey) {
m_qos2Store.remove(publishKey);
}
public PublishEvent retrieveQoS2Message(String publishKey) {
StoredPublishEvent storedEvt = m_qos2Store.get(publishKey);
return convertFromStored(storedEvt);
}
private StoredPublishEvent convertToStored(PublishEvent evt) {
StoredPublishEvent storedEvt = new StoredPublishEvent(evt);
return storedEvt;
}
private PublishEvent convertFromStored(StoredPublishEvent evt) {
byte[] message = evt.getMessage();
ByteBuffer bbmessage = ByteBuffer.wrap(message);
//bbmessage.flip();
PublishEvent liveEvt = new PublishEvent(evt.getTopic(), evt.getQos(),
bbmessage, evt.isRetain(), evt.getClientID(), evt.getMessageID());
return liveEvt;
}
}