Merge branch 'porting_mina_2_netty'

This commit is contained in:
selva.andre
2013-09-15 17:19:04 +02:00
105 changed files with 3629 additions and 182 deletions
+1
View File
@@ -5,3 +5,4 @@
*.iml
broker/target/
client/target/
/parser_commons/target/
+3 -3
View File
@@ -6,7 +6,7 @@
<relativePath>../pom.xml</relativePath>
<artifactId>moquette-parent</artifactId>
<groupId>org.dna.mqtt</groupId>
<version>0.1</version>
<version>0.2</version>
</parent>
<artifactId>moquette-benchmark</artifactId>
@@ -40,8 +40,8 @@
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-parser</artifactId>
<version>0.1</version>
<artifactId>moquette-mina-parser</artifactId>
<version>0.2</version>
</dependency>
<dependency>
+7 -8
View File
@@ -21,7 +21,13 @@
<dependencies>
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-parser</artifactId>
<artifactId>moquette-mina-parser</artifactId>
<version>0.2</version>
</dependency>
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-netty-parser</artifactId>
<version>0.2</version>
</dependency>
@@ -79,13 +85,6 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-client</artifactId>
<version>0.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.fusesource.hawtbuf</groupId>
<artifactId>hawtbuf</artifactId>
@@ -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);
}
@@ -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<String> topics, int messageID) {
void processUnsubscribe(ServerChannel session, String clientID, List<String> 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()) {
@@ -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<ValueEvent> {
SequenceBarrier barrier = m_ringBuffer.newBarrier();
m_eventProcessor = new BatchEventProcessor<ValueEvent>(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<ValueEvent> {
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<ValueEvent> {
}
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<ValueEvent> {
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<ValueEvent> {
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);
@@ -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;
}
@@ -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());
}
}
@@ -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;
}
@@ -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;
}
}
@@ -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";
}
@@ -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 <broker.jar> [mina|netty]\n");
System.out.println("while was invoked with type: " + nioFrameworkType);
}
}
@@ -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();
}
@@ -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);
}
@@ -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();
}
}
@@ -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);
}
}
@@ -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<IoSession, MinaChannel> m_channelMapper = new HashMap<IoSession, MinaChannel>();
@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
}
}
@@ -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<SocketChannel>() {
@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();
}
}
@@ -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<Object, AttributeKey<Object>> m_attributesKeys = new HashMap<Object, AttributeKey<Object>>();
private static final AttributeKey<Object> ATTR_KEY_KEEPALIVE = new AttributeKey<Object>(Constants.KEEP_ALIVE);
private static final AttributeKey<Object> ATTR_KEY_CLEANSESSION = new AttributeKey<Object>(Constants.CLEAN_SESSION);
private static final AttributeKey<Object> ATTR_KEY_CLIENTID = new AttributeKey<Object>(Constants.ATTR_CLIENTID);
NettyChannel(ChannelHandlerContext ctx) {
m_channel = ctx;
m_attributesKeys.put(Constants.KEEP_ALIVE, ATTR_KEY_KEEPALIVE);
m_attributesKeys.put(Constants.CLEAN_SESSION, ATTR_KEY_CLEANSESSION);
m_attributesKeys.put(Constants.ATTR_CLIENTID, ATTR_KEY_CLIENTID);
}
public Object getAttribute(Object key) {
Attribute<Object> attr = m_channel.attr(mapKey(key));
return attr.get();
}
public void setAttribute(Object key, Object value) {
Attribute<Object> attr = m_channel.attr(mapKey(key));
attr.set(value);
}
private synchronized AttributeKey<Object> mapKey(Object key) {
if (!m_attributesKeys.containsKey(key)) {
throw new IllegalArgumentException("mapKey can't find a matching AttributeKey for " + key);
}
return m_attributesKeys.get(key);
}
public void setIdleTime(int idleTime) {
if (m_channel.pipeline().names().contains("idleStateHandler")) {
m_channel.pipeline().remove("idleStateHandler");
}
if (m_channel.pipeline().names().contains("idleEventHandler")) {
m_channel.pipeline().remove("idleEventHandler");
}
m_channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
m_channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new IdleTimoutHandler());
}
public void close(boolean immediately) {
m_channel.close();
}
public void write(Object value) {
m_channel.writeAndFlush(value);
}
}
@@ -0,0 +1,72 @@
package org.dna.mqtt.moquette.server.netty;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.util.HashMap;
import java.util.Map;
import org.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;
/**
*
* @author andrea
*/
@Sharable
public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);
private IMessaging m_messaging;
private Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<ChannelHandlerContext, NettyChannel>();
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) {
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:
NettyChannel channel;
synchronized(m_channelMapper) {
if (!m_channelMapper.containsKey(ctx)) {
m_channelMapper.put(ctx, new NettyChannel(ctx));
}
channel = m_channelMapper.get(ctx);
}
m_messaging.handleProtocolMessage(channel, msg);
break;
case PINGREQ:
PingRespMessage pingResp = new PingRespMessage();
ctx.write(pingResp);
break;
}
} catch (Exception ex) {
LOG.error("Bad error in processing the message", ex);
}
}
@Override
public void channelInactive(ChannelHandlerContext ctx)throws Exception {
ctx.close(/*false*/);
synchronized(m_channelMapper) {
m_channelMapper.remove(ctx);
}
}
public void setMessaging(IMessaging messaging) {
m_messaging = messaging;
}
}
@@ -21,6 +21,8 @@ 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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@@ -35,7 +37,7 @@ public class ProtocolProcessorTest {
final static String FAKE_CLIENT_ID = "FAKE_123";
final static String FAKE_TOPIC = "/news";
IoSession m_session;
ServerChannel m_session;
byte m_returnCode;
ConnectMessage connMsg;
ProtocolProcessor m_processor;
@@ -44,15 +46,19 @@ public class ProtocolProcessorTest {
SubscriptionsStore subscriptions;
AbstractMessage m_receivedMessage;
IoSession m_minaSession;
@Before
public void setUp() throws InterruptedException {
connMsg = new ConnectMessage();
connMsg.setProcotolVersion((byte) 0x03);
m_session = new DummySession();
m_minaSession = new DummySession();
m_session = new MinaChannel(m_minaSession);
m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
m_minaSession.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
@Override
public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,
@@ -190,8 +196,8 @@ public class ProtocolProcessorTest {
@Test
public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception {
m_session.getFilterChain().remove("MessageCatcher");
m_session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
m_minaSession.getFilterChain().remove("MessageCatcher");
m_minaSession.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() {
@Override
public void filterWrite(IoFilter.NextFilter nextFilter, IoSession session,
@@ -17,7 +17,7 @@ import static org.junit.Assert.*;
* Test the broker using the widely used FuseSource libraries
* */
public class ServerIntegrationFuseTest {
Server m_server;
MQTT m_mqtt;
BlockingConnection m_connection;
@@ -51,14 +51,15 @@ public class ServerIntegrationFuseTest {
@Test
public void testSubscribe() throws Exception {
System.out.println("*** testSubscribe ***");
m_connection = m_mqtt.blockingConnection();
m_connection.connect();
Topic[] topics = {new Topic("/topic", QoS.AT_MOST_ONCE)};
byte[] futSub = m_connection.subscribe(topics);
m_connection.publish("/topic", "Test my payload".getBytes(), QoS.AT_MOST_ONCE, false);
Message msg = m_connection.receive();
msg.ack();
assertEquals("/topic", msg.getTopic());
@@ -66,6 +67,7 @@ public class ServerIntegrationFuseTest {
@Test
public void testCleanSession_maintainClientSubscriptions() throws Exception {
System.out.println("*** testCleanSession_maintainClientSubscriptions ***");
MQTT localMqtt = new MQTT(m_mqtt);
localMqtt.setCleanSession(false);
BlockingConnection connection = localMqtt.blockingConnection();
@@ -87,6 +89,7 @@ public class ServerIntegrationFuseTest {
@Test
public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() throws Exception {
System.out.println("*** testCleanSession_maintainClientSubscriptions_againstClientDestruction ***");
MQTT localMqtt = new MQTT(m_mqtt);
localMqtt.setCleanSession(false);
BlockingConnection connection = localMqtt.blockingConnection();
@@ -114,6 +117,7 @@ public class ServerIntegrationFuseTest {
*/
@Test
public void testCleanSession_correctlyClientSubscriptions() throws Exception {
System.out.println("*** testCleanSession_correctlyClientSubscriptions ***");
MQTT localMqtt = new MQTT(m_mqtt);
localMqtt.setCleanSession(false);
BlockingConnection connection = localMqtt.blockingConnection();
@@ -134,6 +138,7 @@ public class ServerIntegrationFuseTest {
@Test
public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws Exception {
System.out.println("*** testCleanSession_maintainClientSubscriptions_withServerRestart ***");
MQTT localMqtt = new MQTT(m_mqtt);
localMqtt.setCleanSession(false);
BlockingConnection connection = localMqtt.blockingConnection();
@@ -160,6 +165,7 @@ public class ServerIntegrationFuseTest {
@Test
public void testRetain_maintainMessage_againstClientDestruction() throws Exception {
System.out.println("*** testRetain_maintainMessage_againstClientDestruction ***");
m_connection = m_mqtt.blockingConnection();
m_connection.connect();
m_connection.publish("/topic", "Test my payload".getBytes(), QoS.AT_MOST_ONCE, true);
@@ -176,6 +182,7 @@ public class ServerIntegrationFuseTest {
@Test
public void testUnsubscribe_do_not_notify_anymore_same_session() throws Exception {
System.out.println("*** testUnsubscribe_do_not_notify_anymore_same_session ***");
m_connection = m_mqtt.blockingConnection();
m_connection.connect();
m_connection.subscribe(new Topic[] {new Topic("/topic", QoS.AT_MOST_ONCE)});
@@ -195,6 +202,7 @@ public class ServerIntegrationFuseTest {
@Test
public void testUnsubscribe_do_not_notify_anymore_new_session() throws Exception {
System.out.println("*** testUnsubscribe_do_not_notify_anymore_new_session ***");
m_connection = m_mqtt.blockingConnection();
m_connection.connect();
m_connection.subscribe(new Topic[] {new Topic("/topic", QoS.AT_MOST_ONCE)});
@@ -217,6 +225,7 @@ public class ServerIntegrationFuseTest {
@Test
public void testPublishWithQoS1() throws Exception {
System.out.println("*** testPublishWithQoS1 ***");
m_connection = m_mqtt.blockingConnection();
m_connection.connect();
@@ -236,6 +245,7 @@ public class ServerIntegrationFuseTest {
@Test
public void testPublishWithQoS1_notCleanSession() throws Exception {
System.out.println("*** testPublishWithQoS1_notCleanSession ***");
MQTT mqtt = new MQTT(m_mqtt);
mqtt.setCleanSession(false);
m_connection = mqtt.blockingConnection();
@@ -275,6 +285,7 @@ public class ServerIntegrationFuseTest {
@Test
public void testPublishWithQoS2() throws Exception {
System.out.println("*** testPublishWithQoS2 ***");
m_mqtt.setClientId("TestClient");
m_connection = m_mqtt.blockingConnection();
m_connection.connect();
@@ -295,6 +306,7 @@ public class ServerIntegrationFuseTest {
@Test
public void testPublishReceiveWithQoS2() throws Exception {
System.out.println("*** testPublishReceiveWithQoS2 ***");
m_mqtt.setCleanSession(false);
m_connection = m_mqtt.blockingConnection();
m_connection.connect();
+131
View File
@@ -0,0 +1,131 @@
<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>../pom.xml</relativePath>
<artifactId>moquette-parent</artifactId>
<groupId>org.dna.mqtt</groupId>
<version>0.2</version>
</parent>
<artifactId>moquette-netty-parser</artifactId>
<packaging>jar</packaging>
<version>0.2</version>
<name>Moquette - Netty Parser</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-parser-commons</artifactId>
<version>0.2</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.0.8.Final</version>
</dependency>
<!--dependency>
<groupId>com.googlecode.disruptor</groupId>
<artifactId>disruptor</artifactId>
<version>2.7.1</version>
</dependency -->
<!-- <dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-jdk14</artifactId>
<version>1.6.4</version>
</dependency>-->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.6.4</version>
</dependency>
<!--dependency>
<groupId>org.fusesource.hawtdb</groupId>
<artifactId>hawtdb</artifactId>
<version>1.6</version>
</dependency-->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.8.2</version>
<scope>test</scope>
</dependency>
<!-- needs extra dependencies: objenesis & hamcrest -->
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>1.9.0</version>
<scope>test</scope>
</dependency>
<!--dependency>
<groupId>org.fusesource.mqtt-client</groupId>
<artifactId>mqtt-client</artifactId>
<version>1.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.fusesource.hawtbuf</groupId>
<artifactId>hawtbuf</artifactId>
<version>1.9</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>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<excludes>
<exclude>org/dna/mqtt/moquette/proto/TestUtils.java</exclude>
<exclude>org/dna/mqtt/moquette/proto/TestUtils$*</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<!-- <repository>
<id>Fusesource Releases</id>
<url>http://repo.fusesource.com/nexus/content/repositories/snapshots</url>
</repository>-->
<!-- <repository>
<id>fusesource.snapshots</id>
<name>FuseSource Snapshot Repository</name>
<url>http://repo.fusesource.com/nexus/content/repositories/snapshots</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>-->
</repositories>
</project>
@@ -0,0 +1,31 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
/**
*
* @author andrea
*/
class ConnAckDecoder extends DemuxDecoder {
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
in.resetReaderIndex();
//Common decoding part
ConnAckMessage message = new ConnAckMessage();
if (!decodeCommonHeader(message, in)) {
in.resetReaderIndex();
return;
}
//skip reserved byte
in.skipBytes(1);
//read return code
message.setReturnCode(in.readByte());
out.add(message);
}
}
@@ -0,0 +1,22 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
/**
*
* @author andrea
*/
class ConnAckEncoder extends DemuxEncoder<ConnAckMessage>{
@Override
protected void encode(ChannelHandlerContext chc, ConnAckMessage message, ByteBuf out) {
out.writeByte(AbstractMessage.CONNACK << 4);
out.writeBytes(Utils.encodeRemainingLength(2));
out.writeByte(0);
out.writeByte(message.getReturnCode());
}
}
@@ -0,0 +1,146 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.CorruptedFrameException;
import java.io.UnsupportedEncodingException;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
/**
*
* @author andrea
*/
public class ConnectDecoder extends DemuxDecoder {
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws UnsupportedEncodingException {
in.resetReaderIndex();
//Common decoding part
ConnectMessage message = new ConnectMessage();
if (!decodeCommonHeader(message, in)) {
in.resetReaderIndex();
return;
}
int remainingLength = message.getRemainingLength();
int start = in.readerIndex();
//Connect specific decoding part
//ProtocolName 8 bytes
if (in.readableBytes() < 12) {
in.resetReaderIndex();
return;
}
byte[] encProtoName = new byte[6];
in.skipBytes(2); //size, is 0x06
in.readBytes(encProtoName);
String protoName = new String(encProtoName, "UTF-8");
if (!"MQIsdp".equals(protoName)) {
in.resetReaderIndex();
throw new CorruptedFrameException("Invalid protoName: " + protoName);
}
message.setProtocolName(protoName);
//ProtocolVersion 1 byte (value 0x03)
message.setProcotolVersion(in.readByte());
//Connection flag
byte connFlags = in.readByte();
boolean cleanSession = ((connFlags & 0x02) >> 1) == 1 ? true : false;
boolean willFlag = ((connFlags & 0x04) >> 2) == 1 ? true : false;
byte willQos = (byte) ((connFlags & 0x18) >> 3);
if (willQos > 2) {
in.resetReaderIndex();
throw new CorruptedFrameException("Expected will QoS in range 0..2 but found: " + willQos);
}
boolean willRetain = ((connFlags & 0x20) >> 5) == 1 ? true : false;
boolean passwordFlag = ((connFlags & 0x40) >> 6) == 1 ? true : false;
boolean userFlag = ((connFlags & 0x80) >> 7) == 1 ? true : false;
//a password is true iff user is true.
if (!userFlag && passwordFlag) {
in.resetReaderIndex();
throw new CorruptedFrameException("Expected password flag to true if the user flag is true but was: " + passwordFlag);
}
message.setCleanSession(cleanSession);
message.setWillFlag(willFlag);
message.setWillQos(willQos);
message.setWillRetain(willRetain);
message.setPasswordFlag(passwordFlag);
message.setUserFlag(userFlag);
//Keep Alive timer 2 bytes
//int keepAlive = Utils.readWord(in);
int keepAlive = in.readUnsignedShort();
message.setKeepAlive(keepAlive);
if (remainingLength == 12) {
out.add(message);
return;
}
//Decode the ClientID
String clientID = Utils.decodeString(in);
if (clientID == null) {
in.resetReaderIndex();
return;
}
message.setClientID(clientID);
//Decode willTopic
if (willFlag) {
String willTopic = Utils.decodeString(in);
if (willTopic == null) {
in.resetReaderIndex();
return;
}
message.setWillTopic(willTopic);
}
//Decode willMessage
if (willFlag) {
String willMessage = Utils.decodeString(in);
if (willMessage == null) {
in.resetReaderIndex();
return;
}
message.setWillMessage(willMessage);
}
//Compatibility check wieth v3.0, remaining length has precedence over
//the user and password flags
int readed = in.readerIndex() - start;
if (readed == remainingLength) {
out.add(message);
return;
}
//Decode username
if (userFlag) {
String userName = Utils.decodeString(in);
if (userName == null) {
in.resetReaderIndex();
return;
}
message.setUsername(userName);
}
readed = in.readerIndex() - start;
if (readed == remainingLength) {
out.add(message);
return;
}
//Decode password
if (passwordFlag) {
String password = Utils.decodeString(in);
if (password == null) {
in.resetReaderIndex();
return;
}
message.setPassword(password);
}
out.add(message);
}
}
@@ -0,0 +1,70 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
/**
*
* @author andrea
*/
public class ConnectEncoder extends DemuxEncoder<ConnectMessage> {
@Override
protected void encode(ChannelHandlerContext chc, ConnectMessage message, ByteBuf out) {
ByteBuf staticHeaderBuff = chc.alloc().buffer(12);
staticHeaderBuff.writeBytes(Utils.encodeString("MQIsdp"));
//version
staticHeaderBuff.writeByte(0x03);
//connection flags and Strings
byte connectionFlags = 0;
if (message.isCleanSession()) {
connectionFlags |= 0x02;
}
if (message.isWillFlag()) {
connectionFlags |= 0x04;
}
connectionFlags |= ((message.getWillQos() & 0x03) << 3);
if (message.isWillRetain()) {
connectionFlags |= 0x020;
}
if (message.isPasswordFlag()) {
connectionFlags |= 0x040;
}
if (message.isUserFlag()) {
connectionFlags |= 0x080;
}
staticHeaderBuff.writeByte(connectionFlags);
//Keep alive timer
staticHeaderBuff.writeShort(message.getKeepAlive());
//Variable part
ByteBuf variableHeaderBuff = chc.alloc().buffer(12);
if (message.getClientID() != null) {
variableHeaderBuff.writeBytes(Utils.encodeString(message.getClientID()));
if (message.isWillFlag()) {
variableHeaderBuff.writeBytes(Utils.encodeString(message.getWillTopic()));
variableHeaderBuff.writeBytes(Utils.encodeString(message.getWillMessage()));
}
if (message.isUserFlag() && message.getUsername() != null) {
variableHeaderBuff.writeBytes(Utils.encodeString(message.getUsername()));
if (message.isPasswordFlag() && message.getPassword() != null) {
variableHeaderBuff.writeBytes(Utils.encodeString(message.getPassword()));
}
}
}
int variableHeaderSize = variableHeaderBuff.readableBytes();
ByteBuf buff = chc.alloc().buffer();
buff.writeByte(AbstractMessage.CONNECT << 4);
buff.writeBytes(Utils.encodeRemainingLength(12 + variableHeaderSize));
buff.writeBytes(staticHeaderBuff).writeBytes(variableHeaderBuff);
out.writeBytes(buff);
}
}
@@ -0,0 +1,38 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
/**
*
* @author andrea
*/
abstract class DemuxDecoder {
abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
protected boolean decodeCommonHeader(AbstractMessage message, ByteBuf in) {
//Common decoding part
if (in.readableBytes() < 2) {
return false;
}
byte h1 = in.readByte();
byte messageType = (byte) ((h1 & 0x00F0) >> 4);
boolean dupFlag = ((byte) ((h1 & 0x0008) >> 3) == 1);
byte qosLevel = (byte) ((h1 & 0x0006) >> 1);
boolean retainFlag = ((byte) (h1 & 0x0001) == 1);
int remainingLength = Utils.decodeRemainingLenght(in);
if (remainingLength == -1) {
return false;
}
message.setMessageType(messageType);
message.setDupFlag(dupFlag);
message.setQos(AbstractMessage.QOSType.values()[qosLevel]);
message.setRetainFlag(retainFlag);
message.setRemainingLength(remainingLength);
return true;
}
}
@@ -0,0 +1,13 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
/**
*
* @author andrea
*/
abstract class DemuxEncoder<T extends AbstractMessage> {
abstract protected void encode(ChannelHandlerContext chc, T msg, ByteBuf bb);
}
@@ -0,0 +1,26 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.DisconnectMessage;
/**
*
* @author andrea
*/
class DisconnectDecoder extends DemuxDecoder {
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
//Common decoding part
in.resetReaderIndex();
DisconnectMessage message = new DisconnectMessage();
if (!decodeCommonHeader(message, in)) {
in.resetReaderIndex();
return;
}
out.add(message);
}
}
@@ -0,0 +1,19 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.DisconnectMessage;
/**
*
* @author andrea
*/
public class DisconnectEncoder extends DemuxEncoder<DisconnectMessage> {
@Override
protected void encode(ChannelHandlerContext chc, DisconnectMessage msg, ByteBuf out) {
out.writeByte(AbstractMessage.DISCONNECT << 4).writeByte(0);
}
}
@@ -0,0 +1,54 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.CorruptedFrameException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
/**
*
* @author andrea
*/
public class MQTTDecoder extends ByteToMessageDecoder {
private Map<Byte, DemuxDecoder> m_decoderMap = new HashMap<Byte, DemuxDecoder>();
public MQTTDecoder() {
m_decoderMap.put(AbstractMessage.CONNECT, new ConnectDecoder());
m_decoderMap.put(AbstractMessage.CONNACK, new ConnAckDecoder());
m_decoderMap.put(AbstractMessage.PUBLISH, new PublishDecoder());
m_decoderMap.put(AbstractMessage.PUBACK, new PubAckDecoder());
m_decoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeDecoder());
m_decoderMap.put(AbstractMessage.SUBACK, new SubAckDecoder());
m_decoderMap.put(AbstractMessage.UNSUBSCRIBE, new UnsubscribeDecoder());
m_decoderMap.put(AbstractMessage.DISCONNECT, new DisconnectDecoder());
m_decoderMap.put(AbstractMessage.PINGREQ, new PingReqDecoder());
m_decoderMap.put(AbstractMessage.PINGRESP, new PingRespDecoder());
m_decoderMap.put(AbstractMessage.UNSUBACK, new UnsubAckDecoder());
m_decoderMap.put(AbstractMessage.PUBCOMP, new PubCompDecoder());
m_decoderMap.put(AbstractMessage.PUBREC, new PubRecDecoder());
m_decoderMap.put(AbstractMessage.PUBREL, new PubRelDecoder());
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
in.markReaderIndex();
if (!Utils.checkHeaderAvailability(in)) {
in.resetReaderIndex();
return;
}
in.resetReaderIndex();
byte messageType = Utils.readMessageType(in);
DemuxDecoder decoder = m_decoderMap.get(messageType);
if (decoder == null) {
throw new CorruptedFrameException("Can't find any suitable decoder for message type: " + messageType);
}
decoder.decode(ctx, in, out);
}
}
@@ -0,0 +1,44 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.CorruptedFrameException;
import io.netty.handler.codec.MessageToByteEncoder;
import java.util.HashMap;
import java.util.Map;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
/**
*
* @author andrea
*/
public class MQTTEncoder extends MessageToByteEncoder<AbstractMessage> {
private Map<Byte, DemuxEncoder> m_encoderMap = new HashMap<Byte, DemuxEncoder>();
public MQTTEncoder() {
m_encoderMap.put(AbstractMessage.CONNECT, new ConnectEncoder());
m_encoderMap.put(AbstractMessage.CONNACK, new ConnAckEncoder());
m_encoderMap.put(AbstractMessage.PUBLISH, new PublishEncoder());
m_encoderMap.put(AbstractMessage.PUBACK, new PubAckEncoder());
m_encoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeEncoder());
m_encoderMap.put(AbstractMessage.SUBACK, new SubAckEncoder());
m_encoderMap.put(AbstractMessage.UNSUBSCRIBE, new UnsubscribeEncoder());
m_encoderMap.put(AbstractMessage.DISCONNECT, new DisconnectEncoder());
m_encoderMap.put(AbstractMessage.PINGREQ, new PingReqEncoder());
m_encoderMap.put(AbstractMessage.PINGRESP, new PingRespEncoder());
m_encoderMap.put(AbstractMessage.UNSUBACK, new UnsubAckEncoder());
m_encoderMap.put(AbstractMessage.PUBCOMP, new PubCompEncoder());
m_encoderMap.put(AbstractMessage.PUBREC, new PubRecEncoder());
m_encoderMap.put(AbstractMessage.PUBREL, new PubRelEncoder());
}
@Override
protected void encode(ChannelHandlerContext chc, AbstractMessage msg, ByteBuf bb) throws Exception {
DemuxEncoder encoder = m_encoderMap.get(msg.getMessageType());
if (encoder == null) {
throw new CorruptedFrameException("Can't find any suitable decoder for message type: " + msg.getMessageType());
}
encoder.encode(chc, msg, bb);
}
}
@@ -0,0 +1,31 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.MessageIDMessage;
/**
*
* @author andrea
*/
abstract class MessageIDDecoder extends DemuxDecoder {
protected abstract MessageIDMessage createMessage();
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
in.resetReaderIndex();
//Common decoding part
MessageIDMessage message = createMessage();
if (!decodeCommonHeader(message, in)) {
in.resetReaderIndex();
return;
}
//read messageIDs
message.setMessageID(in.readUnsignedShort());
out.add(message);
}
}
@@ -0,0 +1,26 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.dna.mqtt.moquette.parser.netty.DemuxDecoder;
import org.dna.mqtt.moquette.proto.messages.PingReqMessage;
/**
*
* @author andrea
*/
class PingReqDecoder extends DemuxDecoder {
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
//Common decoding part
in.resetReaderIndex();
PingReqMessage message = new PingReqMessage();
if (!decodeCommonHeader(message, in)) {
in.resetReaderIndex();
return;
}
out.add(message);
}
}
@@ -0,0 +1,19 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.parser.netty.DemuxEncoder;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PingReqMessage;
/**
*
* @author andrea
*/
class PingReqEncoder extends DemuxEncoder<PingReqMessage> {
@Override
protected void encode(ChannelHandlerContext chc, PingReqMessage msg, ByteBuf out) {
out.writeByte(AbstractMessage.PINGREQ << 4).writeByte(0);
}
}
@@ -0,0 +1,25 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
/**
*
* @author andrea
*/
class PingRespDecoder extends DemuxDecoder {
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
//Common decoding part
in.resetReaderIndex();
PingRespMessage message = new PingRespMessage();
if (!decodeCommonHeader(message, in)) {
in.resetReaderIndex();
return;
}
out.add(message);
}
}
@@ -0,0 +1,18 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
/**
*
* @author andrea
*/
class PingRespEncoder extends DemuxEncoder<PingRespMessage> {
@Override
protected void encode(ChannelHandlerContext chc, PingRespMessage msg, ByteBuf out) {
out.writeByte(AbstractMessage.PINGRESP << 4).writeByte(0);
}
}
@@ -0,0 +1,17 @@
package org.dna.mqtt.moquette.parser.netty;
import org.dna.mqtt.moquette.proto.messages.MessageIDMessage;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
/**
*
* @author andrea
*/
class PubAckDecoder extends MessageIDDecoder {
@Override
protected MessageIDMessage createMessage() {
return new PubAckMessage();
}
}
@@ -0,0 +1,23 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
/**
*
* @author andrea
*/
class PubAckEncoder extends DemuxEncoder<PubAckMessage> {
@Override
protected void encode(ChannelHandlerContext chc, PubAckMessage msg, ByteBuf out) {
ByteBuf buff = chc.alloc().buffer(4);
buff.writeByte(AbstractMessage.PUBACK << 4);
buff.writeBytes(Utils.encodeRemainingLength(2));
buff.writeShort(msg.getMessageID());
out.writeBytes(buff);
}
}
@@ -0,0 +1,17 @@
package org.dna.mqtt.moquette.parser.netty;
import org.dna.mqtt.moquette.proto.messages.MessageIDMessage;
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
/**
*
* @author andrea
*/
class PubCompDecoder extends MessageIDDecoder {
@Override
protected MessageIDMessage createMessage() {
return new PubCompMessage();
}
}
@@ -0,0 +1,20 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
/**
*
* @author andrea
*/
class PubCompEncoder extends DemuxEncoder<PubCompMessage> {
@Override
protected void encode(ChannelHandlerContext chc, PubCompMessage msg, ByteBuf out) {
out.writeByte(AbstractMessage.PUBCOMP << 4);
out.writeBytes(Utils.encodeRemainingLength(2));
out.writeShort(msg.getMessageID());
}
}
@@ -0,0 +1,16 @@
package org.dna.mqtt.moquette.parser.netty;
import org.dna.mqtt.moquette.proto.messages.MessageIDMessage;
import org.dna.mqtt.moquette.proto.messages.PubRecMessage;
/**
*
* @author andrea
*/
class PubRecDecoder extends MessageIDDecoder {
@Override
protected MessageIDMessage createMessage() {
return new PubRecMessage();
}
}
@@ -0,0 +1,20 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PubRecMessage;
/**
*
* @author andrea
*/
class PubRecEncoder extends DemuxEncoder<PubRecMessage> {
@Override
protected void encode(ChannelHandlerContext chc, PubRecMessage msg, ByteBuf out) {
out.writeByte(AbstractMessage.PUBREC << 4);
out.writeBytes(Utils.encodeRemainingLength(2));
out.writeShort(msg.getMessageID());
}
}
@@ -0,0 +1,18 @@
package org.dna.mqtt.moquette.parser.netty;
import org.dna.mqtt.moquette.proto.messages.MessageIDMessage;
import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
/**
*
* @author andrea
*/
class PubRelDecoder extends MessageIDDecoder {
@Override
protected MessageIDMessage createMessage() {
return new PubRelMessage();
}
}
@@ -0,0 +1,20 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
/**
*
* @author andrea
*/
class PubRelEncoder extends DemuxEncoder<PubRelMessage> {
@Override
protected void encode(ChannelHandlerContext chc, PubRelMessage msg, ByteBuf out) {
out.writeByte(AbstractMessage.PUBREL << 4);
out.writeBytes(Utils.encodeRemainingLength(2));
out.writeShort(msg.getMessageID());
}
}
@@ -0,0 +1,61 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
class PublishDecoder extends DemuxDecoder {
private static Logger LOG = LoggerFactory.getLogger(PublishDecoder.class);
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
LOG.info("decode invoked with buffer " + in);
in.resetReaderIndex();
int startPos = in.readerIndex();
//Common decoding part
PublishMessage message = new PublishMessage();
if (!decodeCommonHeader(message, in)) {
LOG.info("decode ask for more data after " + in);
in.resetReaderIndex();
return;
}
int remainingLength = message.getRemainingLength();
//Topic name
String topic = Utils.decodeString(in);
if (topic == null) {
in.resetReaderIndex();
return;
}
message.setTopicName(topic);
if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE ||
message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE) {
message.setMessageID(in.readUnsignedShort());
}
int stopPos = in.readerIndex();
//read the payload
int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1);
if (in.readableBytes() < payloadSize) {
in.resetReaderIndex();
return;
}
byte[] b = new byte[payloadSize];
in.readBytes(b);
message.setPayload(b);
out.add(message);
}
}
@@ -0,0 +1,45 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
/**
*
* @author andrea
*/
class PublishEncoder extends DemuxEncoder<PublishMessage> {
@Override
protected void encode(ChannelHandlerContext ctx, PublishMessage message, ByteBuf out) {
if (message.getQos() == AbstractMessage.QOSType.RESERVED) {
throw new IllegalArgumentException("Found a message with RESERVED Qos");
}
if (message.getTopicName() == null || message.getTopicName().isEmpty()) {
throw new IllegalArgumentException("Found a message with empty or null topic name");
}
ByteBuf variableHeaderBuff = ctx.alloc().buffer(2);
variableHeaderBuff.writeBytes(Utils.encodeString(message.getTopicName()));
if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE ||
message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE ) {
if (message.getMessageID() == null) {
throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted");
}
variableHeaderBuff.writeShort(message.getMessageID());
}
variableHeaderBuff.writeBytes(message.getPayload());
int variableHeaderSize = variableHeaderBuff.readableBytes();
byte flags = Utils.encodeFlags(message);
ByteBuf buff = ctx.alloc().buffer(2 + variableHeaderSize);
buff.writeByte(AbstractMessage.PUBLISH << 4 | flags);
buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize));
buff.writeBytes(variableHeaderBuff);
out.writeBytes(buff);
}
}
@@ -0,0 +1,43 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
/**
*
* @author andrea
*/
class SubAckDecoder extends DemuxDecoder {
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
//Common decoding part
in.resetReaderIndex();
SubAckMessage message = new SubAckMessage();
if (!decodeCommonHeader(message, in)) {
in.resetReaderIndex();
return;
}
int remainingLength = message.getRemainingLength();
//MessageID
message.setMessageID(in.readUnsignedShort());
remainingLength -= 2;
//Qos array
if (in.readableBytes() < remainingLength ) {
in.resetReaderIndex();
return;
}
for (int i = 0; i < remainingLength; i++) {
byte qos = in.readByte();
message.addType(AbstractMessage.QOSType.values()[qos]);
}
out.add(message);
}
}
@@ -0,0 +1,36 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
/**
*
* @author andrea
*/
class SubAckEncoder extends DemuxEncoder<SubAckMessage> {
@Override
protected void encode(ChannelHandlerContext chc, SubAckMessage message, ByteBuf out) {
if (message.types().isEmpty()) {
throw new IllegalArgumentException("Found a suback message with empty topics");
}
ByteBuf variableHeaderBuff = chc.alloc().buffer(4);
variableHeaderBuff.writeShort(message.getMessageID());
for (AbstractMessage.QOSType c : message.types()) {
variableHeaderBuff.writeByte(c.ordinal());
}
int variableHeaderSize = variableHeaderBuff.readableBytes();
ByteBuf buff = chc.alloc().buffer(2 + variableHeaderSize);
buff.writeByte(AbstractMessage.SUBACK << 4 );
buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize));
buff.writeBytes(variableHeaderBuff);
out.writeBytes(buff);
}
}
@@ -0,0 +1,53 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.CorruptedFrameException;
import java.io.UnsupportedEncodingException;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
/**
*
* @author andrea
*/
class SubscribeDecoder extends DemuxDecoder {
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
//Common decoding part
SubscribeMessage message = new SubscribeMessage();
in.resetReaderIndex();
if (!decodeCommonHeader(message, in)) {
in.resetReaderIndex();
return;
}
//check qos level
if (message.getQos() != QOSType.LEAST_ONE) {
throw new CorruptedFrameException("Received Subscribe message with QoS other than LEAST_ONE, was: " + message.getQos());
}
int start = in.readerIndex();
//read messageIDs
message.setMessageID(in.readUnsignedShort());
int readed = in.readerIndex() - start;
while (readed < message.getRemainingLength()) {
decodeSubscription(in, message);
readed = in.readerIndex()- start;
}
out.add(message);
}
/**
* Populate the message with couple of Qos, topic
*/
private void decodeSubscription(ByteBuf in, SubscribeMessage message) throws UnsupportedEncodingException {
String topic = Utils.decodeString(in);
byte qos = (byte)(in.readByte() & 0x03);
message.addSubscription(new SubscribeMessage.Couple(qos, topic));
}
}
@@ -0,0 +1,42 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
/**
*
* @author andrea
*/
class SubscribeEncoder extends DemuxEncoder<SubscribeMessage> {
@Override
protected void encode(ChannelHandlerContext chc, SubscribeMessage message, ByteBuf out) {
if (message.subscriptions().isEmpty()) {
throw new IllegalArgumentException("Found a subscribe message with empty topics");
}
if (message.getQos() != AbstractMessage.QOSType.LEAST_ONE) {
throw new IllegalArgumentException("Expected a message with QOS 1, found " + message.getQos());
}
ByteBuf variableHeaderBuff = chc.alloc().buffer(4);
variableHeaderBuff.writeShort(message.getMessageID());
for (SubscribeMessage.Couple c : message.subscriptions()) {
variableHeaderBuff.writeBytes(Utils.encodeString(c.getTopic()));
variableHeaderBuff.writeByte(c.getQos());
}
int variableHeaderSize = variableHeaderBuff.readableBytes();
byte flags = Utils.encodeFlags(message);
ByteBuf buff = chc.alloc().buffer(2 + variableHeaderSize);
buff.writeByte(AbstractMessage.SUBSCRIBE << 4 | flags);
buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize));
buff.writeBytes(variableHeaderBuff);
out.writeBytes(buff);
}
}
@@ -0,0 +1,17 @@
package org.dna.mqtt.moquette.parser.netty;
import org.dna.mqtt.moquette.proto.messages.MessageIDMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
/**
*
* @author andrea
*/
class UnsubAckDecoder extends MessageIDDecoder {
@Override
protected MessageIDMessage createMessage() {
return new UnsubAckMessage();
}
}
@@ -0,0 +1,20 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
/**
*
* @author andrea
*/
class UnsubAckEncoder extends DemuxEncoder<UnsubAckMessage> {
@Override
protected void encode(ChannelHandlerContext chc, UnsubAckMessage msg, ByteBuf out) {
out.writeByte(AbstractMessage.UNSUBACK << 4).
writeBytes(Utils.encodeRemainingLength(2)).
writeShort(msg.getMessageID());
}
}
@@ -0,0 +1,43 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.CorruptedFrameException;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
/**
*
* @author andrea
*/
class UnsubscribeDecoder extends DemuxDecoder {
@Override
void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
//Common decoding part
in.resetReaderIndex();
UnsubscribeMessage message = new UnsubscribeMessage();
if (!decodeCommonHeader(message, in)) {
in.resetReaderIndex();
return;
}
//check qos level
if (message.getQos() != AbstractMessage.QOSType.LEAST_ONE) {
throw new CorruptedFrameException("Found an Usubscribe message with qos other than LEAST_ONE, was: " + message.getQos());
}
int start = in.readerIndex();
//read messageIDs
message.setMessageID(in.readUnsignedShort());
int readed = in.readerIndex()- start;
while (readed < message.getRemainingLength()) {
message.addTopic(Utils.decodeString(in));
readed = in.readerIndex()- start;
}
out.add(message);
}
}
@@ -0,0 +1,42 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
/**
*
* @author andrea
*/
class UnsubscribeEncoder extends DemuxEncoder<UnsubscribeMessage> {
@Override
protected void encode(ChannelHandlerContext chc, UnsubscribeMessage message, ByteBuf out) {
if (message.topics().isEmpty()) {
throw new IllegalArgumentException("Found an unsubscribe message with empty topics");
}
if (message.getQos() != AbstractMessage.QOSType.LEAST_ONE) {
throw new IllegalArgumentException("Expected a message with QOS 1, found " + message.getQos());
}
ByteBuf variableHeaderBuff = chc.alloc().buffer(4);
variableHeaderBuff.writeShort(message.getMessageID());
for (String topic : message.topics()) {
variableHeaderBuff.writeBytes(Utils.encodeString(topic));
}
int variableHeaderSize = variableHeaderBuff.readableBytes();
byte flags = Utils.encodeFlags(message);
ByteBuf buff = chc.alloc().buffer(2 + variableHeaderSize);
buff.writeByte(AbstractMessage.UNSUBSCRIBE << 4 | flags);
buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize));
buff.writeBytes(variableHeaderBuff);
out.writeBytes(buff);
}
}
@@ -0,0 +1,159 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.CorruptedFrameException;
import java.io.UnsupportedEncodingException;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class Utils {
public static final int MAX_LENGTH_LIMIT = 268435455;
static byte readMessageType(ByteBuf in) {
byte h1 = in.readByte();
byte messageType = (byte) ((h1 & 0x00F0) >> 4);
return messageType;
}
static boolean checkHeaderAvailability(ByteBuf in) {
if (in.readableBytes() < 1) {
return false;
}
//byte h1 = in.get();
//byte messageType = (byte) ((h1 & 0x00F0) >> 4);
in.skipBytes(1); //skip the messageType byte
int remainingLength = Utils.decodeRemainingLenght(in);
if (remainingLength == -1) {
return false;
}
//check remaining length
if (in.readableBytes() < remainingLength) {
return false;
}
//return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK;
return true;
}
/**
* Decode the variable remaining length 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(ByteBuf in) {
int multiplier = 1;
int value = 0;
byte digit;
do {
if (in.readableBytes() < 1) {
return -1;
}
digit = in.readByte();
value += (digit & 0x7F) * multiplier;
multiplier *= 128;
} while ((digit & 0x80) != 0);
return value;
}
/**
* 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 ByteBuf encodeRemainingLength(int value) throws CorruptedFrameException {
if (value > MAX_LENGTH_LIMIT || value < 0) {
throw new CorruptedFrameException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value);
}
ByteBuf encoded = Unpooled.buffer(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.writeByte(digit);
} while (value > 0);
return encoded;
}
/**
* 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(ByteBuf in) throws UnsupportedEncodingException {
if (in.readableBytes() < 2) {
return null;
}
//int strLen = Utils.readWord(in);
int strLen = in.readUnsignedShort();
if (in.readableBytes() < strLen) {
return null;
}
byte[] strRaw = new byte[strLen];
in.readBytes(strRaw);
return new String(strRaw, "UTF-8");
}
/**
* Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded
* string content.
*/
static ByteBuf encodeString(String str) {
ByteBuf out = Unpooled.buffer(2);
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(Utils.class).error(null, ex);
return null;
}
//Utils.writeWord(out, raw.length);
out.writeShort(raw.length);
out.writeBytes(raw);
return out;
}
/**
* Return the number of bytes to encode the given 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]");
}
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;
}
}
@@ -0,0 +1,53 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.util.ArrayList;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import org.junit.Before;
import org.junit.Test;
/**
*
* @author andrea
*/
public class ConnAckDecoderTest {
ByteBuf m_buff;
ConnAckDecoder m_msgdec;
@Before
public void setUp() {
m_msgdec = new ConnAckDecoder();
}
@Test
public void testHeader() throws Exception {
m_buff = Unpooled.buffer(14);
initHeader(m_buff);
List<Object> results = new ArrayList<Object >();
//Excercise
m_msgdec.decode(null, m_buff, results);
//Verify
assertFalse(results.isEmpty());
ConnAckMessage message = (ConnAckMessage)results.get(0);
assertNotNull(message);
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, message.getReturnCode());
assertEquals(AbstractMessage.CONNACK, message.getMessageType());
}
private void initHeader(ByteBuf buff) {
buff.clear().writeByte(AbstractMessage.CONNACK << 4).writeByte(2);
//reserved
buff.writeByte((byte)0);
//return code
buff.writeByte(ConnAckMessage.CONNECTION_ACCEPTED);
}
}
@@ -0,0 +1,48 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.*;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
*
* @author andrea
*/
public class ConnAckEncoderTest {
ConnAckEncoder m_encoder = new ConnAckEncoder();
ChannelHandlerContext m_mockedContext;
@Before
public void setUp() {
//mock the ChannelHandlerContext to return an UnpooledAllocator
m_mockedContext = mock(ChannelHandlerContext.class);
ByteBufAllocator allocator = UnpooledByteBufAllocator.DEFAULT;
when(m_mockedContext.alloc()).thenReturn(allocator);
}
@Test
public void testHeaderEncode() throws Exception {
ConnAckMessage msg = new ConnAckMessage();
//Exercise
ByteBuf out = Unpooled.buffer();
//Exercise
m_encoder.encode(m_mockedContext, msg, out);
//Verify
assertEquals(0x20, out.readByte()); //1 byte
assertEquals(0x02, out.readByte()); //2 byte, length
assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, out.skipBytes(1).readByte());
}
}
@@ -0,0 +1,146 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.CorruptedFrameException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import org.junit.Before;
import org.junit.Test;
public class ConnectDecoderTest {
ByteBuf m_buff;
ConnectDecoder m_msgdec;
@Before
public void setUp() {
m_msgdec = new ConnectDecoder();
}
@Test
public void testBaseHeader() throws Exception {
m_buff = Unpooled.buffer(14);
initBaseHeader(m_buff);
List<Object> results = new ArrayList<Object >();
//Excercise
m_msgdec.decode(null, m_buff, results);
//Verify
assertFalse(results.isEmpty());
verifyBaseHeader((ConnectMessage)results.get(0));
}
@Test
public void testBaseHeader_ClientID() throws UnsupportedEncodingException, Exception {
m_buff = Unpooled.buffer(40);
initHeader(m_buff, (byte) 38);
encodeString(m_buff, "ABCDEFGH");
encodeString(m_buff, "Topic");
encodeString(m_buff, "Message");
List<Object> results = new ArrayList<Object >();
//Excercise
m_msgdec.decode(null, m_buff, results);
//Verify
assertFalse(results.isEmpty());
ConnectMessage message = (ConnectMessage)results.get(0);
verifyBaseHeader(message);
assertEquals("ABCDEFGH", message.getClientID());
assertEquals("Topic", message.getWillTopic());
assertEquals("Message", message.getWillMessage());
}
@Test
public void testBaseHeader_extra_with_user_pwd() throws UnsupportedEncodingException, Exception {
m_buff = Unpooled.buffer(55);
initHeader(m_buff, (byte) 53);
encodeString(m_buff, "ABCDEFGH");
encodeString(m_buff, "Topic");
encodeString(m_buff, "Message");
encodeString(m_buff, "Fakeuser");
encodeString(m_buff, "pwd");
List<Object> results = new ArrayList<Object >();
//Excercise
m_msgdec.decode(null, m_buff, results);
//Verify
assertFalse(results.isEmpty());
ConnectMessage message = (ConnectMessage)results.get(0);
verifyBaseHeader(message);
assertEquals(AbstractMessage.CONNECT, message.getMessageType());
assertEquals("ABCDEFGH", message.getClientID());
assertEquals("Topic", message.getWillTopic());
assertEquals("Message", message.getWillMessage());
assertEquals("Fakeuser", message.getUsername());
assertEquals("pwd", message.getPassword());
}
@Test(expected = CorruptedFrameException.class)
public void testBadFlagUserPwd() throws UnsupportedEncodingException, Exception {
m_buff = Unpooled.buffer(14);
m_buff.writeByte((AbstractMessage.CONNECT << 4)).writeByte(12);
//Proto name
encodeString(m_buff, "MQIsdp");
//version
m_buff.writeByte(3);
//conn flags
m_buff.writeByte(0x4E); //sets user to false and password to true
//keepAlive
m_buff.writeByte(0).writeByte(0x0A);
List<Object> results = new ArrayList<Object >();
//Excercise
m_msgdec.decode(null, m_buff, results);
}
private void initBaseHeader(ByteBuf buff) throws UnsupportedEncodingException {
initHeader(buff, (byte)0x0C);
}
private void initHeader(ByteBuf buff, byte remaingLength) throws UnsupportedEncodingException {
buff.clear().writeByte((byte)(AbstractMessage.CONNECT << 4)).writeByte(remaingLength);
//Proto name
encodeString(buff, "MQIsdp");
//version
buff.writeByte(3);
//conn flags
buff.writeByte(0xCE);
//keepAlive
buff.writeBytes(new byte[]{(byte)0, (byte) 0x0A});
}
/**
* Encode and insert the given string into the given buff
*/
private ByteBuf encodeString(ByteBuf buff, String str) throws UnsupportedEncodingException {
buff.writeBytes(Utils.encodeString(str));
return buff;
}
private void verifyBaseHeader(ConnectMessage connMessage) {
assertNotNull(connMessage);
assertEquals("MQIsdp", connMessage.getProtocolName());
assertEquals(3, connMessage.getProcotolVersion());
assertTrue(connMessage.isUserFlag());
assertTrue(connMessage.isPasswordFlag());
assertTrue(connMessage.isCleanSession());
assertEquals(10, connMessage.getKeepAlive());
assertTrue(connMessage.isWillFlag());
assertFalse(connMessage.isWillRetain());
assertEquals(1, connMessage.getWillQos());
}
}
@@ -0,0 +1,124 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import org.junit.Test;
import org.junit.Before;
import static org.junit.Assert.*;
import static org.dna.mqtt.moquette.parser.netty.TestUtils.*;
/**
*
* @author andrea
*/
public class ConnectEncoderTest {
ConnectEncoder m_encoder = new ConnectEncoder();
ChannelHandlerContext m_mockedContext;
@Before
public void setUp() {
//mock the ChannelHandlerContext to return an UnpooledAllocator
m_mockedContext = mockChannelHandler();
}
@Test
public void testBaseHeader() throws Exception {
ConnectMessage msg = new ConnectMessage();
msg.setWillRetain(true);
msg.setWillQos((byte)2);
msg.setWillFlag(false);
msg.setCleanSession(true);
msg.setKeepAlive(512);
ByteBuf out = Unpooled.buffer();
//Exercise
m_encoder.encode(m_mockedContext, msg, out);
//Verify
assertEquals(0x10, out.readByte()); //1 byte
assertEquals(12, out.readByte()); //remaining length
verifyString("MQIsdp", out);
assertEquals(0x03, out.readByte()); //protocol version
assertEquals(0x32, out.readByte()); //flags
assertEquals(2, out.readByte()); //keepAliveTimer msb
assertEquals(0, out.readByte()); //keepAliveTimer lsb
}
@Test
public void testCompleteHeader() throws Exception {
ConnectMessage msg = new ConnectMessage();
msg.setWillRetain(true);
msg.setWillQos((byte)2);
msg.setWillFlag(true);
msg.setCleanSession(true);
msg.setKeepAlive(512);
//variable part
msg.setClientID("ABCDEF");
msg.setWillTopic("Topic");
msg.setWillMessage("Message");
ByteBuf out = Unpooled.buffer();
//Exercise
m_encoder.encode(m_mockedContext, msg, out);
//Verify
assertEquals(0x10, out.readByte()); //1 byte
assertEquals(36, out.readByte()); //remaining length
verifyString("MQIsdp", out);
assertEquals(0x03, out.readByte()); //protocol version
assertEquals(0x36, out.readByte()); //flags
assertEquals(2, out.readByte()); //keepAliveTimer msb
assertEquals(0, out.readByte()); //keepAliveTimer lsb
//Variable part
verifyString("ABCDEF", out);
verifyString("Topic", out);
verifyString("Message", out);
}
@Test
public void testCompleteHeaderWIthUser_password() throws Exception {
ConnectMessage msg = new ConnectMessage();
msg.setUserFlag(true);
msg.setPasswordFlag(true);
msg.setWillRetain(true);
msg.setWillQos((byte)2);
msg.setWillFlag(true);
msg.setCleanSession(true);
msg.setKeepAlive(512);
//variable part
msg.setClientID("ABCDEF");
msg.setWillTopic("Topic");
msg.setWillMessage("Message");
msg.setUsername("Pablo");
msg.setPassword("PBL");
ByteBuf out = Unpooled.buffer();
//Exercise
m_encoder.encode(m_mockedContext, msg, out);
//Verify
assertEquals(0x10, out.readByte()); //1 byte
assertEquals(48, out.readByte()); //remaining length
verifyString("MQIsdp", out);
assertEquals(0x03, out.readByte()); //protocol version
assertEquals((byte)0xF6, (byte)out.readByte()); //flags
assertEquals(2, out.readByte()); //keepAliveTimer msb
assertEquals(0, out.readByte()); //keepAliveTimer lsb
//Variable part
verifyString("ABCDEF", out);
verifyString("Topic", out);
verifyString("Message", out);
verifyString("Pablo", out);//username
verifyString("PBL", out);//password
}
}
@@ -0,0 +1,51 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.util.ArrayList;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import org.junit.Before;
import org.junit.Test;
/**
*
* @author andrea
*/
public class PubAckDecoderTest {
ByteBuf m_buff;
PubAckDecoder m_msgdec;
@Before
public void setUp() {
m_msgdec = new PubAckDecoder();
}
@Test
public void testHeader() throws Exception {
m_buff = Unpooled.buffer(14);
int messageId = 0xAABB;
initHeader(m_buff, messageId);
List<Object> results = new ArrayList<Object >();
//Excercise
m_msgdec.decode(null, m_buff, results);
assertFalse(results.isEmpty());
PubAckMessage message = (PubAckMessage)results.get(0);
assertNotNull(message);
assertEquals(messageId, message.getMessageID().intValue());
assertEquals(AbstractMessage.PUBACK, message.getMessageType());
}
private void initHeader(ByteBuf buff, int messageID) {
buff.clear().writeByte(AbstractMessage.PUBACK << 4).writeByte(2);
//return code
buff.writeShort(messageID);
}
}
@@ -0,0 +1,42 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import org.junit.Test;
import org.junit.Before;
import static org.junit.Assert.*;
import static org.dna.mqtt.moquette.parser.netty.TestUtils.*;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
/**
*
* @author andrea
*/
public class PubAckEncoderTest {
ChannelHandlerContext m_mockedContext;
@Before
public void setUp() {
//mock the ChannelHandlerContext to return an UnpooledAllocator
m_mockedContext = mockChannelHandler();
}
@Test
public void testHeaderEncode() throws Exception {
int messageID = 0xAABB;
PubAckEncoder encoder = new PubAckEncoder();
PubAckMessage msg = new PubAckMessage();
msg.setMessageID(messageID);
ByteBuf out = Unpooled.buffer();
//Exercise
encoder.encode(m_mockedContext, msg, out);
//Verify
assertEquals(0x40, out.readByte()); //1 byte
assertEquals(0x02, out.readByte()); //2 byte, length
assertEquals((byte)0xAA, out.readByte());
assertEquals((byte)0xBB, out.readByte());
}
}
@@ -0,0 +1,188 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.util.ArrayList;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import org.junit.Before;
import org.junit.Test;
/**
*
* @author andrea
*/
public class PublishDecoderTest {
ByteBuf m_buff;
PublishDecoder m_msgdec;
List<Object> m_results;
@Before
public void setUp() {
m_msgdec = new PublishDecoder();
m_results = new ArrayList<Object >();
}
@Test
public void testHeader() throws Exception {
m_buff = Unpooled.buffer(14);
initHeader(m_buff);
//Excercise
m_msgdec.decode(null, m_buff, m_results);
assertFalse(m_results.isEmpty());
PublishMessage message = (PublishMessage)m_results.get(0);
assertNotNull(message);
assertEquals("Fake Topic", message.getTopicName());
assertNull(message.getMessageID());
assertEquals(AbstractMessage.PUBLISH, message.getMessageType());
}
@Test
public void testHeaderWithMessageID() throws Exception {
m_buff = Unpooled.buffer(14);
int messageID = 123;
initHeaderWithMessageID(m_buff, messageID);
//Exercise
m_msgdec.decode(null, m_buff, m_results);
assertFalse(m_results.isEmpty());
PublishMessage message = (PublishMessage)m_results.get(0);
assertNotNull(message);
assertEquals("Fake Topic", message.getTopicName());
assertEquals(messageID, (int) message.getMessageID());
}
@Test
public void testHeaderWithMessageID_Payload() throws Exception {
m_buff = Unpooled.buffer(14);
int messageID = 123;
byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
initHeaderWithMessageID_Payload(m_buff, messageID, payload);
//Exercise
m_msgdec.decode(null, m_buff, m_results);
assertFalse(m_results.isEmpty());
PublishMessage message = (PublishMessage)m_results.get(0);
assertNotNull(message);
assertEquals("Fake Topic", message.getTopicName());
assertEquals(messageID, (int) message.getMessageID());
TestUtils.verifyEquals(payload, message.getPayload());
}
@Test
public void testBugOnRealCase() throws Exception {
byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte lenght
0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x69, 0x63, //[/topic] string 2 len + 6 content
0x54, 0x65, 0x73, 0x74, 0x20, 0x6d, 0x79, // [Test my payload] encoding
0x20, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
m_buff = Unpooled.buffer(overallMessage.length);
m_buff.writeBytes(overallMessage);
//Exercise
m_msgdec.decode(null, m_buff, m_results);
assertFalse(m_results.isEmpty());
PublishMessage message = (PublishMessage)m_results.get(0);
assertNotNull(message);
}
@Test
public void testDecodeBigContent() throws Exception {
int size = 129;
ByteBuf payload = TestUtils.generateRandomPayload(size);
ByteBuf firstPublish = generatePublishQoS0(payload);
ByteBuf secondPublish = generatePublishQoS0(TestUtils.generateRandomPayload(size));
ByteBuf doubleMessageBuf = Unpooled.buffer(size * 2);
doubleMessageBuf.writeBytes(firstPublish).writeBytes(secondPublish);
//Exercise
m_msgdec.decode(null, doubleMessageBuf, m_results);
assertFalse(m_results.isEmpty());
PublishMessage message = (PublishMessage)m_results.get(0);
assertNotNull(message);
m_results.clear();
m_msgdec.decode(null, doubleMessageBuf, m_results);
assertFalse(m_results.isEmpty());
message = (PublishMessage)m_results.get(0);
assertNotNull(message);
}
@Test
public void testBadClaimMoreData() throws Exception {
byte[] rawMessage = new byte[] {
0x30, 0x17, 0x00, 0x06, 0x2f, (byte)0x74, 0x6f, (byte)0x70, 0x69, 0x63, 0x54, 0x65, (byte)0x73, 0x74, 0x20, 0x6d,
(byte)0x79, 0x20, (byte)0x70, 0x61, (byte)0x79, 0x6c, 0x6f, 0x61, 0x64
};
ByteBuf msgBuf = Unpooled.buffer(25);
msgBuf.writeBytes(rawMessage);
msgBuf.readByte(); //to simulate the reading of messageType done by MQTTDecoder dispatcher
//Exercise
m_msgdec.decode(null, msgBuf, m_results);
assertFalse(m_results.isEmpty());
PublishMessage message = (PublishMessage)m_results.get(0);
assertNotNull(message);
assertEquals("/topic", message.getTopicName());
assertEquals("Test my payload", new String(message.getPayload()));
}
private void initHeader(ByteBuf buff) throws IllegalAccessException {
ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic"));
buff.clear().writeByte(AbstractMessage.PUBLISH << 4).writeBytes(Utils.encodeRemainingLength(tmp.readableBytes()));
//topic name
buff.writeBytes(tmp);
}
private void initHeaderWithMessageID(ByteBuf buff, int messageID) throws IllegalAccessException {
ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic"));
tmp.writeShort(messageID);
buff.clear().writeByte(AbstractMessage.PUBLISH << 4 | 0x02) //set Qos to 1
.writeBytes(Utils.encodeRemainingLength(tmp.readableBytes()));
//topic name
buff.writeBytes(tmp);
}
private void initHeaderWithMessageID_Payload(ByteBuf buff, int messageID, byte[] payload) throws IllegalAccessException {
ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic"));
tmp.writeShort(messageID);
tmp.writeBytes(payload);
buff.clear().writeByte(AbstractMessage.PUBLISH << 4 | 0x02) //set Qos to 1
.writeBytes(Utils.encodeRemainingLength(tmp.readableBytes()));
//topic name
buff.writeBytes(tmp);
}
private ByteBuf generatePublishQoS0(ByteBuf payload) throws IllegalAccessException {
int size = payload.capacity();
ByteBuf messageBody = Unpooled.buffer(size);
messageBody.writeBytes(Utils.encodeString("/topic"));
//ONLY for QoS > 1 Utils.writeWord(messageBody, messageID);
messageBody.writeBytes(payload);
ByteBuf completeMsg = Unpooled.buffer(size);
completeMsg.clear().writeByte(AbstractMessage.PUBLISH << 4 | 0x00) //set Qos to 0
.writeBytes(Utils.encodeRemainingLength(messageBody.readableBytes()));
completeMsg.writeBytes(messageBody);
return completeMsg;
}
}
@@ -0,0 +1,175 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.*;
import static org.dna.mqtt.moquette.parser.netty.TestUtils.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
/**
*
* @author andrea
*/
public class PublishEncoderTest {
PublishEncoder m_encoder = new PublishEncoder();
ChannelHandlerContext m_mockedContext;
ByteBuf m_out;
@Before
public void setUp() {
//mock the ChannelHandlerContext to return an UnpooledAllocator
m_mockedContext = mockChannelHandler();
m_out = Unpooled.buffer();
}
@Test
public void testEncodeWithQos_0_noMessageID() throws Exception {
String topic = "/photos";
PublishMessage msg = new PublishMessage();
msg.setQos(QOSType.MOST_ONE);
msg.setTopicName(topic);
//variable part
byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
msg.setPayload(payload);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
//Verify
assertEquals(0x30, m_out.readByte()); //1 byte
//(2+7) topic + 3 payload
assertEquals(12, m_out.readByte()); //remaining length
//Variable part
verifyString(topic, m_out);
verifyBuff(payload.length, payload, m_out);
}
@Test
public void testEncodeWithQos_1_MessageID() throws Exception {
String topic = "/photos";
PublishMessage msg = new PublishMessage();
msg.setQos(QOSType.LEAST_ONE);
msg.setMessageID(1);
msg.setTopicName(topic);
//variable part
byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
msg.setPayload(payload);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
//Verify
assertEquals(0x32, m_out.readByte()); //1 byte
//(2+7) topic + 2 messageID + 3 payload
assertEquals(14, m_out.readByte()); //remaining length
//Variable part
verifyString(topic, m_out);
assertEquals(0, m_out.readByte()); //MessageID MSB
assertEquals(1, m_out.readByte()); //MessageID LSB
verifyBuff(payload.length, payload, m_out);
}
@Test(expected = IllegalArgumentException.class)
public void testEncodeWithQos_1_noMessageID_fail() throws Exception {
String topic = "pictures/photos";
PublishMessage msg = new PublishMessage();
msg.setQos(QOSType.LEAST_ONE);
msg.setTopicName(topic);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
}
@Test(expected = IllegalArgumentException.class)
public void testEncode_empty_topic() throws Exception {
PublishMessage msg = new PublishMessage();
msg.setQos(QOSType.LEAST_ONE);
msg.setMessageID(1);
ByteBuf out = Unpooled.buffer();
//Exercise
m_encoder.encode(m_mockedContext, msg, out);
}
@Test(expected = IllegalArgumentException.class)
public void testEncode_null_topic() throws Exception {
PublishMessage msg = new PublishMessage();
msg.setQos(QOSType.LEAST_ONE);
msg.setMessageID(1);
msg.setTopicName(null);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
}
@Test(expected = IllegalArgumentException.class)
public void testEncode_qos_reserved() throws Exception {
PublishMessage msg = new PublishMessage();
msg.setQos(QOSType.RESERVED);
msg.setMessageID(1);
msg.setTopicName(null);
//variable part
byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
msg.setPayload(payload);
ByteBuf out = Unpooled.buffer();
//Exercise
m_encoder.encode(m_mockedContext, msg, out);
}
@Test
public void testBugMissedMessageIDEncoding() throws Exception{
String topic = "/topic";
PublishMessage msg = new PublishMessage();
msg.setQos(QOSType.MOST_ONE);
msg.setTopicName(topic);
//variable part
byte[] payload = "Test my payload".getBytes();
msg.setPayload(payload);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
//Verify
//2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload
int size = m_out.readableBytes();
assertEquals(10+payload.length, size);
assertEquals(0x30, m_out.readByte()); //1 byte
assertEquals(23, m_out.readByte()); //1 byte the length
}
@Test
public void testEncodeBigMessage() throws Exception {
byte[] payload = new byte[256];
TestUtils.generateRandomPayload(256).readBytes(payload);
String topic = "/topic";
PublishMessage msg = new PublishMessage();
msg.setQos(QOSType.MOST_ONE);
msg.setTopicName(topic);
//variable part
msg.setPayload(payload);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
int size = m_out.readableBytes();
assertEquals(11+payload.length, size);
assertEquals(0x30, m_out.readByte()); //1 byte
}
}
@@ -0,0 +1,78 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.util.ArrayList;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
/**
*
* @author andrea
*/
public class SubAckDecoderTest {
ByteBuf m_buff;
SubAckDecoder m_msgdec;
List<Object> m_results;
@Before
public void setUp() {
m_msgdec = new SubAckDecoder();
m_results = new ArrayList<Object>();
m_buff = Unpooled.buffer(7);
}
@Test
public void testBadQos() throws Exception {
initHeaderQos(m_buff, 0xAABB, AbstractMessage.QOSType.LEAST_ONE, AbstractMessage.QOSType.MOST_ONE, AbstractMessage.QOSType.MOST_ONE);
//Excercise
m_msgdec.decode(null, m_buff, m_results);
//Verify
assertFalse(m_results.isEmpty());
SubAckMessage message = (SubAckMessage)m_results.get(0);
assertNotNull(message);
assertEquals(0xAABB, message.getMessageID().intValue());
List<AbstractMessage.QOSType> qoses = message.types();
assertEquals(3, qoses.size());
assertEquals(AbstractMessage.QOSType.LEAST_ONE, qoses.get(0));
assertEquals(AbstractMessage.QOSType.MOST_ONE, qoses.get(1));
assertEquals(AbstractMessage.QOSType.MOST_ONE, qoses.get(2));
assertEquals(AbstractMessage.SUBACK, message.getMessageType());
}
@Test
public void testBugBadRemainingCalculation() throws Exception {
byte[] overallMessage = new byte[] {(byte)0x90, 0x03, //fixed header
0x00, 0x0A, //MSG ID
0x01}; //QoS array
m_buff = Unpooled.buffer(overallMessage.length);
m_buff.writeBytes(overallMessage);
//Exercise
m_msgdec.decode(null, m_buff, m_results);
assertFalse(m_results.isEmpty());
SubAckMessage message = (SubAckMessage)m_results.get(0);
assertNotNull(message);
assertEquals(0x0A, message.getMessageID().intValue());
assertEquals(1, message.types().size());
assertEquals(AbstractMessage.QOSType.LEAST_ONE, message.types().get(0));
}
private void initHeaderQos(ByteBuf buff, int messageID, AbstractMessage.QOSType... qoss) throws IllegalAccessException {
buff.clear().writeByte(AbstractMessage.SUBACK << 4).
writeBytes(Utils.encodeRemainingLength(2 + qoss.length));
buff.writeShort(messageID);
for (AbstractMessage.QOSType qos : qoss) {
buff.writeByte(qos.ordinal());
}
}
}
@@ -0,0 +1,62 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import static org.dna.mqtt.moquette.parser.netty.TestUtils.mockChannelHandler;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import static org.junit.Assert.assertEquals;
import org.junit.Before;
import org.junit.Test;
/**
*
* @author andrea
*/
public class SubAckEncoderTest {
SubAckEncoder m_encoder = new SubAckEncoder();
ChannelHandlerContext m_mockedContext;
ByteBuf m_out;
@Before
public void setUp() {
//mock the ChannelHandlerContext to return an UnpooledAllocator
m_mockedContext = mockChannelHandler();
m_out = Unpooled.buffer();
}
@Test(expected = IllegalArgumentException.class)
public void testEncodeWithNoQoss() throws Exception {
SubAckMessage msg = new SubAckMessage();
msg.setMessageID(123);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
}
@Test
public void testEncodeWithMultipleQos() throws Exception {
SubAckMessage msg = new SubAckMessage();
int messageID = 0xAABB;
msg.setMessageID(messageID);
msg.addType(AbstractMessage.QOSType.MOST_ONE);
msg.addType(AbstractMessage.QOSType.LEAST_ONE);
msg.addType(AbstractMessage.QOSType.EXACTLY_ONCE);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
//Verify
assertEquals((byte) (AbstractMessage.SUBACK << 4 ), m_out.readByte()); //1 byte
assertEquals(5, m_out.readByte()); //remaining length
//Variable part
assertEquals((byte)0xAA, m_out.readByte()); //MessageID MSB
assertEquals((byte)0xBB, m_out.readByte()); //MessageID LSB
assertEquals((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), m_out.readByte());
assertEquals((byte)AbstractMessage.QOSType.LEAST_ONE.ordinal(), m_out.readByte());
assertEquals((byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), m_out.readByte());
}
}
@@ -0,0 +1,71 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.CorruptedFrameException;
import java.util.ArrayList;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
/**
*
* @author andrea
*/
public class SubscribeDecoderTest {
ByteBuf m_buff;
SubscribeDecoder m_msgdec;
List<Object> m_results;
@Before
public void setUp() {
m_msgdec = new SubscribeDecoder();
m_results = new ArrayList<Object >();
}
@Test(expected = CorruptedFrameException.class)
public void testBadQos() throws Exception {
m_buff = Unpooled.buffer(2);
initHeaderBadQos(m_buff);
//Excercise
m_msgdec.decode(null, m_buff, m_results);
}
@Test
public void testMultiTopic() throws Exception {
m_buff = Unpooled.buffer(4);
SubscribeMessage.Couple c1 = new SubscribeMessage.Couple((byte)2, "a/b");
SubscribeMessage.Couple c2 = new SubscribeMessage.Couple((byte)1, "c/d/e");
initMultiTopic(m_buff, 123, c1, c2);
//Excercise
m_msgdec.decode(null, m_buff, m_results);
//Verify
assertFalse(m_results.isEmpty());
SubscribeMessage message = (SubscribeMessage)m_results.get(0);
assertEquals(2, message.subscriptions().size());
assertEquals(AbstractMessage.SUBSCRIBE, message.getMessageType());
}
private void initHeaderBadQos(ByteBuf buff) {
buff.clear().writeByte(AbstractMessage.SUBSCRIBE << 4).writeByte(0);
}
private void initMultiTopic(ByteBuf buff, int messageID, SubscribeMessage.Couple... topics) throws IllegalAccessException {
ByteBuf topicBuffer = Unpooled.buffer(4);
topicBuffer.writeShort(messageID);
for (SubscribeMessage.Couple couple : topics) {
topicBuffer.writeBytes(Utils.encodeString(couple.getTopic()));
topicBuffer.writeByte(couple.getQos());
}
buff.clear().writeByte(AbstractMessage.SUBSCRIBE << 4 | (byte)0x02)
.writeBytes(Utils.encodeRemainingLength(topicBuffer.readableBytes()));
buff.writeBytes(topicBuffer);
}
}
@@ -0,0 +1,78 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import static org.junit.Assert.assertEquals;
import org.junit.Before;
import org.junit.Test;
import static org.dna.mqtt.moquette.parser.netty.TestUtils.*;
/**
*
* @author andrea
*/
public class SubscribeEncoderTest {
SubscribeEncoder m_encoder = new SubscribeEncoder();
ChannelHandlerContext m_mockedContext;
ByteBuf m_out;
@Before
public void setUp() {
//mock the ChannelHandlerContext to return an UnpooledAllocator
m_mockedContext = mockChannelHandler();
m_out = Unpooled.buffer();
}
@Test
public void testEncodeWithMultiTopic() throws Exception {
SubscribeMessage msg = new SubscribeMessage();
msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
msg.setMessageID(0xAABB);
//variable part
SubscribeMessage.Couple c1 = new SubscribeMessage.Couple((byte)1, "a/b");
SubscribeMessage.Couple c2 = new SubscribeMessage.Couple((byte)0, "a/b/c");
msg.addSubscription(c1);
msg.addSubscription(c2);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
//Verify
assertEquals((byte)0x82, (byte)m_out.readByte()); //1 byte
assertEquals(16, m_out.readByte()); //remaining length
//verify MessageID
assertEquals((byte)0xAA, m_out.readByte());
assertEquals((byte)0xBB, m_out.readByte());
//Variable part
verifyString(c1.getTopic(), m_out);
assertEquals(c1.getQos(), m_out.readByte());
verifyString(c2.getTopic(), m_out);
assertEquals(c2.getQos(), m_out.readByte());
}
@Test(expected = IllegalArgumentException.class)
public void testEncode_empty_subscription() throws Exception {
SubscribeMessage msg = new SubscribeMessage();
msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
msg.setMessageID(0xAABB);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
}
@Test(expected = IllegalArgumentException.class)
public void testEncode_badQos() throws Exception {
SubscribeMessage msg = new SubscribeMessage();
msg.setQos(AbstractMessage.QOSType.EXACTLY_ONCE);
msg.setMessageID(0xAABB);
//Exercise
m_encoder.encode(m_mockedContext, msg, m_out);
}
}
@@ -0,0 +1,75 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import java.io.UnsupportedEncodingException;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.*;
/**
* Common test methods
*
* @author andrea
*/
public class TestUtils {
/**
* Verify that the given bytes buffer of the given numBytes length is present
* in the buff starting from the current position.
*/
static void verifyBuff(int numBytes, byte[] bytes, ByteBuf buff) {
assertTrue(numBytes <= buff.readableBytes());
byte[] toCheck = new byte[numBytes];
buff.readBytes(toCheck);
for (int i = 0; i < numBytes; i++) {
assertEquals(bytes[i], toCheck[i]);
}
}
/**
* Verify the presence of the given string starting from the current position
* inside the buffer.
*/
static void verifyString(String str, ByteBuf buff) throws UnsupportedEncodingException {
ByteBuf tmpBuff = Unpooled.buffer(2);
byte[] raw = str.getBytes("UTF-8");
tmpBuff.writeShort(raw.length);
tmpBuff.writeBytes(raw);
int buffLen = raw.length + 2;
verifyByteBuf(tmpBuff, buff.slice(buff.readerIndex(), buffLen));
buff.skipBytes(buffLen);
}
static void verifyByteBuf(ByteBuf expected, ByteBuf found) {
assertEquals(expected, found);
}
static ChannelHandlerContext mockChannelHandler() {
ChannelHandlerContext m_mockedContext = mock(ChannelHandlerContext.class);
ByteBufAllocator allocator = UnpooledByteBufAllocator.DEFAULT;
when(m_mockedContext.alloc()).thenReturn(allocator);
return m_mockedContext;
}
static void verifyEquals(byte[] expected, byte[] found) {
assertEquals(expected.length, found.length);
for (int i = 0; i < expected.length; i++) {
assertEquals(expected[i], found[i]);
}
}
public static ByteBuf generateRandomPayload(int size) {
ByteBuf payloadBuffer = Unpooled.buffer(size);
for (int i = 0; i < size; i++) {
payloadBuffer.writeByte((byte) (Math.random() * 255));
}
return payloadBuffer;
}
}
@@ -0,0 +1,95 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.CorruptedFrameException;
import java.util.ArrayList;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
/**
*
* @author andrea
*/
public class UnsubscribeDecoderTest {
ByteBuf m_buff;
UnsubscribeDecoder m_msgdec;
List<Object> m_results;
@Before
public void setUp() {
m_msgdec = new UnsubscribeDecoder();
m_results = new ArrayList<Object >();
}
@Test(expected = CorruptedFrameException.class)
public void testBadQos() throws Exception {
m_buff = Unpooled.buffer(2);
initHeaderBadQos(m_buff);
//Excercise
m_msgdec.decode(null, m_buff, m_results);
}
@Test
public void testMultiTopic() throws Exception {
m_buff = Unpooled.buffer(4);
String topic1 = "a/b";
String topic2 = "c/d/e";
initMultiTopic(m_buff, 123, topic1, topic2);
//Excercise
m_msgdec.decode(null, m_buff, m_results);
//Verify
assertFalse(m_results.isEmpty());
UnsubscribeMessage message = (UnsubscribeMessage)m_results.get(0);
assertNotNull(message);
assertEquals(2, message.topics().size());
assertEquals(topic1, message.topics().get(0));
assertEquals(topic2, message.topics().get(1));
assertEquals(AbstractMessage.UNSUBSCRIBE, message.getMessageType());
}
@Test
public void testDecodeSingleTopic_bug() throws Exception {
//A2 0C 00 01 00 06 2F 74 6F 70 69 63 //12 byte
byte[] overallMessage = new byte[] {(byte)0xA2, 0x0A, //fixed header
0x00, 0x01, //MSG ID
0x00, 0x06, 0x2F, 0x74, 0x6F, 0x70, 0x69, 0x63}; //"/topic" string
m_buff = Unpooled.buffer(overallMessage.length);
m_buff.writeBytes(overallMessage);
m_msgdec.decode(null, m_buff, m_results);
assertFalse(m_results.isEmpty());
UnsubscribeMessage message = (UnsubscribeMessage)m_results.get(0);
assertNotNull(message);
assertEquals(0x01, (int)message.getMessageID());
}
private void initHeaderBadQos(ByteBuf buff) {
buff.clear().writeByte(AbstractMessage.UNSUBSCRIBE << 4).writeByte(0);
}
private void initMultiTopic(ByteBuf buff, int messageID, String... topics) throws IllegalAccessException {
ByteBuf topicBuffer = Unpooled.buffer(4);
topicBuffer.writeShort(messageID);
for (String topic : topics) {
topicBuffer.writeBytes(Utils.encodeString(topic));
}
buff.clear().writeByte(AbstractMessage.UNSUBSCRIBE << 4 | (byte)0x02)
.writeBytes(Utils.encodeRemainingLength(topicBuffer.readableBytes()));
buff.writeBytes(topicBuffer);
}
}
@@ -0,0 +1,99 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import static org.dna.mqtt.moquette.parser.netty.TestUtils.mockChannelHandler;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
import static org.junit.Assert.assertEquals;
import org.junit.Before;
import org.junit.Test;
import static org.dna.mqtt.moquette.parser.netty.TestUtils.*;
/**
*
* @author andrea
*/
public class UnsubscribeEncoderTest {
UnsubscribeEncoder m_encoder = new UnsubscribeEncoder();
ChannelHandlerContext m_mockedContext;
ByteBuf m_out;
UnsubscribeMessage m_msg;
@Before
public void setUp() {
//mock the ChannelHandlerContext to return an UnpooledAllocator
m_mockedContext = mockChannelHandler();
m_out = Unpooled.buffer();
m_msg = new UnsubscribeMessage();
m_msg.setMessageID(0xAABB);
}
@Test
public void testEncodeWithSingleTopic() throws Exception {
m_msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
//variable part
String topic1 = "/topic";
m_msg.addTopic(topic1);
//Exercise
m_encoder.encode(m_mockedContext, m_msg, m_out);
//Verify
assertEquals((byte)0xA2, (byte)m_out.readByte()); //1 byte
//2 messageID + 2 length + 6 chars = 10
assertEquals(10, m_out.readByte()); //remaining length
//verify M1ssageID
assertEquals((byte)0xAA, m_out.readByte());
assertEquals((byte)0xBB, m_out.readByte());
//Variable part
verifyString(topic1, m_out);
}
@Test
public void testEncodeWithMultiTopic() throws Exception {
m_msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
//variable part
String topic1 = "a/b";
String topic2 = "a/b/c";
m_msg.addTopic(topic1);
m_msg.addTopic(topic2);
//Exercise
m_encoder.encode(m_mockedContext, m_msg, m_out);
//Verify
assertEquals((byte)0xA2, (byte)m_out.readByte()); //1 byte
assertEquals(14, m_out.readByte()); //remaining length
//verify M1ssageID
assertEquals((byte)0xAA, m_out.readByte());
assertEquals((byte)0xBB, m_out.readByte());
//Variable part
verifyString(topic1, m_out);
verifyString(topic2, m_out);
}
@Test(expected = IllegalArgumentException.class)
public void testEncode_empty_topics() throws Exception {
m_msg.setQos(AbstractMessage.QOSType.LEAST_ONE);
//Exercise
m_encoder.encode(m_mockedContext, m_msg, m_out);
}
@Test(expected = IllegalArgumentException.class)
public void testEncode_badQos() throws Exception {
m_msg.setQos(AbstractMessage.QOSType.EXACTLY_ONCE);
//Exercise
m_encoder.encode(m_mockedContext, m_msg, m_out);
}
}
@@ -0,0 +1,93 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.CorruptedFrameException;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.*;
import static org.dna.mqtt.moquette.parser.netty.TestUtils.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
/**
*
* @author andrea
*/
/*@RunWith(Suite.class)
@Suite.SuiteClasses({})*/
public class UtilsTest {
ByteBuf m_buff;
@Before
public void setUp() {
m_buff = Unpooled.buffer(4);
}
@Test
public void testDecodeRemainingLength() {
//1 byte length
m_buff.writeByte(0x0);
assertEquals(0, Utils.decodeRemainingLenght(m_buff));
m_buff.clear().writeByte(0x007F);
assertEquals(127, Utils.decodeRemainingLenght(m_buff));
//2 byte length
m_buff.clear().writeBytes(new byte[]{(byte)0x80, (byte)0x01});
assertEquals(128, Utils.decodeRemainingLenght(m_buff));
m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0x7F});
assertEquals(16383, Utils.decodeRemainingLenght(m_buff));
//3 byte length
m_buff.clear().writeBytes(new byte[]{(byte)0x80, (byte)0x80, (byte)0x01});
assertEquals(16384, Utils.decodeRemainingLenght(m_buff));
m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0xFF, (byte)0x7F});
assertEquals(2097151, Utils.decodeRemainingLenght(m_buff));
//4 byte length
m_buff.clear().writeBytes(new byte[]{(byte)0x80, (byte)0x80, (byte)0x80, (byte)0x01});
assertEquals(2097152, Utils.decodeRemainingLenght(m_buff));
m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0xFF, (byte)0xFF, (byte)0x7F});
assertEquals(268435455, Utils.decodeRemainingLenght(m_buff));
}
@Test(expected=CorruptedFrameException.class)
public void testEncodeRemainingLength_invalid_upper() {
Utils.encodeRemainingLength(Utils.MAX_LENGTH_LIMIT + 1);
}
@Test(expected=CorruptedFrameException.class)
public void testEncodeRemainingLength_invalid_lower() {
Utils.encodeRemainingLength(-1);
}
@Test
public void testEncodeRemainingLenght() {
//1 byte length
verifyBuff(1, new byte[]{0}, Utils.encodeRemainingLength(0));
verifyBuff(1, new byte[]{0x7F}, Utils.encodeRemainingLength(127));
//2 byte length
verifyBuff(2, new byte[]{(byte)0x80, 0x01}, Utils.encodeRemainingLength(128));
verifyBuff(2, new byte[]{(byte)0xFF, 0x7F}, Utils.encodeRemainingLength(16383));
//3 byte length
verifyBuff(3, new byte[]{(byte)0x80, (byte)0x80, 0x01}, Utils.encodeRemainingLength(16384));
verifyBuff(3, new byte[]{(byte)0xFF, (byte)0xFF, 0x7F}, Utils.encodeRemainingLength(2097151));
//4 byte length
verifyBuff(4, new byte[]{(byte)0x80, (byte)0x80, (byte)0x80, 0x01}, Utils.encodeRemainingLength(2097152));
verifyBuff(4, new byte[]{(byte)0xFF, (byte)0xFF, (byte)0xFF, 0x7F}, Utils.encodeRemainingLength(268435455));
}
@Test
public void testEncodeFlags() {
UnsubscribeMessage msg = new UnsubscribeMessage();
msg.setRetainFlag(true);
msg.setQos(AbstractMessage.QOSType.MOST_ONE);
//Exercise
assertEquals(1, Utils.encodeFlags(msg));
}
}
+8 -2
View File
@@ -9,10 +9,10 @@
<version>0.2</version>
</parent>
<artifactId>moquette-parser</artifactId>
<artifactId>moquette-mina-parser</artifactId>
<packaging>jar</packaging>
<version>0.2</version>
<name>Moquette - Parser</name>
<name>Moquette - MINA Parser</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
@@ -25,6 +25,12 @@
<version>2.0.4</version>
</dependency>
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-parser-commons</artifactId>
<version>0.2</version>
</dependency>
<!--dependency>
<groupId>com.googlecode.disruptor</groupId>
<artifactId>disruptor</artifactId>
@@ -1,5 +1,6 @@
package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
@@ -5,6 +5,7 @@ import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolEncoderOutput;
import org.apache.mina.filter.codec.demux.MessageEncoder;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
/**
*
@@ -1,11 +0,0 @@
package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.messages.MessageIDMessage;
/**
*
* @author andrea
*/
public class PubCompMessage extends MessageIDMessage {
}
@@ -1,13 +1,11 @@
package org.dna.mqtt.moquette.proto;
import java.io.UnsupportedEncodingException;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolDecoderOutput;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
/**
@@ -1,6 +1,5 @@
package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.ConnAckEncoder;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.junit.Test;
import static org.junit.Assert.*;
@@ -152,7 +152,7 @@ public class ConnectDecoderTest {
}
/**
* Encode and insert the given stirng into the given buff
* Encode and insert the given string into the given buff
*/
private IoBuffer encodeString(IoBuffer buff, String str) throws UnsupportedEncodingException {
buff.put(Utils.encodeString(str));
@@ -1,6 +1,5 @@
package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.PubAckEncoder;
import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import org.junit.Test;
import static org.junit.Assert.*;
@@ -1,9 +1,5 @@
package org.dna.mqtt.moquette.proto;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.filter.codec.demux.MessageDecoder;
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
@@ -12,7 +8,6 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
/**
@@ -1,6 +1,5 @@
package org.dna.mqtt.moquette.proto;
import org.apache.mina.core.buffer.IoBuffer;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.junit.Before;
@@ -1,6 +1,5 @@
package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.SubAckEncoder;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
@@ -79,7 +79,7 @@ public class TestUtils {
}
/**
* Verify that the given bytes buffer og the given numBytes length is present
* Verify that the given bytes buffer of the given numBytes length is present
* in the buff starting from the current position.
*/
static void verifyBuff(int numBytes, byte[] bytes, IoBuffer buff) {
@@ -1,6 +1,5 @@
package org.dna.mqtt.moquette.proto;
import org.dna.mqtt.moquette.proto.UnsubscribeEncoder;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.junit.Before;
+86
View File
@@ -0,0 +1,86 @@
<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>../pom.xml</relativePath>
<artifactId>moquette-parent</artifactId>
<groupId>org.dna.mqtt</groupId>
<version>0.2</version>
</parent>
<artifactId>moquette-parser-commons</artifactId>
<packaging>jar</packaging>
<version>0.2</version>
<name>Moquette - Parser Commons</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>junit</groupId>
<artifactId>junit</artifactId>
<version>4.8.2</version>
<scope>test</scope>
</dependency>
<!-- needs extra dependencies: objenesis & hamcrest -->
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>1.9.0</version>
<scope>test</scope>
</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>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<excludes>
<exclude>org/dna/mqtt/moquette/proto/TestUtils.java</exclude>
<exclude>org/dna/mqtt/moquette/proto/TestUtils$*</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<!-- <repository>
<id>Fusesource Releases</id>
<url>http://repo.fusesource.com/nexus/content/repositories/snapshots</url>
</repository>-->
<!-- <repository>
<id>fusesource.snapshots</id>
<name>FuseSource Snapshot Repository</name>
<url>http://repo.fusesource.com/nexus/content/repositories/snapshots</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>-->
</repositories>
</project>
@@ -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;
@@ -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;
@@ -6,4 +6,8 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class DisconnectMessage extends ZeroLengthMessage {
public DisconnectMessage() {
m_messageType = AbstractMessage.DISCONNECT;
}
}
@@ -6,4 +6,8 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class PingReqMessage extends ZeroLengthMessage {
public PingReqMessage() {
m_messageType = AbstractMessage.PINGREQ;
}
}
@@ -5,4 +5,8 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class PingRespMessage extends ZeroLengthMessage {
public PingRespMessage() {
m_messageType = AbstractMessage.PINGRESP;
}
}
@@ -6,4 +6,8 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class PubAckMessage extends MessageIDMessage {
public PubAckMessage() {
m_messageType = AbstractMessage.PUBACK;
}
}
@@ -0,0 +1,13 @@
package org.dna.mqtt.moquette.proto.messages;
/**
*
* @author andrea
*/
public class PubCompMessage extends MessageIDMessage {
public PubCompMessage() {
m_messageType = AbstractMessage.PUBCOMP;
}
}
@@ -7,4 +7,7 @@ package org.dna.mqtt.moquette.proto.messages;
*/
public class PubRecMessage extends MessageIDMessage {
public PubRecMessage() {
m_messageType = AbstractMessage.PUBREC;
}
}
@@ -5,4 +5,8 @@ package org.dna.mqtt.moquette.proto.messages;
* @author andrea
*/
public class PubRelMessage extends MessageIDMessage {
public PubRelMessage() {
m_messageType = AbstractMessage.PUBREL;
}
}
@@ -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;
@@ -10,6 +10,10 @@ import java.util.List;
public class SubAckMessage extends MessageIDMessage {
List<QOSType> m_types = new ArrayList<QOSType>();
public SubAckMessage() {
m_messageType = AbstractMessage.SUBACK;
}
public List<QOSType> types() {
return m_types;

Some files were not shown because too many files have changed in this diff Show More