anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removed dependency on Mina IO framework

Browse Source
This commit is contained in:
andrea
2013-11-24 11:51:15 +01:00
parent 817ca8e176
commit 1fa810362f
20 changed files with 428 additions and 1387 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarking/pom.xml
-69
View File
@@ -1,69 +0,0 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<relativePath>../</relativePath>
<!--<relativePath>../pom.xml</relativePath>-->
<artifactId>moquette-parent</artifactId>
<groupId>org.dna.mqtt</groupId>
<version>0.4-SNAPSHOT</version>
</parent>
<artifactId>moquette-benchmark</artifactId>
<packaging>jar</packaging>
<version>0.1</version>
<name>Moquette - Benchmarking tool</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.6.4</version>
</dependency>
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-mina-parser</artifactId>
<version>0.2</version>
</dependency>
<dependency>
<groupId>org.fusesource.mqtt-client</groupId>
<artifactId>mqtt-client</artifactId>
<version>1.3</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.5</source>
<target>1.5</target>
</configuration>
</plugin>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<archive>
<manifest>
<mainClass>org.dna.mqtt.bechnmark.mina.MQTTDropServer</mainClass>
</manifest>
</archive>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
</plugin>
</plugins>
</build>
</project>
benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerBlocking.java
-78
View File
@@ -1,78 +0,0 @@
package org.dna.mqtt.bechnmark;
import java.net.URISyntaxException;
import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class ConsumerBlocking implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(ConsumerBlocking.class);
private String m_clientID;
public ConsumerBlocking(String clientID) {
m_clientID = clientID;
}
public void run() {
MQTT mqtt = new MQTT();
try {
// mqtt.setHost("test.mosquitto.org", 1883);
mqtt.setHost("localhost", 1883);
} catch (URISyntaxException ex) {
LOG.error(null, ex);
return;
}
mqtt.setClientId(m_clientID);
BlockingConnection connection = mqtt.blockingConnection();
try {
connection.connect();
} catch (Exception ex) {
LOG.error("Cant't CONNECT to the server", ex);
return;
}
try {
Topic[] topics = {new Topic("/topic", QoS.AT_MOST_ONCE)};
byte[] qoses = connection.subscribe(topics);
LOG.info("Subscribed to topic");
} catch (Exception ex) {
LOG.error("Cant't PUSBLISH to the server", ex);
return;
}
Message message = null;
for (int i = 0; i < Producer.PUB_LOOP; i++) {
try {
message = connection.receive();
} catch (Exception ex) {
LOG.error(null, ex);
return;
}
byte[] payload = message.getPayload();
StringBuffer sb = new StringBuffer().append("Topic: ").append(message.getTopic())
.append(", payload: ").append(new String(payload));
LOG.info(sb.toString());
}
try {
LOG.info("Disconneting");
connection.disconnect();
LOG.info("Disconnected");
} catch (Exception ex) {
LOG.error("Cant't DISCONNECT to the server", ex);
}
}
}
benchmarking/src/main/java/org/dna/mqtt/bechnmark/ConsumerFuture.java
-109
View File
@@ -1,109 +0,0 @@
package org.dna.mqtt.bechnmark;
import java.io.*;
import java.net.URISyntaxException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.fusesource.mqtt.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class ConsumerFuture implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(ConsumerFuture.class);
private String m_clientID;
private static final String BENCHMARK_FILE = "consumer_bechmark.txt";
private PrintWriter m_benchMarkOut;
private ByteArrayOutputStream m_baos = new ByteArrayOutputStream(1024 * 1024);
public ConsumerFuture(String clientID) {
m_clientID = clientID;
m_benchMarkOut = new PrintWriter(new BufferedWriter(new OutputStreamWriter(m_baos)));
m_benchMarkOut.println("msg ID, ns");
}
public void run() {
MQTT mqtt = new MQTT();
try {
// mqtt.setHost("test.mosquitto.org", 1883);
mqtt.setHost("localhost", 1883);
} catch (URISyntaxException ex) {
LOG.error(null, ex);
return;
}
mqtt.setClientId(m_clientID);
FutureConnection connection = mqtt.futureConnection();
Future<Void> futConn = connection.connect();
try {
futConn.await();
} catch (Exception ex) {
LOG.error("Cant't CONNECT to the server", ex);
return;
}
Topic[] topics = {new Topic("/topic", QoS.AT_MOST_ONCE)};
Future<byte[]> futSub = connection.subscribe(topics);
try {
byte[] qoses = futSub.await();
LOG.info("Subscribed to topic");
} catch (Exception ex) {
LOG.error("Cant't PUSBLISH to the server", ex);
return;
}
Pattern p = Pattern.compile(".*!!(\\d+)");
Message message = null;
long startMillis = System.currentTimeMillis();
for (int i = 0; i < Producer.PUB_LOOP; i++) {
Future<Message> futReceive = connection.receive();
try {
message = futReceive.await();
String content = new String(message.getPayload());
//metrics part
Matcher matcher = p.matcher(content);
matcher.matches();
String numMsg = matcher.group(1);
m_benchMarkOut.println(String.format("%s, %d", numMsg, System.nanoTime()));
} catch (Exception ex) {
LOG.error(null, ex);
return;
}
byte[] payload = message.getPayload();
StringBuffer sb = new StringBuffer().append("Topic: ").append(message.getTopic())
.append(", payload: ").append(new String(payload));
LOG.debug(sb.toString());
}
long time = System.currentTimeMillis() - startMillis;
LOG.info(String.format("Consumer %s received %d messages in %d ms", Thread.currentThread().getName(), Producer.PUB_LOOP, time));
Future<Void> f4 = connection.disconnect();
try {
LOG.info("Disconneting");
f4.await();
LOG.info("Disconnected");
} catch (Exception ex) {
LOG.error("Cant't DISCONNECT to the server", ex);
}
m_benchMarkOut.close();
try {
FileOutputStream fw = new FileOutputStream(BENCHMARK_FILE);
fw.write(m_baos.toByteArray());
fw.close();
} catch (IOException e) {
e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
}
}
}
benchmarking/src/main/java/org/dna/mqtt/bechnmark/Main.java
-34
View File
@@ -1,34 +0,0 @@
package org.dna.mqtt.bechnmark;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
*
* Benchmark tool entry point.
*
* You could set the number of producers and consumers
*
* @author andrea
*/
public class Main {
private static final int PUBLISHER_POOL_SIZE = 2;
private static final int NUM_CONCURRENT_PRODS = 2;
public static void main(String[] args) throws InterruptedException {
ExecutorService consumerPool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE);
ExecutorService producerPool = Executors.newFixedThreadPool(PUBLISHER_POOL_SIZE);
consumerPool.submit(new ConsumerFuture("Cons1"));
//force wait to let the consumer be registered before the producer
Thread.sleep(1000);
int len = Producer.PUB_LOOP / NUM_CONCURRENT_PRODS;
for (int i = 0; i < NUM_CONCURRENT_PRODS; i++) {
producerPool.submit(new Producer("Prod " + i, i * len, len));
}
consumerPool.shutdown();
producerPool.shutdown();
}
}
benchmarking/src/main/java/org/dna/mqtt/bechnmark/Producer.java
-104
View File
@@ -1,104 +0,0 @@
package org.dna.mqtt.bechnmark;
import java.io.*;
import java.net.URISyntaxException;
import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* This class only publish MQTT messages to a define topic with a certain frequency.
*
*
* @author andrea
*/
public class Producer implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(Producer.class);
private String m_clientID;
public static final int PUB_LOOP = 100000;
private int m_starIndex;
private int m_len;
private long m_startMillis;
private static final String BENCHMARK_FILE = "producer_bechmark_%s.txt";
private PrintWriter m_benchMarkOut;
private ByteArrayOutputStream m_baos = new ByteArrayOutputStream(1024 * 1024 * 2);
public Producer(String clientID, int start, int len) {
m_clientID = clientID;
m_starIndex = start;
m_len = len;
m_benchMarkOut = new PrintWriter(new BufferedWriter(new OutputStreamWriter(m_baos)));
m_benchMarkOut.println("msg ID, ns");
}
public void run() {
m_startMillis = System.currentTimeMillis();
MQTT mqtt = new MQTT();
try {
// mqtt.setHost("test.mosquitto.org", 1883);
mqtt.setHost("localhost", 1883);
} catch (URISyntaxException ex) {
LOG.error(null, ex);
return;
}
mqtt.setClientId(m_clientID);
BlockingConnection connection = mqtt.blockingConnection();
try {
connection.connect();
} catch (Exception ex) {
LOG.error("Cant't CONNECT to the server", ex);
return;
}
long time = System.currentTimeMillis() - m_startMillis;
LOG.info(String.format("Producer %s connected in %d ms", Thread.currentThread().getName(), time));
LOG.info("Starting from index " + m_starIndex + " up to " + (m_starIndex + m_len));
m_startMillis = System.currentTimeMillis();
for (int i = m_starIndex; i < m_starIndex + m_len; i++) {
try {
// LOG.info("Publishing");
String payload = "Hello world MQTT!!" + i;
connection.publish("/topic", payload.getBytes(), QoS.AT_MOST_ONCE, false);
m_benchMarkOut.println(String.format("%d, %d", i, System.nanoTime()));
} catch (Exception ex) {
LOG.error("Cant't PUBLISH to the server", ex);
return;
}
}
time = System.currentTimeMillis() - m_startMillis;
LOG.info(String.format("Producer %s published %d messages in %d ms", Thread.currentThread().getName(), m_len, time));
m_startMillis = System.currentTimeMillis();
try {
LOG.info("Disconneting");
connection.disconnect();
LOG.info("Disconnected");
} catch (Exception ex) {
LOG.error("Cant't DISCONNECT to the server", ex);
}
time = System.currentTimeMillis() - m_startMillis;
LOG.info(String.format("Producer %s disconnected in %d ms", Thread.currentThread().getName(), time));
m_benchMarkOut.flush();
m_benchMarkOut.close();
try {
FileOutputStream fw = new FileOutputStream(String.format(BENCHMARK_FILE, m_clientID));
fw.write(m_baos.toByteArray());
fw.close();
} catch (IOException e) {
e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
}
}
}
benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/DummyClientHandler.java
-23
View File
@@ -1,23 +0,0 @@
package org.dna.mqtt.bechnmark.mina;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class DummyClientHandler extends IoHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(DummyClientHandler.class);
@Override
public void messageReceived(IoSession session, Object message) throws Exception {
AbstractMessage msg = (AbstractMessage) message;
LOG.debug("Received a message of type " + msg.getMessageType());
}
}
benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTBulkClient.java
-126
View File
@@ -1,126 +0,0 @@
package org.dna.mqtt.bechnmark.mina;
import java.io.IOException;
import java.net.InetSocketAddress;
import org.apache.mina.core.RuntimeIoException;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.service.IoConnector;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.dna.mqtt.commons.Constants;
import org.dna.mqtt.moquette.MQTTException;
import org.dna.mqtt.moquette.PublishException;
import org.dna.mqtt.moquette.proto.*;
import org.dna.mqtt.moquette.proto.messages.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class MQTTBulkClient {
public int NUM_MESSAGES = 10000;
private static final Logger LOG = LoggerFactory.getLogger(MQTTBulkClient.class);
private String m_hostname;
private int m_port = 1883;
//internal management used for conversation with the server
private IoConnector m_connector;
private IoSession m_session;
boolean m_withWaitWriteFuture = false;
public static void main(String[] args) throws IOException {
MQTTBulkClient client = new MQTTBulkClient();
if (args.length > 0) {
client.m_hostname = args[0];
client.m_withWaitWriteFuture = Boolean.parseBoolean(args[1]);
client.NUM_MESSAGES = Integer.parseInt(args[2]);
} else {
client.m_hostname = "localhost";
}
client.init();
client.connect();
long start = System.currentTimeMillis();
for (int i = 0; i < client.NUM_MESSAGES; i++) {
client.publish("/topic", "Hello world".getBytes(), AbstractMessage.QOSType.MOST_ONE, false);
}
long stop = System.currentTimeMillis();
LOG.info("Client sent " + client.NUM_MESSAGES + " in " + (stop - start) + " ms");
client.shutdown();
}
protected void init() {
DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
decoder.addMessageDecoder(new ConnAckDecoder());
decoder.addMessageDecoder(new SubAckDecoder());
decoder.addMessageDecoder(new UnsubAckDecoder());
decoder.addMessageDecoder(new PublishDecoder());
decoder.addMessageDecoder(new PubAckDecoder());
decoder.addMessageDecoder(new PingRespDecoder());
DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder());
encoder.addMessageEncoder(SubscribeMessage.class, new SubscribeEncoder());
encoder.addMessageEncoder(UnsubscribeMessage.class, new UnsubscribeEncoder());
encoder.addMessageEncoder(DisconnectMessage.class, new DisconnectEncoder());
encoder.addMessageEncoder(PingReqMessage.class, new PingReqEncoder());
m_connector = new NioSocketConnector();
// m_connector.getFilterChain().addLast("logger", new LoggingFilter());
m_connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder));
m_connector.setHandler(new DummyClientHandler());
m_connector.getSessionConfig().setReadBufferSize(2048);
m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Constants.DEFAULT_CONNECT_TIMEOUT);
}
public void connect() throws MQTTException {
try {
ConnectFuture future = m_connector.connect(new InetSocketAddress(m_hostname, m_port));
LOG.debug("Client waiting to connect to server");
future.awaitUninterruptibly();
m_session = future.getSession();
} catch (RuntimeIoException e) {
LOG.debug("Failed to connect", e);
}
}
public void publish(String topic, byte[] payload, AbstractMessage.QOSType qos, boolean retain) throws PublishException {
PublishMessage msg = new PublishMessage();
msg.setRetainFlag(retain);
msg.setTopicName(topic);
msg.setPayload(payload);
//QoS 0 case
msg.setQos(AbstractMessage.QOSType.MOST_ONE);
WriteFuture wf = m_session.write(msg);
if (m_withWaitWriteFuture) {
try {
wf.await();
} catch (InterruptedException ex) {
LOG.debug(null, ex);
throw new PublishException(ex);
}
Throwable ex = wf.getException();
if (ex != null) {
throw new PublishException(ex);
}
}
}
public void shutdown() {
m_connector.dispose();
}
}
benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTDropHandler.java
-44
View File
@@ -1,44 +0,0 @@
package org.dna.mqtt.bechnmark.mina;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
/**
*
* @author andrea
*/
public class MQTTDropHandler extends IoHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(MQTTDropHandler.class);
@Override
public void messageReceived(IoSession session, Object message) throws Exception {
AbstractMessage msg = (AbstractMessage) message;
try {
switch (msg.getMessageType()) {
case CONNECT:
case SUBSCRIBE:
case UNSUBSCRIBE:
case PUBLISH:
case PINGREQ:
case DISCONNECT:
LOG.debug("Received a message of type " + msg.getMessageType());
break;
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
}
}
@Override
public void sessionIdle(IoSession session, IdleStatus status) {
session.close(false);
}
}
benchmarking/src/main/java/org/dna/mqtt/bechnmark/mina/MQTTDropServer.java
-97
View File
@@ -1,97 +0,0 @@
package org.dna.mqtt.bechnmark.mina;
import java.io.IOException;
import java.net.InetSocketAddress;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.service.IoServiceStatistics;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.dna.mqtt.moquette.proto.*;
import org.dna.mqtt.moquette.proto.messages.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class MQTTDropServer {
private static final Logger LOG = LoggerFactory.getLogger(MQTTDropServer.class);
public static final int PORT = 1883;
public static final int DEFAULT_CONNECT_TIMEOUT = 10;
private IoAcceptor m_acceptor;
public static void main(String[] args) throws IOException {
new MQTTDropServer().startServer();
}
protected void startServer() throws IOException {
DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
decoder.addMessageDecoder(new ConnectDecoder());
decoder.addMessageDecoder(new PublishDecoder());
decoder.addMessageDecoder(new SubscribeDecoder());
decoder.addMessageDecoder(new UnsubscribeDecoder());
decoder.addMessageDecoder(new DisconnectDecoder());
decoder.addMessageDecoder(new PingReqDecoder());
DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
// encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder());
encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder());
encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder());
encoder.addMessageEncoder(PubAckMessage.class, new PubAckEncoder());
encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder());
encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder());
m_acceptor = new NioSocketAcceptor();
m_acceptor.getFilterChain().addLast( "logger", new MQTTLoggingFilter("SERVER LOG") );
m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
MQTTDropHandler handler = new MQTTDropHandler();
m_acceptor.setHandler(handler);
((NioSocketAcceptor)m_acceptor).setReuseAddress(true);
((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true);
m_acceptor.getSessionConfig().setReadBufferSize( 2048 );
m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, DEFAULT_CONNECT_TIMEOUT );
m_acceptor.getStatistics().setThroughputCalculationInterval(10);
m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis());
m_acceptor.bind( new InetSocketAddress(PORT) );
LOG.info("Server binded");
//Bind a shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
stopServer();
}
});
}
protected void stopServer() {
LOG.info("Server stopping...");
//log statistics
IoServiceStatistics statistics = m_acceptor.getStatistics();
statistics.updateThroughput(System.currentTimeMillis());
System.out.println(String.format("Total read bytes: %d, read throughtput: %f (b/s)", statistics.getReadBytes(), statistics.getReadBytesThroughput()));
System.out.println(String.format("Total read msgs: %d, read msg throughtput: %f (msg/s)", statistics.getReadMessages(), statistics.getReadMessagesThroughput()));
for(IoSession session: m_acceptor.getManagedSessions().values()) {
if(session.isConnected() && !session.isClosing()){
session.close(false);
}
}
m_acceptor.unbind();
m_acceptor.dispose();
LOG.info("Server stopped");
}
}
broker/pom.xml
+1 -7
View File
@@ -22,7 +22,7 @@
<dependencies>
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-mina-parser</artifactId>
<artifactId>moquette-parser-commons</artifactId>
<version>${project.version}</version>
</dependency>
@@ -31,12 +31,6 @@
<artifactId>moquette-netty-parser</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.mina</groupId>
<artifactId>mina-core</artifactId>
<version>${mina-version}</version>
</dependency>
<dependency>
<groupId>com.googlecode.disruptor</groupId>
broker/src/main/java/org/dna/mqtt/moquette/server/Server.java
+3 -33
View File
@@ -3,7 +3,7 @@ package org.dna.mqtt.moquette.server;
import java.io.File;
import java.io.IOException;
import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging;
import org.dna.mqtt.moquette.server.mina.MinaAcceptor;
//import org.dna.mqtt.moquette.server.mina.MinaAcceptor;
import org.dna.mqtt.moquette.server.netty.NettyAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -17,27 +17,14 @@ public class Server {
public static final String STORAGE_FILE_PATH = System.getProperty("user.home") +
File.separator + "moquette_store.hawtdb";
public static String NETTY_NIO_FRAMEWORK = "netty";
public static String MINA_NIO_FRAMEWORK = "mina";
public static String DEFAULT_NIO_FRAMEWORK = NETTY_NIO_FRAMEWORK;
private ServerAcceptor m_acceptor;
SimpleMessaging messaging;
public static void main(String[] args) throws IOException {
String nioFrameworkType = DEFAULT_NIO_FRAMEWORK;
if (args.length > 0) {
//TODO check the typing and in case print usge message
nioFrameworkType = args[1];
if (!(nioFrameworkType.equals(DEFAULT_NIO_FRAMEWORK) ||
nioFrameworkType.equals(DEFAULT_NIO_FRAMEWORK))) {
printUsage(nioFrameworkType);
return;
}
}
final Server server = new Server();
server.startServer(nioFrameworkType);
server.startServer();
//Bind a shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
@@ -49,19 +36,10 @@ public class Server {
}
public void startServer() throws IOException {
startServer(DEFAULT_NIO_FRAMEWORK);
}
public void startServer(String nioFrameworkType) throws IOException {
System.out.println("Starting server with " + nioFrameworkType + " connectors");
messaging = SimpleMessaging.getInstance();
messaging.init();
if (nioFrameworkType.equals(MINA_NIO_FRAMEWORK)) {
m_acceptor = new MinaAcceptor();
} else if (nioFrameworkType.equals(NETTY_NIO_FRAMEWORK)) {
m_acceptor = new NettyAcceptor();
}
m_acceptor = new NettyAcceptor();
m_acceptor.initialize(messaging);
}
@@ -71,12 +49,4 @@ public class Server {
m_acceptor.close();
System.out.println("Server stopped");
}
private static void printUsage(String nioFrameworkType) {
System.out.println("The broker should be invoked with the following command line: \n");
System.out.println("> java -jar <broker.jar> [mina|netty]\n");
System.out.println("while was invoked with type: " + nioFrameworkType);
}
}
broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java
-121
View File
@@ -1,121 +0,0 @@
package org.dna.mqtt.moquette.server.mina;
import java.io.IOException;
import java.net.InetSocketAddress;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.service.IoServiceStatistics;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.dna.mqtt.commons.Constants;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.proto.ConnAckEncoder;
import org.dna.mqtt.moquette.proto.ConnectDecoder;
import org.dna.mqtt.moquette.proto.DisconnectDecoder;
import org.dna.mqtt.moquette.proto.MQTTLoggingFilter;
import org.dna.mqtt.moquette.proto.PingReqDecoder;
import org.dna.mqtt.moquette.proto.PingRespEncoder;
import org.dna.mqtt.moquette.proto.PubAckDecoder;
import org.dna.mqtt.moquette.proto.PubAckEncoder;
import org.dna.mqtt.moquette.proto.PubCompDecoder;
import org.dna.mqtt.moquette.proto.PubCompEncoder;
import org.dna.mqtt.moquette.proto.PubRecDecoder;
import org.dna.mqtt.moquette.proto.PubRecEncoder;
import org.dna.mqtt.moquette.proto.PubRelDecoder;
import org.dna.mqtt.moquette.proto.PubRelEncoder;
import org.dna.mqtt.moquette.proto.PublishDecoder;
import org.dna.mqtt.moquette.proto.PublishEncoder;
import org.dna.mqtt.moquette.proto.SubAckEncoder;
import org.dna.mqtt.moquette.proto.SubscribeDecoder;
import org.dna.mqtt.moquette.proto.UnsubAckEncoder;
import org.dna.mqtt.moquette.proto.UnsubscribeDecoder;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
import org.dna.mqtt.moquette.proto.messages.PubRecMessage;
import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
import org.dna.mqtt.moquette.server.ServerAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class MinaAcceptor implements ServerAcceptor {
private static final Logger LOG = LoggerFactory.getLogger(MinaAcceptor.class);
private IoAcceptor m_acceptor;
public void initialize(IMessaging messaging) throws IOException {
DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
decoder.addMessageDecoder(new ConnectDecoder());
decoder.addMessageDecoder(new PublishDecoder());
decoder.addMessageDecoder(new PubAckDecoder());
decoder.addMessageDecoder(new PubRelDecoder());
decoder.addMessageDecoder(new PubRecDecoder());
decoder.addMessageDecoder(new PubCompDecoder());
decoder.addMessageDecoder(new SubscribeDecoder());
decoder.addMessageDecoder(new UnsubscribeDecoder());
decoder.addMessageDecoder(new DisconnectDecoder());
decoder.addMessageDecoder(new PingReqDecoder());
DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
// encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder());
encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder());
encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder());
encoder.addMessageEncoder(PubAckMessage.class, new PubAckEncoder());
encoder.addMessageEncoder(PubRecMessage.class, new PubRecEncoder());
encoder.addMessageEncoder(PubCompMessage.class, new PubCompEncoder());
encoder.addMessageEncoder(PubRelMessage.class, new PubRelEncoder());
encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder());
encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder());
m_acceptor = new NioSocketAcceptor();
m_acceptor.getFilterChain().addLast( "logger", new MQTTLoggingFilter("SERVER LOG") );
m_acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
MinaMQTTHandler handler = new MinaMQTTHandler();
handler.setMessaging(messaging);
m_acceptor.setHandler(handler);
((NioSocketAcceptor)m_acceptor).setReuseAddress(true);
((NioSocketAcceptor)m_acceptor).getSessionConfig().setReuseAddress(true);
m_acceptor.getSessionConfig().setReadBufferSize( 2048 );
m_acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, Constants.DEFAULT_CONNECT_TIMEOUT );
m_acceptor.getStatistics().setThroughputCalculationInterval(10);
m_acceptor.getStatistics().updateThroughput(System.currentTimeMillis());
m_acceptor.bind( new InetSocketAddress(Constants.PORT) );
LOG.info("Server binded");
}
public void close() {
//log statistics
IoServiceStatistics statistics = m_acceptor.getStatistics();
statistics.updateThroughput(System.currentTimeMillis());
System.out.println(String.format("Total read bytes: %d, read throughtput: %f (b/s)", statistics.getReadBytes(), statistics.getReadBytesThroughput()));
System.out.println(String.format("Total read msgs: %d, read msg throughtput: %f (msg/s)", statistics.getReadMessages(), statistics.getReadMessagesThroughput()));
for(IoSession session: m_acceptor.getManagedSessions().values()) {
if(session.isConnected() && !session.isClosing()){
session.close(false);
}
}
m_acceptor.unbind();
m_acceptor.dispose();
}
}
broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java
-38
View File
@@ -1,38 +0,0 @@
package org.dna.mqtt.moquette.server.mina;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.server.ServerChannel;
/**
*
* @author andrea
*/
public class MinaChannel implements ServerChannel {
private IoSession m_session;
public MinaChannel(IoSession session) {
m_session = session;
}
public Object getAttribute(Object key) {
return m_session.getAttribute(key);
}
public void setAttribute(Object key, Object value) {
m_session.setAttribute(key, value);
}
public void close(boolean immediately) {
m_session.close(immediately);
}
public void write(Object value) {
m_session.write(value);
}
public void setIdleTime(int idleTime) {
m_session.getConfig().setIdleTime(IdleStatus.READER_IDLE, idleTime);
}
}
broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaMQTTHandler.java
-75
View File
@@ -1,75 +0,0 @@
package org.dna.mqtt.moquette.server.mina;
import io.netty.channel.ChannelHandler;
import java.util.HashMap;
import java.util.Map;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.proto.Utils;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* MINA MQTT Handler used to route messages to protocol logic
*
* @author andrea
*/
public class MinaMQTTHandler extends IoHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(MinaMQTTHandler.class);
private IMessaging m_messaging;
private Map<IoSession, MinaChannel> m_channelMapper = new HashMap<IoSession, MinaChannel>();
@Override
public void messageReceived(IoSession session, Object message) throws Exception {
AbstractMessage msg = (AbstractMessage) message;
LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType())));
try {
switch (msg.getMessageType()) {
case CONNECT:
case SUBSCRIBE:
case UNSUBSCRIBE:
case PUBLISH:
case PUBREC:
case PUBCOMP:
case PUBREL:
case DISCONNECT:
MinaChannel channel;
synchronized(m_channelMapper) {
if (!m_channelMapper.containsKey(session)) {
m_channelMapper.put(session, new MinaChannel(session));
}
channel = m_channelMapper.get(session);
}
m_messaging.handleProtocolMessage(channel, msg);
break;
case PINGREQ:
PingRespMessage pingResp = new PingRespMessage();
session.write(pingResp);
break;
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
}
}
@Override
public void sessionIdle(IoSession session, IdleStatus status) {
if (status == IdleStatus.READER_IDLE) {
session.close(false);
m_channelMapper.remove(session);
//TODO send a notification to messaging part to remove the bining clientID-ConnConfig
}
}
public void setMessaging(IMessaging messaging) {
m_messaging = messaging;
}
}
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+232 -235
View File
@@ -1,14 +1,12 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.mina.core.filterchain.IoFilter;
import org.apache.mina.core.filterchain.IoFilterAdapter;
import org.apache.mina.core.session.DummySession;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.WriteRequest;
//import org.apache.mina.core.filterchain.IoFilter;
//import org.apache.mina.core.filterchain.IoFilterAdapter;
//import org.apache.mina.core.session.DummySession;
//import org.apache.mina.core.session.IoSession;
//import org.apache.mina.core.write.WriteRequest;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
@@ -20,9 +18,8 @@ import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.ServerChannel;
import org.dna.mqtt.moquette.server.mina.MinaChannel;
//import org.dna.mqtt.moquette.server.mina.MinaChannel;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@@ -34,231 +31,231 @@ import org.junit.Test;
* @author andrea
*/
public class ProtocolProcessorTest {
final static String FAKE_CLIENT_ID = "FAKE_123";
final static String FAKE_TOPIC = "/news";
ServerChannel m_session;
byte m_returnCode;
ConnectMessage connMsg;
ProtocolProcessor m_processor;
IStorageService m_storageService;
SubscriptionsStore subscriptions;
AbstractMessage m_receivedMessage;
IoSession m_minaSession;
@Before
public void setUp() throws InterruptedException {
connMsg = new ConnectMessage();
connMsg.setProcotolVersion((byte) 0x03);
m_minaSession = new DummySession();
m_session = new MinaChannel(m_minaSession);
m_minaSession.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
@Override
public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,
WriteRequest writeRequest) throws Exception {
try {
m_receivedMessage = (AbstractMessage) writeRequest.getMessage();
if (m_receivedMessage instanceof ConnAckMessage) {
ConnAckMessage buf = (ConnAckMessage) m_receivedMessage;
m_returnCode = buf.getReturnCode();
}
} catch (Exception ex) {
throw new AssertionError("Wrong return code");
}
}
});
//sleep to let the messaging batch processor to process the initEvent
Thread.sleep(300);
m_storageService = new MemoryStorageService();
//m_storageService.initStore();
subscriptions = new SubscriptionsStore();
subscriptions.init(m_storageService);
m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_storageService);
}
@Test
public void testHandleConnect_BadProtocol() {
connMsg.setProcotolVersion((byte) 0x02);
//Exercise
m_processor.processConnect(m_session, connMsg);
//Verify
assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode);
}
@Test
public void testConnect_badClientID() {
connMsg.setClientID("extremely_long_clientID_greater_than_23");
//Exercise
m_processor.processConnect(m_session, connMsg);
//Verify
assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode);
}
@Test
public void testWill() {
connMsg.setClientID("123");
connMsg.setWillFlag(true);
connMsg.setWillTopic("topic");
connMsg.setWillMessage("Topic message");
//Exercise
//m_handler.setMessaging(mockedMessaging);
m_processor.processConnect(m_session, connMsg);
//Verify
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);
//TODO verify the call
/*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/
}
@Test
public void testPublish() throws InterruptedException {
final Subscription subscription = new Subscription(FAKE_CLIENT_ID,
FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true);
//subscriptions.matches(topic) redefine the method to return true
SubscriptionsStore subs = new SubscriptionsStore() {
@Override
public List<Subscription> matches(String topic) {
if (topic.equals(FAKE_TOPIC)) {
return Arrays.asList(subscription);
} else {
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
}
}
};
//simulate a connect that register a clientID to an IoSession
subs.init(m_storageService);
m_processor.init(subs, m_storageService);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProcotolVersion((byte)3);
connectMessage.setClientID(FAKE_CLIENT_ID);
connectMessage.setCleanSession(subscription.isCleanSession());
m_processor.processConnect(m_session, connectMessage);
//Exercise
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
m_processor.processPublish(pubEvt);
//Verify
assertNotNull(m_receivedMessage);
//TODO check received message attributes
}
@Test
public void testSubscribe() {
//Exercise
SubscribeMessage msg = new SubscribeMessage();
msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//Verify
assertTrue(m_receivedMessage instanceof SubAckMessage);
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false);
assertTrue(subscriptions.contains(expectedSubscription));
}
@Test
public void testDoubleSubscribe() {
SubscribeMessage msg = new SubscribeMessage();
msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
subscriptions.clearAllSubscriptions();
assertEquals(0, subscriptions.size());
m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//Exercise
m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//Verify
assertEquals(1, subscriptions.size());
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false);
assertTrue(subscriptions.contains(expectedSubscription));
}
@Test
public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception {
m_minaSession.getFilterChain().remove("MessageCatcher");
m_minaSession.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
@Override
public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,
WriteRequest writeRequest) throws Exception {
try {
System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName());
if (writeRequest.getMessage() instanceof PublishMessage) {
m_receivedMessage = (AbstractMessage) writeRequest.getMessage();
}
if (m_receivedMessage instanceof ConnAckMessage) {
ConnAckMessage buf = (ConnAckMessage) m_receivedMessage;
m_returnCode = buf.getReturnCode();
}
} catch (Exception ex) {
throw new AssertionError("Wrong return code");
}
}
});
// final static String FAKE_CLIENT_ID = "FAKE_123";
// final static String FAKE_TOPIC = "/news";
//
// ServerChannel m_session;
// byte m_returnCode;
// ConnectMessage connMsg;
// ProtocolProcessor m_processor;
//
// IStorageService m_storageService;
// SubscriptionsStore subscriptions;
// AbstractMessage m_receivedMessage;
//
// IoSession m_minaSession;
//
//
// @Before
// public void setUp() throws InterruptedException {
// connMsg = new ConnectMessage();
// connMsg.setProcotolVersion((byte) 0x03);
//
// m_minaSession = new DummySession();
//
// m_session = new MinaChannel(m_minaSession);
//
// m_minaSession.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
//
// @Override
// public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,
// WriteRequest writeRequest) throws Exception {
// try {
// m_receivedMessage = (AbstractMessage) writeRequest.getMessage();
// if (m_receivedMessage instanceof ConnAckMessage) {
// ConnAckMessage buf = (ConnAckMessage) m_receivedMessage;
// m_returnCode = buf.getReturnCode();
// }
// } catch (Exception ex) {
// throw new AssertionError("Wrong return code");
// }
// }
// });
//
// //sleep to let the messaging batch processor to process the initEvent
// Thread.sleep(300);
//
// m_storageService = new MemoryStorageService();
// //m_storageService.initStore();
//
// subscriptions = new SubscriptionsStore();
// subscriptions.init(m_storageService);
// m_processor = new ProtocolProcessor();
// m_processor.init(subscriptions, m_storageService);
// }
//
// @Test
// public void testHandleConnect_BadProtocol() {
// connMsg.setProcotolVersion((byte) 0x02);
//
// //Exercise
// m_processor.processConnect(m_session, connMsg);
//
// //Verify
// assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode);
// }
//
// @Test
// public void testConnect_badClientID() {
// connMsg.setClientID("extremely_long_clientID_greater_than_23");
//
// //Exercise
// m_processor.processConnect(m_session, connMsg);
//
// //Verify
// assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, m_returnCode);
// }
//
// @Test
// public void testWill() {
// connMsg.setClientID("123");
// connMsg.setWillFlag(true);
// connMsg.setWillTopic("topic");
// connMsg.setWillMessage("Topic message");
//
//
// //Exercise
// //m_handler.setMessaging(mockedMessaging);
// m_processor.processConnect(m_session, connMsg);
//
// //Verify
// assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode);
// //TODO verify the call
// /*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
// any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/
// }
//
// @Test
// public void testPublish() throws InterruptedException {
// final Subscription subscription = new Subscription(FAKE_CLIENT_ID,
// FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true);
//
// //subscriptions.matches(topic) redefine the method to return true
// SubscriptionsStore subs = new SubscriptionsStore() {
// @Override
// public List<Subscription> matches(String topic) {
// if (topic.equals(FAKE_TOPIC)) {
// return Arrays.asList(subscription);
// } else {
// throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
// }
// }
//
//
// };
//
// //simulate a connect that register a clientID to an IoSession
final Subscription subscription = new Subscription(FAKE_CLIENT_ID,
FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true);
//subscriptions.matches(topic) redefine the method to return true
SubscriptionsStore subs = new SubscriptionsStore() {
@Override
public List<Subscription> matches(String topic) {
if (topic.equals(FAKE_TOPIC)) {
return Arrays.asList(subscription);
} else {
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
}
}
};
subs.init(new MemoryStorageService());
//simulate a connect that register a clientID to an IoSession
m_processor.init(subs, m_storageService);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setClientID(FAKE_CLIENT_ID);
connectMessage.setProcotolVersion((byte)3);
connectMessage.setCleanSession(subscription.isCleanSession());
m_processor.processConnect(m_session, connectMessage);
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null);
m_processor.processPublish(pubEvt);
//Exercise
SubscribeMessage msg = new SubscribeMessage();
msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#"));
m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//Verify
assertNotNull(m_receivedMessage);
assertTrue(m_receivedMessage instanceof PublishMessage);
PublishMessage pubMessage = (PublishMessage) m_receivedMessage;
assertEquals(FAKE_TOPIC, pubMessage.getTopicName());
}
// subs.init(m_storageService);
// m_processor.init(subs, m_storageService);
// ConnectMessage connectMessage = new ConnectMessage();
// connectMessage.setProcotolVersion((byte)3);
// connectMessage.setClientID(FAKE_CLIENT_ID);
// connectMessage.setCleanSession(subscription.isCleanSession());
// m_processor.processConnect(m_session, connectMessage);
//
//
// //Exercise
// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
// m_processor.processPublish(pubEvt);
//
// //Verify
// assertNotNull(m_receivedMessage);
// //TODO check received message attributes
// }
//
// @Test
// public void testSubscribe() {
// //Exercise
// SubscribeMessage msg = new SubscribeMessage();
// msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
// m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//
// //Verify
// assertTrue(m_receivedMessage instanceof SubAckMessage);
// Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false);
// assertTrue(subscriptions.contains(expectedSubscription));
// }
//
// @Test
// public void testDoubleSubscribe() {
// SubscribeMessage msg = new SubscribeMessage();
// msg.addSubscription(new SubscribeMessage.Couple((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
//
// subscriptions.clearAllSubscriptions();
// assertEquals(0, subscriptions.size());
//
// m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//
// //Exercise
// m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//
// //Verify
// assertEquals(1, subscriptions.size());
// Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false);
// assertTrue(subscriptions.contains(expectedSubscription));
// }
//
//
// @Test
// public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception {
// m_minaSession.getFilterChain().remove("MessageCatcher");
// m_minaSession.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
//
// @Override
// public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,
// WriteRequest writeRequest) throws Exception {
// try {
// System.out.println("filterReceived class " + writeRequest.getMessage().getClass().getName());
// if (writeRequest.getMessage() instanceof PublishMessage) {
// m_receivedMessage = (AbstractMessage) writeRequest.getMessage();
// }
//
// if (m_receivedMessage instanceof ConnAckMessage) {
// ConnAckMessage buf = (ConnAckMessage) m_receivedMessage;
// m_returnCode = buf.getReturnCode();
// }
// } catch (Exception ex) {
// throw new AssertionError("Wrong return code");
// }
// }
// });
//
//// //simulate a connect that register a clientID to an IoSession
// final Subscription subscription = new Subscription(FAKE_CLIENT_ID,
// FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true);
//
// //subscriptions.matches(topic) redefine the method to return true
// SubscriptionsStore subs = new SubscriptionsStore() {
// @Override
// public List<Subscription> matches(String topic) {
// if (topic.equals(FAKE_TOPIC)) {
// return Arrays.asList(subscription);
// } else {
// throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
// }
// }
// };
// subs.init(new MemoryStorageService());
//
// //simulate a connect that register a clientID to an IoSession
// m_processor.init(subs, m_storageService);
// ConnectMessage connectMessage = new ConnectMessage();
// connectMessage.setClientID(FAKE_CLIENT_ID);
// connectMessage.setProcotolVersion((byte)3);
// connectMessage.setCleanSession(subscription.isCleanSession());
// m_processor.processConnect(m_session, connectMessage);
// PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, "Hello".getBytes(), true, "FakeCLI", null);
// m_processor.processPublish(pubEvt);
//
// //Exercise
// SubscribeMessage msg = new SubscribeMessage();
// msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), "#"));
// m_processor.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//
// //Verify
// assertNotNull(m_receivedMessage);
// assertTrue(m_receivedMessage instanceof PublishMessage);
// PublishMessage pubMessage = (PublishMessage) m_receivedMessage;
// assertEquals(FAKE_TOPIC, pubMessage.getTopicName());
// }
}
parser/src/main/java/org/dna/mqtt/moquette/proto/Utils.java
-191
View File
@@ -1,191 +0,0 @@
package org.dna.mqtt.moquette.proto;
import java.io.UnsupportedEncodingException;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.slf4j.LoggerFactory;
/**
* Common utils methodd used in codecs.
*
* @author andrea
*/
public class Utils {
public static final int MAX_LENGTH_LIMIT = 268435455;
/**
* Read 2 bytes from in buffer first MSB, and then LSB returning as int.
*/
static int readWord(IoBuffer in) {
int msb = in.get() & 0x00FF; //remove sign extension due to casting
int lsb = in.get() & 0x00FF;
msb = (msb << 8) | lsb ;
return msb;
}
/**
* Writes as 2 bytes the int value into buffer first MSB, and then LSB.
*/
static void writeWord(IoBuffer out, int value) {
out.put((byte) ((value & 0xFF00) >> 8)); //msb
out.put((byte) (value & 0x00FF)); //lsb
}
/**
* Decode the variable remaining lenght as defined in MQTT v3.1 specification
* (section 2.1).
*
* @return the decoded length or -1 if needed more data to decode the length field.
*/
static int decodeRemainingLenght(IoBuffer in) {
int multiplier = 1;
int value = 0;
byte digit;
do {
if (in.remaining() < 1) {
return -1;
}
digit = in.get();
value += (digit & 0x7F) * multiplier;
multiplier *= 128;
} while ((digit & 0x80) != 0);
return value;
}
/**
* Return the number of bytes to encode the gicen remaining length value
*/
static int numBytesToEncode(int len) {
if (0 <= len && len <= 127) return 1;
if (128 <= len && len <= 16383) return 2;
if (16384 <= len && len <= 2097151) return 3;
if (2097152 <= len && len <= 268435455) return 4;
throw new IllegalArgumentException("value shoul be in the range [0..268435455]");
}
/**
* Encode the value in the format defined in specification as variable length
* array.
*
* @throws IllegalArgumentException if the value is not in the specification bounds
* [0..268435455].
*/
static IoBuffer encodeRemainingLength(int value) throws IllegalAccessException {
if (value > MAX_LENGTH_LIMIT || value < 0) {
throw new IllegalAccessException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value);
}
IoBuffer encoded = IoBuffer.allocate(4);
byte digit;
do {
digit = (byte) (value % 128);
value = value / 128;
// if there are more digits to encode, set the top bit of this digit
if (value > 0) {
digit = (byte) (digit | 0x80);
}
encoded.put(digit);
} while (value > 0);
encoded.flip();
return encoded;
}
static MessageDecoderResult checkDecodable(byte type, IoBuffer in) {
if (in.remaining() < 1) {
return MessageDecoderResult.NEED_DATA;
}
byte h1 = in.get();
byte messageType = (byte) ((h1 & 0x00F0) >> 4);
int remainingLength = Utils.decodeRemainingLenght(in);
if (remainingLength == -1) {
return MessageDecoderResult.NEED_DATA;
}
//check remaining length
if (in.remaining() < remainingLength) {
return MessageDecoderResult.NEED_DATA;
}
return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK;
}
/**
* Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded
* string content.
*/
static IoBuffer encodeString(String str) {
IoBuffer out = IoBuffer.allocate(2).setAutoExpand(true);
byte[] raw;
try {
raw = str.getBytes("UTF-8");
//NB every Java platform has got UTF-8 encoding by default, so this
//exception are never raised.
} catch (UnsupportedEncodingException ex) {
LoggerFactory.getLogger(ConnectEncoder.class).error(null, ex);
return null;
}
Utils.writeWord(out, raw.length);
out.put(raw).flip();
return out;
}
/**
* Load a string from the given buffer, reading first the two bytes of len
* and then the UTF-8 bytes of the string.
*
* @return the decoded string or null if NEED_DATA
*/
static String decodeString(IoBuffer in) throws UnsupportedEncodingException {
if (in.remaining() < 2) {
return null;
}
int strLen = Utils.readWord(in);
if (in.remaining() < strLen) {
return null;
}
byte[] strRaw = new byte[strLen];
in.get(strRaw);
return new String(strRaw, "UTF-8");
}
static byte encodeFlags(AbstractMessage message) {
byte flags = 0;
if (message.isDupFlag()) {
flags |= 0x08;
}
if (message.isRetainFlag()) {
flags |= 0x01;
}
flags |= ((message.getQos().ordinal() & 0x03) << 1);
return flags;
}
/**
* Converts MQTT message type to a textual description.
* */
public static String msgType2String(int type) {
switch (type) {
case AbstractMessage.CONNECT: return "CONNECT";
case AbstractMessage.CONNACK: return "CONNACK";
case AbstractMessage.PUBLISH: return "PUBLISH";
case AbstractMessage.PUBACK: return "PUBACK";
case AbstractMessage.PUBREC: return "PUBREC";
case AbstractMessage.PUBREL: return "PUBREL";
case AbstractMessage.PUBCOMP: return "PUBCOMP";
case AbstractMessage.SUBSCRIBE: return "SUBSCRIBE";
case AbstractMessage.SUBACK: return "SUBACK";
case AbstractMessage.UNSUBSCRIBE: return "UNSUBSCRIBE";
case AbstractMessage.UNSUBACK: return "UNSUBACK";
case AbstractMessage.PINGREQ: return "PINGREQ";
case AbstractMessage.PINGRESP: return "PINGRESP";
case AbstractMessage.DISCONNECT: return "DISCONNECT";
default: throw new RuntimeException("Can't decode message type " + type);
}
}
}
parser/src/main/java/org/dna/mqtt/commons/Constants.java → parser_commons/src/main/java/org/dna/mqtt/commons/Constants.java
+1 -1
View File
@@ -1,7 +1,7 @@
package org.dna.mqtt.commons;
/**
* Contains some usefull constants.
* Contains some useful constants.
*/
public class Constants {
public static final int PORT = 1883;
parser/src/main/java/org/dna/mqtt/moquette/MQTTException.java → parser_commons/src/main/java/org/dna/mqtt/moquette/MQTTException.java
View File
parser_commons/src/main/java/org/dna/mqtt/moquette/proto/Utils.java
+191
View File
@@ -0,0 +1,191 @@
package org.dna.mqtt.moquette.proto;
import java.io.UnsupportedEncodingException;
//import org.apache.mina.core.buffer.IoBuffer;
//import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.slf4j.LoggerFactory;
/**
* Common utils methodd used in codecs.
*
* @author andrea
*/
public class Utils {
public static final int MAX_LENGTH_LIMIT = 268435455;
// /**
// * Read 2 bytes from in buffer first MSB, and then LSB returning as int.
// */
// static int readWord(IoBuffer in) {
// int msb = in.get() & 0x00FF; //remove sign extension due to casting
// int lsb = in.get() & 0x00FF;
// msb = (msb << 8) | lsb ;
// return msb;
// }
//
// /**
// * Writes as 2 bytes the int value into buffer first MSB, and then LSB.
// */
// static void writeWord(IoBuffer out, int value) {
// out.put((byte) ((value & 0xFF00) >> 8)); //msb
// out.put((byte) (value & 0x00FF)); //lsb
// }
//
// /**
// * Decode the variable remaining lenght as defined in MQTT v3.1 specification
// * (section 2.1).
// *
// * @return the decoded length or -1 if needed more data to decode the length field.
// */
// static int decodeRemainingLenght(IoBuffer in) {
// int multiplier = 1;
// int value = 0;
// byte digit;
// do {
// if (in.remaining() < 1) {
// return -1;
// }
// digit = in.get();
// value += (digit & 0x7F) * multiplier;
// multiplier *= 128;
// } while ((digit & 0x80) != 0);
// return value;
// }
/**
* Return the number of bytes to encode the gicen remaining length value
*/
static int numBytesToEncode(int len) {
if (0 <= len && len <= 127) return 1;
if (128 <= len && len <= 16383) return 2;
if (16384 <= len && len <= 2097151) return 3;
if (2097152 <= len && len <= 268435455) return 4;
throw new IllegalArgumentException("value shoul be in the range [0..268435455]");
}
// /**
// * Encode the value in the format defined in specification as variable length
// * array.
// *
// * @throws IllegalArgumentException if the value is not in the specification bounds
// * [0..268435455].
// */
// static IoBuffer encodeRemainingLength(int value) throws IllegalAccessException {
// if (value > MAX_LENGTH_LIMIT || value < 0) {
// throw new IllegalAccessException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value);
// }
//
// IoBuffer encoded = IoBuffer.allocate(4);
// byte digit;
// do {
// digit = (byte) (value % 128);
// value = value / 128;
// // if there are more digits to encode, set the top bit of this digit
// if (value > 0) {
// digit = (byte) (digit | 0x80);
// }
// encoded.put(digit);
// } while (value > 0);
// encoded.flip();
// return encoded;
// }
//
// static MessageDecoderResult checkDecodable(byte type, IoBuffer in) {
// if (in.remaining() < 1) {
// return MessageDecoderResult.NEED_DATA;
// }
// byte h1 = in.get();
// byte messageType = (byte) ((h1 & 0x00F0) >> 4);
//
// int remainingLength = Utils.decodeRemainingLenght(in);
// if (remainingLength == -1) {
// return MessageDecoderResult.NEED_DATA;
// }
//
// //check remaining length
// if (in.remaining() < remainingLength) {
// return MessageDecoderResult.NEED_DATA;
// }
//
// return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK;
// }
//
// /**
// * Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded
// * string content.
// */
// static IoBuffer encodeString(String str) {
// IoBuffer out = IoBuffer.allocate(2).setAutoExpand(true);
// byte[] raw;
// try {
// raw = str.getBytes("UTF-8");
// //NB every Java platform has got UTF-8 encoding by default, so this
// //exception are never raised.
// } catch (UnsupportedEncodingException ex) {
// LoggerFactory.getLogger(ConnectEncoder.class).error(null, ex);
// return null;
// }
// Utils.writeWord(out, raw.length);
// out.put(raw).flip();
// return out;
// }
//
//
// /**
// * Load a string from the given buffer, reading first the two bytes of len
// * and then the UTF-8 bytes of the string.
// *
// * @return the decoded string or null if NEED_DATA
// */
// static String decodeString(IoBuffer in) throws UnsupportedEncodingException {
// if (in.remaining() < 2) {
// return null;
// }
// int strLen = Utils.readWord(in);
// if (in.remaining() < strLen) {
// return null;
// }
// byte[] strRaw = new byte[strLen];
// in.get(strRaw);
//
// return new String(strRaw, "UTF-8");
// }
static byte encodeFlags(AbstractMessage message) {
byte flags = 0;
if (message.isDupFlag()) {
flags |= 0x08;
}
if (message.isRetainFlag()) {
flags |= 0x01;
}
flags |= ((message.getQos().ordinal() & 0x03) << 1);
return flags;
}
/**
* Converts MQTT message type to a textual description.
* */
public static String msgType2String(int type) {
switch (type) {
case AbstractMessage.CONNECT: return "CONNECT";
case AbstractMessage.CONNACK: return "CONNACK";
case AbstractMessage.PUBLISH: return "PUBLISH";
case AbstractMessage.PUBACK: return "PUBACK";
case AbstractMessage.PUBREC: return "PUBREC";
case AbstractMessage.PUBREL: return "PUBREL";
case AbstractMessage.PUBCOMP: return "PUBCOMP";
case AbstractMessage.SUBSCRIBE: return "SUBSCRIBE";
case AbstractMessage.SUBACK: return "SUBACK";
case AbstractMessage.UNSUBSCRIBE: return "UNSUBSCRIBE";
case AbstractMessage.UNSUBACK: return "UNSUBACK";
case AbstractMessage.PINGREQ: return "PINGREQ";
case AbstractMessage.PINGRESP: return "PINGRESP";
case AbstractMessage.DISCONNECT: return "DISCONNECT";
default: throw new RuntimeException("Can't decode message type " + type);
}
}
}
pom.xml
-2
View File
@@ -17,10 +17,8 @@
<modules>
<module>parser_commons</module>
<module>parser</module>
<module>netty_parser</module>
<module>broker</module>
<module>benchmarking</module>
</modules>