diff --git a/.gitignore b/.gitignore
index b00a78f7..2bbda7ed 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,3 +5,4 @@
*.iml
broker/target/
client/target/
+/parser_commons/target/
\ No newline at end of file
diff --git a/benchmarking/pom.xml b/benchmarking/pom.xml
index d3947ca0..8cfb0af2 100644
--- a/benchmarking/pom.xml
+++ b/benchmarking/pom.xml
@@ -6,7 +6,7 @@
../pom.xml
moquette-parent
org.dna.mqtt
- 0.1
+ 0.2
moquette-benchmark
@@ -40,8 +40,8 @@
org.dna.mqtt
- moquette-parser
- 0.1
+ moquette-mina-parser
+ 0.2
diff --git a/broker/pom.xml b/broker/pom.xml
index 0b5cdc10..59744733 100644
--- a/broker/pom.xml
+++ b/broker/pom.xml
@@ -21,7 +21,13 @@
org.dna.mqtt
- moquette-parser
+ moquette-mina-parser
+ 0.2
+
+
+
+ org.dna.mqtt
+ moquette-netty-parser
0.2
@@ -79,13 +85,6 @@
test
-
- org.dna.mqtt
- moquette-client
- 0.1
- test
-
-
org.fusesource.hawtbuf
hawtbuf
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
index 379ebe53..8fbd48a1 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
@@ -1,7 +1,7 @@
package org.dna.mqtt.moquette.messaging.spi;
-import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
+import org.dna.mqtt.moquette.server.ServerChannel;
/**
* Interface to the underling messaging system used to publish, subscribe.
@@ -12,7 +12,7 @@ public interface IMessaging {
void stop();
- void disconnect(IoSession session);
+ void disconnect(ServerChannel session);
- void handleProtocolMessage(IoSession session, AbstractMessage msg);
+ void handleProtocolMessage(ServerChannel session, AbstractMessage msg);
}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
index 2ea592eb..dd4053a2 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
@@ -3,15 +3,13 @@ package org.dna.mqtt.moquette.messaging.spi.impl;
import java.util.Collection;
import java.util.List;
import java.util.Map;
-import org.apache.mina.core.session.IdleStatus;
-import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
-import org.dna.mqtt.moquette.proto.PubCompMessage;
+import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
@@ -25,6 +23,7 @@ import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.IAuthenticator;
+import org.dna.mqtt.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,7 +55,7 @@ class ProtocolProcessor {
m_storageService = storageService;
}
- void processConnect(IoSession session, ConnectMessage msg) {
+ void processConnect(ServerChannel session, ConnectMessage msg) {
if (msg.getProcotolVersion() != 0x03) {
ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
@@ -75,7 +74,7 @@ class ProtocolProcessor {
//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
- IoSession oldSession = m_clientIDs.get(msg.getClientID()).getSession();
+ ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).getSession();
boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
@@ -90,12 +89,12 @@ class ProtocolProcessor {
int keepAlive = msg.getKeepAlive();
LOG.debug(String.format("Connect with keepAlive %d s", keepAlive));
- session.setAttribute("keepAlive", 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.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f));
+ session.setIdleTime(Math.round(keepAlive * 1.5f));
//Handle will flag
if (msg.isWillFlag()) {
@@ -187,6 +186,7 @@ class ProtocolProcessor {
}
m_storageService.cleanInFlight(publishKey);
sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
+ LOG.info("replying with PubAck to MSG ID " + evt.getMessageID());
}
if (retain) {
@@ -331,7 +331,7 @@ class ProtocolProcessor {
m_storageService.cleanInFlight(publishKey);
}
- void processDisconnect(IoSession session, String clientID, boolean cleanSession) throws InterruptedException {
+ void processDisconnect(ServerChannel session, String clientID, boolean cleanSession) throws InterruptedException {
if (cleanSession) {
//cleanup topic subscriptions
processRemoveAllSubscriptions(clientID);
@@ -349,7 +349,7 @@ class ProtocolProcessor {
* Remove the clientID from topic subscription, if not previously subscribed,
* doesn't reply any error
*/
- void processUnsubscribe(IoSession session, String clientID, List topics, int messageID) {
+ void processUnsubscribe(ServerChannel session, String clientID, List topics, int messageID) {
LOG.debug("processSubscribe invoked");
for (String topic : topics) {
@@ -364,7 +364,7 @@ class ProtocolProcessor {
}
- void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) {
+ void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) {
LOG.debug("processSubscribe invoked");
for (SubscribeMessage.Couple req : msg.subscriptions()) {
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
index e00a772f..7d42c95c 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
@@ -5,20 +5,19 @@ import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
-import org.dna.mqtt.moquette.proto.PubCompMessage;
+import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
import org.dna.mqtt.moquette.proto.messages.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.Constants;
+import org.dna.mqtt.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,6 +65,7 @@ public class SimpleMessaging implements IMessaging, EventHandler {
SequenceBarrier barrier = m_ringBuffer.newBarrier();
m_eventProcessor = new BatchEventProcessor(m_ringBuffer, barrier, this);
+ //TODO in a presentation is said to don't do the followinf line!!
m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
m_executor.submit(m_eventProcessor);
@@ -74,6 +74,7 @@ public class SimpleMessaging implements IMessaging, EventHandler {
private void disruptorPublish(MessagingEvent msgEvent) {
+ LOG.info("disruptorPublish publishing event " + msgEvent);
long sequence = m_ringBuffer.next();
ValueEvent event = m_ringBuffer.get(sequence);
@@ -83,11 +84,11 @@ public class SimpleMessaging implements IMessaging, EventHandler {
}
- public void disconnect(IoSession session) {
+ public void disconnect(ServerChannel session) {
disruptorPublish(new DisconnectEvent(session));
}
- public void handleProtocolMessage(IoSession session, AbstractMessage msg) {
+ public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) {
disruptorPublish(new ProtocolEvent(session, msg));
}
@@ -97,7 +98,7 @@ public class SimpleMessaging implements IMessaging, EventHandler {
public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
MessagingEvent evt = t.getEvent();
- LOG.debug("onEvent processing messaging event " + evt);
+ LOG.info("onEvent processing messaging event " + evt);
if (evt instanceof PublishEvent) {
m_processor.processPublish((PublishEvent) evt);
} else if (evt instanceof StopEvent) {
@@ -107,7 +108,7 @@ public class SimpleMessaging implements IMessaging, EventHandler {
String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID);
m_processor.processDisconnect(disEvt.getSession(), clientID, false);
} else if (evt instanceof ProtocolEvent) {
- IoSession session = ((ProtocolEvent) evt).getSession();
+ ServerChannel session = ((ProtocolEvent) evt).getSession();
AbstractMessage message = ((ProtocolEvent) evt).getMessage();
if (message instanceof ConnectMessage) {
m_processor.processConnect(session, (ConnectMessage) message);
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java
index bf8ecd27..f90ac9c7 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/DisconnectEvent.java
@@ -1,6 +1,6 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
-import org.apache.mina.core.session.IoSession;
+import org.dna.mqtt.moquette.server.ServerChannel;
/**
*
@@ -8,13 +8,13 @@ import org.apache.mina.core.session.IoSession;
*/
public class DisconnectEvent extends MessagingEvent {
- IoSession m_session;
+ ServerChannel m_session;
- public DisconnectEvent(IoSession session) {
+ public DisconnectEvent(ServerChannel session) {
m_session = session;
}
- public IoSession getSession() {
+ public ServerChannel getSession() {
return m_session;
}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java
index 7bc6036d..2b0232c4 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ProtocolEvent.java
@@ -1,25 +1,31 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
-import org.apache.mina.core.session.IoSession;
+import org.dna.mqtt.moquette.proto.Utils;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
+import org.dna.mqtt.moquette.server.ServerChannel;
/**
* Event used to carry ProtocolMessages from front handler to event processor
*/
public class ProtocolEvent extends MessagingEvent {
- IoSession m_session;
+ ServerChannel m_session;
AbstractMessage message;
- public ProtocolEvent(IoSession session, AbstractMessage message) {
+ public ProtocolEvent(ServerChannel session, AbstractMessage message) {
this.m_session = session;
this.message = message;
}
- public IoSession getSession() {
+ public ServerChannel getSession() {
return m_session;
}
public AbstractMessage getMessage() {
return message;
}
+
+ @Override
+ public String toString() {
+ return "ProtocolEvent wrapping " + Utils.msgType2String(message.getMessageType());
+ }
}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
index 316c3a65..7e2c2b5b 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
@@ -1,9 +1,9 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
-import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import java.io.Serializable;
+import org.dna.mqtt.moquette.server.ServerChannel;
/**
*
@@ -18,10 +18,10 @@ public class PublishEvent extends MessagingEvent implements Serializable {
//Optional attribute, available only fo QoS 1 and 2
int m_msgID;
- transient IoSession m_session;
+ transient ServerChannel m_session;
public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain,
- String clientID, IoSession session) {
+ String clientID, ServerChannel session) {
m_topic = topic;
m_qos = qos;
m_message = message;
@@ -31,7 +31,7 @@ public class PublishEvent extends MessagingEvent implements Serializable {
}
public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain,
- String clientID, int msgID, IoSession session) {
+ String clientID, int msgID, ServerChannel session) {
this(topic, qos, message, retain, clientID, session);
m_msgID = msgID;
}
@@ -60,7 +60,7 @@ public class PublishEvent extends MessagingEvent implements Serializable {
return m_msgID;
}
- public IoSession getSession() {
+ public ServerChannel getSession() {
return m_session;
}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java
index 326bc196..f2a4634f 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/ConnectionDescriptor.java
@@ -1,7 +1,5 @@
package org.dna.mqtt.moquette.server;
-import org.apache.mina.core.session.IoSession;
-
/**
* Maintains the information of single connection, like ClientID, IoSession,
* and other connection related flags.
@@ -12,10 +10,10 @@ import org.apache.mina.core.session.IoSession;
public class ConnectionDescriptor {
private String m_clientID;
- private IoSession m_session;
+ private ServerChannel m_session;
private boolean m_cleanSession;
- public ConnectionDescriptor(String clientID, IoSession session, boolean cleanSession) {
+ public ConnectionDescriptor(String clientID, ServerChannel session, boolean cleanSession) {
this.m_clientID = clientID;
this.m_session = session;
this.m_cleanSession = cleanSession;
@@ -29,7 +27,7 @@ public class ConnectionDescriptor {
return m_clientID;
}
- public IoSession getSession() {
+ public ServerChannel getSession() {
return m_session;
}
}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Constants.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Constants.java
index 0a2ea7fd..17a691c5 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/server/Constants.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Constants.java
@@ -6,4 +6,5 @@ package org.dna.mqtt.moquette.server;
public class Constants {
public static final String ATTR_CLIENTID = "ClientID";
public static final String CLEAN_SESSION = "cleanSession";
+ public static final String KEEP_ALIVE = "keepAlive";
}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java
index 5c658f28..5a9412bb 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java
@@ -2,19 +2,9 @@ package org.dna.mqtt.moquette.server;
import java.io.File;
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.impl.SimpleMessaging;
-import org.dna.mqtt.moquette.proto.*;
-import org.dna.mqtt.moquette.proto.messages.*;
+import org.dna.mqtt.moquette.server.mina.MinaAcceptor;
+import org.dna.mqtt.moquette.server.netty.NettyAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
@@ -25,21 +15,29 @@ public class Server {
private static final Logger LOG = LoggerFactory.getLogger(Server.class);
-// public static final int PORT = 1883;
-// public static final int DEFAULT_CONNECT_TIMEOUT = 10;
public static final String STORAGE_FILE_PATH = System.getProperty("user.home") +
File.separator + "moquette_store.hawtdb";
-
- //TODO one notifier per outbound queue, else no serialization of message flow is granted!!
- //private static final int NOTIFIER_POOL_SIZE = 1;
- private IoAcceptor m_acceptor;
+ public static String NETTY_NIO_FRAMEWORK = "netty";
+ public static String MINA_NIO_FRAMEWORK = "mina";
+ public static String DEFAULT_NIO_FRAMEWORK = MINA_NIO_FRAMEWORK;
+
+ private ServerAcceptor m_acceptor;
SimpleMessaging messaging;
-// Thread messagingEventLoop;
- //private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/;
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();
+ server.startServer(nioFrameworkType);
//Bind a shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
@@ -51,88 +49,34 @@ public class Server {
}
public void startServer() 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));
-
- MQTTHandler handler = new MQTTHandler();
+ startServer(DEFAULT_NIO_FRAMEWORK);
+ }
+
+ public void startServer(String nioFrameworkType) throws IOException {
+ LOG.info("Starting server with " + nioFrameworkType + " connectors");
messaging = SimpleMessaging.getInstance();
messaging.init();
- //TODO fix this hugly wiring
- handler.setMessaging(messaging);
-// messaging.setNotifier(handler);
-// messagingEventLoop = new Thread(messaging);
-// messagingEventLoop.setName("Event Loop" + System.currentTimeMillis());
-// messagingEventLoop.start();
-
- 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");
-
+ if (nioFrameworkType.equals(MINA_NIO_FRAMEWORK)) {
+ m_acceptor = new MinaAcceptor();
+ } else if (nioFrameworkType.equals(NETTY_NIO_FRAMEWORK)) {
+ m_acceptor = new NettyAcceptor();
+ }
+ m_acceptor.initialize(messaging);
}
public void stopServer() {
LOG.info("Server stopping...");
-
messaging.stop();
- //messaging.stop();
-// messagingEventLoop.interrupt();
-// LOG.info("shutting down evet loop");
-// try {
-// messagingEventLoop.join();
-// } catch (InterruptedException ex) {
-// LOG.error(null, ex);
-// }
-
- /*m_notifierPool.shutdown();*/
-
- //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();
+ m_acceptor.close();
LOG.info("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 [mina|netty]\n");
+ System.out.println("while was invoked with type: " + nioFrameworkType);
+ }
}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java
new file mode 100644
index 00000000..7ce5e549
--- /dev/null
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/ServerAcceptor.java
@@ -0,0 +1,15 @@
+package org.dna.mqtt.moquette.server;
+
+import java.io.IOException;
+import org.dna.mqtt.moquette.messaging.spi.IMessaging;
+
+/**
+ *
+ * @author andrea
+ */
+public interface ServerAcceptor {
+
+ void initialize(IMessaging messaging) throws IOException;
+
+ void close();
+}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/ServerChannel.java b/broker/src/main/java/org/dna/mqtt/moquette/server/ServerChannel.java
new file mode 100644
index 00000000..b741de5f
--- /dev/null
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/ServerChannel.java
@@ -0,0 +1,18 @@
+package org.dna.mqtt.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);
+}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java
new file mode 100644
index 00000000..98be370b
--- /dev/null
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaAcceptor.java
@@ -0,0 +1,121 @@
+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();
+ }
+
+}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java
new file mode 100644
index 00000000..14064ddd
--- /dev/null
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaChannel.java
@@ -0,0 +1,38 @@
+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);
+ }
+}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaMQTTHandler.java
similarity index 67%
rename from broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
rename to broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaMQTTHandler.java
index 507d4e06..bc6d8ad9 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/mina/MinaMQTTHandler.java
@@ -1,5 +1,8 @@
-package org.dna.mqtt.moquette.server;
+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;
@@ -16,10 +19,11 @@ import org.slf4j.LoggerFactory;
*
* @author andrea
*/
-public class MQTTHandler extends IoHandlerAdapter {
+public class MinaMQTTHandler extends IoHandlerAdapter {
- private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
+ private static final Logger LOG = LoggerFactory.getLogger(MinaMQTTHandler.class);
private IMessaging m_messaging;
+ private Map m_channelMapper = new HashMap();
@Override
public void messageReceived(IoSession session, Object message) throws Exception {
@@ -35,7 +39,15 @@ public class MQTTHandler extends IoHandlerAdapter {
case PUBCOMP:
case PUBREL:
case DISCONNECT:
- m_messaging.handleProtocolMessage(session, msg);
+ 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();
@@ -51,6 +63,7 @@ public class MQTTHandler extends IoHandlerAdapter {
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
}
}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java
new file mode 100644
index 00000000..b10e576d
--- /dev/null
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java
@@ -0,0 +1,79 @@
+package org.dna.mqtt.moquette.server.netty;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+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.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+import java.io.IOException;
+import org.dna.mqtt.commons.Constants;
+import org.dna.mqtt.moquette.messaging.spi.IMessaging;
+import org.dna.mqtt.moquette.parser.netty.MQTTDecoder;
+import org.dna.mqtt.moquette.parser.netty.MQTTEncoder;
+import org.dna.mqtt.moquette.server.ServerAcceptor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * @author andrea
+ */
+public class NettyAcceptor implements ServerAcceptor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NettyAcceptor.class);
+
+ EventLoopGroup m_bossGroup;
+ EventLoopGroup m_workerGroup;
+
+
+ public void initialize(IMessaging messaging) throws IOException {
+ m_bossGroup = new NioEventLoopGroup();
+ m_workerGroup = new NioEventLoopGroup();
+
+ final NettyMQTTHandler handler = new NettyMQTTHandler();
+ handler.setMessaging(messaging);
+
+ ServerBootstrap b = new ServerBootstrap();
+ b.group(m_bossGroup, m_workerGroup)
+ .channel(NioServerSocketChannel.class)
+ .childHandler(new ChannelInitializer() {
+ @Override
+ public void initChannel(SocketChannel ch) throws Exception {
+ ChannelPipeline pipeline = ch.pipeline();
+ pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
+ pipeline.addLast("decoder", new MQTTDecoder());
+ pipeline.addLast("encoder", new MQTTEncoder());
+ pipeline.addLast("handler", handler);
+ }
+ })
+ .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(Constants.PORT);
+ LOG.info("Server binded");
+ f.sync();
+ } catch (InterruptedException ex) {
+ LOG.error(null, ex);
+ }
+ }
+
+ 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();
+ }
+
+}
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java
new file mode 100644
index 00000000..e83aa609
--- /dev/null
+++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyChannel.java
@@ -0,0 +1,86 @@
+package org.dna.mqtt.moquette.server.netty;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.timeout.IdleState;
+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.dna.mqtt.moquette.server.Constants;
+import org.dna.mqtt.moquette.server.ServerChannel;
+
+/**
+ *
+ * @author andrea
+ */
+public class NettyChannel implements ServerChannel {
+
+ private static class IdleTimoutHandler extends ChannelDuplexHandler {
+ @Override
+ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+ if (evt instanceof IdleState) {
+ IdleState e = (IdleState) evt;
+ if (e == IdleState.ALL_IDLE) {
+ ctx.close();
+ } /*else if (e.getState() == IdleState.WRITER_IDLE) {
+ ctx.writeAndFlush(new PingMessage());
+ }*/
+ }
+ }
+ }
+
+
+ private ChannelHandlerContext m_channel;
+
+ private Map
+
+ org.dna.mqtt
+ moquette-parser-commons
+ 0.2
+
+
+
+ org.mockito
+ mockito-core
+ 1.9.0
+ test
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-compiler-plugin
+
+ 1.5
+ 1.5
+
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+
+
+ org/dna/mqtt/moquette/proto/TestUtils.java
+ org/dna/mqtt/moquette/proto/TestUtils$*
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java
similarity index 100%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java
similarity index 89%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java
index 9d8d4a7f..8102b534 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java
@@ -14,6 +14,10 @@ public class ConnAckMessage extends AbstractMessage {
public static final byte NOT_AUTHORIZED = 0x05;
private byte m_returnCode;
+
+ public ConnAckMessage() {
+ m_messageType = AbstractMessage.CONNACK;
+ }
public byte getReturnCode() {
return m_returnCode;
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnectMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnectMessage.java
similarity index 97%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnectMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnectMessage.java
index f286d682..123041c7 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnectMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnectMessage.java
@@ -24,6 +24,10 @@ public class ConnectMessage extends AbstractMessage {
String m_clientID;
String m_willtopic;
String m_willMessage;
+
+ public ConnectMessage() {
+ m_messageType = AbstractMessage.CONNECT;
+ }
public boolean isCleanSession() {
return m_cleanSession;
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/DisconnectMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/DisconnectMessage.java
similarity index 65%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/DisconnectMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/DisconnectMessage.java
index 12c4f179..df1c998a 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/DisconnectMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/DisconnectMessage.java
@@ -6,4 +6,8 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class DisconnectMessage extends ZeroLengthMessage {
+
+ public DisconnectMessage() {
+ m_messageType = AbstractMessage.DISCONNECT;
+ }
}
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java
similarity index 100%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PingReqMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PingReqMessage.java
similarity index 66%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PingReqMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PingReqMessage.java
index 4bd1f1ba..672187d0 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PingReqMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PingReqMessage.java
@@ -6,4 +6,8 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class PingReqMessage extends ZeroLengthMessage {
+
+ public PingReqMessage() {
+ m_messageType = AbstractMessage.PINGREQ;
+ }
}
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PingRespMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PingRespMessage.java
similarity index 65%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PingRespMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PingRespMessage.java
index aad82e57..4f876ed5 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PingRespMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PingRespMessage.java
@@ -5,4 +5,8 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class PingRespMessage extends ZeroLengthMessage {
+
+ public PingRespMessage() {
+ m_messageType = AbstractMessage.PINGRESP;
+ }
}
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PubAckMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubAckMessage.java
similarity index 65%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PubAckMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubAckMessage.java
index 5c0a87e7..67448532 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PubAckMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubAckMessage.java
@@ -6,4 +6,8 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class PubAckMessage extends MessageIDMessage {
+
+ public PubAckMessage() {
+ m_messageType = AbstractMessage.PUBACK;
+ }
}
diff --git a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubCompMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubCompMessage.java
new file mode 100644
index 00000000..c28314ba
--- /dev/null
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubCompMessage.java
@@ -0,0 +1,13 @@
+package org.dna.mqtt.moquette.proto.messages;
+
+/**
+ *
+ * @author andrea
+ */
+public class PubCompMessage extends MessageIDMessage {
+
+ public PubCompMessage() {
+ m_messageType = AbstractMessage.PUBCOMP;
+ }
+}
+
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRecMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRecMessage.java
similarity index 67%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRecMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRecMessage.java
index 648e6e42..d2b41123 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRecMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRecMessage.java
@@ -7,4 +7,7 @@ package org.dna.mqtt.moquette.proto.messages;
*/
public class PubRecMessage extends MessageIDMessage {
+ public PubRecMessage() {
+ m_messageType = AbstractMessage.PUBREC;
+ }
}
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRelMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRelMessage.java
similarity index 60%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRelMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRelMessage.java
index aef4a5bd..5bbd29a9 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRelMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRelMessage.java
@@ -5,4 +5,8 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class PubRelMessage extends MessageIDMessage {
+
+ public PubRelMessage() {
+ m_messageType = AbstractMessage.PUBREL;
+ }
}
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java
similarity index 89%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java
index 41742e2a..67a5da3f 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java
@@ -17,6 +17,10 @@ public class PublishMessage extends MessageIDMessage {
public void setMessageID(Integer messageID) {
this.m_messageID = messageID;
}*/
+
+ public PublishMessage() {
+ m_messageType = AbstractMessage.PUBLISH;
+ }
public String getTopicName() {
return m_topicName;
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubAckMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubAckMessage.java
similarity index 81%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubAckMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubAckMessage.java
index debe6247..d4ea6316 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubAckMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubAckMessage.java
@@ -10,6 +10,10 @@ import java.util.List;
public class SubAckMessage extends MessageIDMessage {
List m_types = new ArrayList();
+
+ public SubAckMessage() {
+ m_messageType = AbstractMessage.SUBACK;
+ }
public List types() {
return m_types;
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java
similarity index 94%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java
index d93a52eb..67accee3 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java
@@ -31,6 +31,7 @@ public class SubscribeMessage extends MessageIDMessage {
public SubscribeMessage() {
//Subscribe has always QoS 1
+ m_messageType = AbstractMessage.SUBSCRIBE;
m_qos = AbstractMessage.QOSType.LEAST_ONE;
}
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubAckMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubAckMessage.java
similarity index 59%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubAckMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubAckMessage.java
index e5f92f6d..75bfd98e 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubAckMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubAckMessage.java
@@ -5,5 +5,9 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class UnsubAckMessage extends MessageIDMessage {
+
+ public UnsubAckMessage() {
+ m_messageType = AbstractMessage.UNSUBACK;
+ }
}
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java
similarity index 79%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java
index 1d3912db..a1befe17 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java
+++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java
@@ -9,6 +9,10 @@ import java.util.List;
*/
public class UnsubscribeMessage extends MessageIDMessage {
List m_types = new ArrayList();
+
+ public UnsubscribeMessage() {
+ m_messageType = AbstractMessage.UNSUBSCRIBE;
+ }
public List topics() {
return m_types;
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/ZeroLengthMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ZeroLengthMessage.java
similarity index 100%
rename from parser/src/main/java/org/dna/mqtt/moquette/proto/messages/ZeroLengthMessage.java
rename to parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/ZeroLengthMessage.java
diff --git a/pom.xml b/pom.xml
index 8d25bbba..0c2343a3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -16,7 +16,9 @@
+ parser_commons
parser
+ netty_parser
broker
benchmarking