Merge branch 'master' of https://code.google.com/p/moquette-mqtt
This commit is contained in:
selva.andre
+544
-544
File diff suppressed because it is too large
Load Diff
@@ -1,56 +1,62 @@
|
||||
package org.dna.mqtt.moquette.server;
|
||||
|
||||
import org.apache.mina.core.service.IoHandlerAdapter;
|
||||
import org.apache.mina.core.session.IdleStatus;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
|
||||
import org.dna.mqtt.moquette.proto.Utils;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
|
||||
|
||||
/**
|
||||
* MINA MQTT Handler used to route messages to protocol logic
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class MQTTHandler extends IoHandlerAdapter {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
|
||||
private IMessaging m_messaging;
|
||||
|
||||
@Override
|
||||
public void messageReceived(IoSession session, Object message) throws Exception {
|
||||
AbstractMessage msg = (AbstractMessage) message;
|
||||
LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType())));
|
||||
try {
|
||||
switch (msg.getMessageType()) {
|
||||
case CONNECT:
|
||||
case SUBSCRIBE:
|
||||
case UNSUBSCRIBE:
|
||||
case PUBLISH:
|
||||
case PUBREC:
|
||||
case PUBCOMP:
|
||||
case PUBREL:
|
||||
case PINGREQ:
|
||||
case DISCONNECT:
|
||||
m_messaging.handleProtocolMessage(session, msg);
|
||||
break;
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
LOG.error("Bad error in processing the message", ex);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sessionIdle(IoSession session, IdleStatus status) {
|
||||
session.close(false);
|
||||
}
|
||||
|
||||
public void setMessaging(IMessaging messaging) {
|
||||
m_messaging = messaging;
|
||||
}
|
||||
|
||||
}
|
||||
package org.dna.mqtt.moquette.server;
|
||||
|
||||
import org.apache.mina.core.service.IoHandlerAdapter;
|
||||
import org.apache.mina.core.session.IdleStatus;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
|
||||
import org.dna.mqtt.moquette.proto.Utils;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
|
||||
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* MINA MQTT Handler used to route messages to protocol logic
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class MQTTHandler extends IoHandlerAdapter {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(MQTTHandler.class);
|
||||
private IMessaging m_messaging;
|
||||
|
||||
@Override
|
||||
public void messageReceived(IoSession session, Object message) throws Exception {
|
||||
AbstractMessage msg = (AbstractMessage) message;
|
||||
LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType())));
|
||||
try {
|
||||
switch (msg.getMessageType()) {
|
||||
case CONNECT:
|
||||
case SUBSCRIBE:
|
||||
case UNSUBSCRIBE:
|
||||
case PUBLISH:
|
||||
case PUBREC:
|
||||
case PUBCOMP:
|
||||
case PUBREL:
|
||||
case DISCONNECT:
|
||||
m_messaging.handleProtocolMessage(session, msg);
|
||||
break;
|
||||
case PINGREQ:
|
||||
PingRespMessage pingResp = new PingRespMessage();
|
||||
session.write(pingResp);
|
||||
break;
|
||||
}
|
||||
} catch (Exception ex) {
|
||||
LOG.error("Bad error in processing the message", ex);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sessionIdle(IoSession session, IdleStatus status) {
|
||||
if (status == IdleStatus.READER_IDLE) {
|
||||
session.close(false);
|
||||
//TODO send a notification to messaging part to remove the bining clientID-ConnConfig
|
||||
}
|
||||
}
|
||||
|
||||
public void setMessaging(IMessaging messaging) {
|
||||
m_messaging = messaging;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,137 +1,136 @@
|
||||
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.moquette.messaging.spi.impl.SimpleMessaging;
|
||||
import org.dna.mqtt.moquette.proto.*;
|
||||
import org.dna.mqtt.moquette.proto.messages.*;
|
||||
import org.dna.mqtt.commons.Constants;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
/**
|
||||
* Launch a configured version of the server.
|
||||
* @author andrea
|
||||
*/
|
||||
public class Server {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Server.class);
|
||||
|
||||
// public static final 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;
|
||||
SimpleMessaging messaging;
|
||||
// Thread messagingEventLoop;
|
||||
//private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/;
|
||||
|
||||
public static void main(String[] args) throws IOException {
|
||||
new Server().startServer();
|
||||
|
||||
}
|
||||
|
||||
protected 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();
|
||||
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");
|
||||
|
||||
//Bind a shutdown hook
|
||||
Runtime.getRuntime().addShutdownHook(new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
stopServer();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
protected 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();
|
||||
LOG.info("Server stopped");
|
||||
}
|
||||
|
||||
}
|
||||
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.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
/**
|
||||
* Launch a configured version of the server.
|
||||
* @author andrea
|
||||
*/
|
||||
public class Server {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Server.class);
|
||||
|
||||
// public static final 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;
|
||||
SimpleMessaging messaging;
|
||||
// Thread messagingEventLoop;
|
||||
//private ExecutorService m_notifierPool/* = Executors.newFixedThreadPool(NOTIFIER_POOL_SIZE)*/;
|
||||
|
||||
public static void main(String[] args) throws IOException {
|
||||
new Server().startServer();
|
||||
|
||||
}
|
||||
|
||||
protected 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();
|
||||
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");
|
||||
|
||||
//Bind a shutdown hook
|
||||
Runtime.getRuntime().addShutdownHook(new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
stopServer();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
protected 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();
|
||||
LOG.info("Server stopped");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,57 +1,63 @@
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
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.PublishMessage;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PublishDecoder extends MqttDecoder {
|
||||
|
||||
public MessageDecoderResult decodable(IoSession session, IoBuffer in) {
|
||||
return Utils.checkDecodable(AbstractMessage.PUBLISH, in);
|
||||
}
|
||||
|
||||
public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
|
||||
int startPos = in.position();
|
||||
|
||||
//Common decoding part
|
||||
PublishMessage message = new PublishMessage();
|
||||
if (decodeCommonHeader(message, in) == NEED_DATA) {
|
||||
return NEED_DATA;
|
||||
}
|
||||
int remainingLength = message.getRemainingLength();
|
||||
|
||||
//Topic name
|
||||
String topic = Utils.decodeString(in);
|
||||
if (topic == null) {
|
||||
return NEED_DATA;
|
||||
}
|
||||
message.setTopicName(topic);
|
||||
|
||||
if (message.getQos() == QOSType.LEAST_ONE ||
|
||||
message.getQos() == QOSType.EXACTLY_ONCE) {
|
||||
message.setMessageID(Utils.readWord(in));
|
||||
}
|
||||
int stopPos = in.position();
|
||||
|
||||
//read the payload
|
||||
int payloadSize = remainingLength - (stopPos - startPos - 2);
|
||||
byte[] b = new byte[payloadSize];
|
||||
if (in.remaining() < payloadSize) {
|
||||
return NEED_DATA;
|
||||
}
|
||||
in.get(b);
|
||||
message.setPayload(b);
|
||||
|
||||
out.write(message);
|
||||
return OK;
|
||||
}
|
||||
|
||||
}
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
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.PublishMessage;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PublishDecoder extends MqttDecoder {
|
||||
|
||||
private static Logger LOG = LoggerFactory.getLogger(PublishDecoder.class);
|
||||
|
||||
public MessageDecoderResult decodable(IoSession session, IoBuffer in) {
|
||||
return Utils.checkDecodable(AbstractMessage.PUBLISH, in);
|
||||
}
|
||||
|
||||
public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
|
||||
LOG.debug("decode invoked with buffer " + in);
|
||||
int startPos = in.position();
|
||||
|
||||
//Common decoding part
|
||||
PublishMessage message = new PublishMessage();
|
||||
if (decodeCommonHeader(message, in) == NEED_DATA) {
|
||||
LOG.info("decode ask for more data after " + in);
|
||||
return NEED_DATA;
|
||||
}
|
||||
int remainingLength = message.getRemainingLength();
|
||||
|
||||
//Topic name
|
||||
String topic = Utils.decodeString(in);
|
||||
if (topic == null) {
|
||||
return NEED_DATA;
|
||||
}
|
||||
message.setTopicName(topic);
|
||||
|
||||
if (message.getQos() == QOSType.LEAST_ONE ||
|
||||
message.getQos() == QOSType.EXACTLY_ONCE) {
|
||||
message.setMessageID(Utils.readWord(in));
|
||||
}
|
||||
int stopPos = in.position();
|
||||
|
||||
//read the payload
|
||||
int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1);
|
||||
if (in.remaining() < payloadSize) {
|
||||
return NEED_DATA;
|
||||
}
|
||||
byte[] b = new byte[payloadSize];
|
||||
in.get(b);
|
||||
message.setPayload(b);
|
||||
|
||||
out.write(message);
|
||||
return OK;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,48 +1,48 @@
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import org.apache.mina.core.buffer.IoBuffer;
|
||||
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.AbstractMessage.QOSType;
|
||||
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PublishEncoder implements MessageEncoder<PublishMessage> {
|
||||
|
||||
public void encode(IoSession session, PublishMessage message, ProtocolEncoderOutput out) throws Exception {
|
||||
if (message.getQos() == 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");
|
||||
}
|
||||
|
||||
IoBuffer variableHeaderBuff = IoBuffer.allocate(2).setAutoExpand(true);
|
||||
variableHeaderBuff.put(Utils.encodeString(message.getTopicName()));
|
||||
if (message.getQos() == QOSType.LEAST_ONE ||
|
||||
message.getQos() == QOSType.EXACTLY_ONCE ) {
|
||||
if (message.getMessageID() == null) {
|
||||
throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted");
|
||||
}
|
||||
Utils.writeWord(variableHeaderBuff, message.getMessageID());
|
||||
}
|
||||
variableHeaderBuff.put(message.getPayload());
|
||||
variableHeaderBuff.flip();
|
||||
int variableHeaderSize = variableHeaderBuff.remaining();
|
||||
|
||||
byte flags = Utils.encodeFlags(message);
|
||||
|
||||
IoBuffer buff = IoBuffer.allocate(2 + variableHeaderSize);
|
||||
buff.put((byte) (AbstractMessage.PUBLISH << 4 | flags));
|
||||
buff.put(Utils.encodeRemainingLength(variableHeaderSize));
|
||||
buff.put(variableHeaderBuff).flip();
|
||||
|
||||
out.write(buff);
|
||||
}
|
||||
|
||||
}
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import org.apache.mina.core.buffer.IoBuffer;
|
||||
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.AbstractMessage.QOSType;
|
||||
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PublishEncoder implements MessageEncoder<PublishMessage> {
|
||||
|
||||
public void encode(IoSession session, PublishMessage message, ProtocolEncoderOutput out) throws Exception {
|
||||
if (message.getQos() == 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");
|
||||
}
|
||||
|
||||
IoBuffer variableHeaderBuff = IoBuffer.allocate(2).setAutoExpand(true);
|
||||
variableHeaderBuff.put(Utils.encodeString(message.getTopicName()));
|
||||
if (message.getQos() == QOSType.LEAST_ONE ||
|
||||
message.getQos() == QOSType.EXACTLY_ONCE ) {
|
||||
if (message.getMessageID() == null) {
|
||||
throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted");
|
||||
}
|
||||
Utils.writeWord(variableHeaderBuff, message.getMessageID());
|
||||
}
|
||||
variableHeaderBuff.put(message.getPayload());
|
||||
variableHeaderBuff.flip();
|
||||
int variableHeaderSize = variableHeaderBuff.remaining();
|
||||
|
||||
byte flags = Utils.encodeFlags(message);
|
||||
|
||||
IoBuffer buff = IoBuffer.allocate(2 + variableHeaderSize).setAutoExpand(true);;
|
||||
buff.put((byte) (AbstractMessage.PUBLISH << 4 | flags));
|
||||
buff.put(Utils.encodeRemainingLength(variableHeaderSize));
|
||||
buff.put(variableHeaderBuff).flip();
|
||||
|
||||
out.write(buff);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,180 +1,191 @@
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import java.io.UnsupportedEncodingException;
|
||||
import org.apache.mina.core.buffer.IoBuffer;
|
||||
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* Common utils methodd used in codecs.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class Utils {
|
||||
|
||||
public static final int MAX_LENGTH_LIMIT = 268435455;
|
||||
|
||||
/**
|
||||
* Read 2 bytes from in buffer first MSB, and then LSB returning as int.
|
||||
*/
|
||||
static int readWord(IoBuffer in) {
|
||||
int msb = in.get() & 0x00FF; //remove sign extension due to casting
|
||||
int lsb = in.get() & 0x00FF;
|
||||
msb = (msb << 8) | lsb ;
|
||||
return msb;
|
||||
}
|
||||
|
||||
/**
|
||||
* Writes as 2 bytes the int value into buffer first MSB, and then LSB.
|
||||
*/
|
||||
static void writeWord(IoBuffer out, int value) {
|
||||
out.put((byte) ((value & 0xFF00) >> 8)); //msb
|
||||
out.put((byte) (value & 0x00FF)); //lsb
|
||||
}
|
||||
|
||||
/**
|
||||
* Decode the variable remaining lenght as defined in MQTT v3.1 specification
|
||||
* (section 2.1).
|
||||
*
|
||||
* @return the decoded length or -1 if needed more data to decode the length field.
|
||||
*/
|
||||
static int decodeRemainingLenght(IoBuffer in) {
|
||||
int multiplier = 1;
|
||||
int value = 0;
|
||||
byte digit;
|
||||
do {
|
||||
if (in.remaining() < 1) {
|
||||
return -1;
|
||||
}
|
||||
digit = in.get();
|
||||
value += (digit & 0x7F) * multiplier;
|
||||
multiplier *= 128;
|
||||
} while ((digit & 0x80) != 0);
|
||||
return value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Encode the value in the format defined in specification as variable length
|
||||
* array.
|
||||
*
|
||||
* @throws IllegalArgumentException if the value is not in the specification bounds
|
||||
* [0..268435455].
|
||||
*/
|
||||
static IoBuffer encodeRemainingLength(int value) throws IllegalAccessException {
|
||||
if (value > MAX_LENGTH_LIMIT || value < 0) {
|
||||
throw new IllegalAccessException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value);
|
||||
}
|
||||
|
||||
IoBuffer encoded = IoBuffer.allocate(4);
|
||||
byte digit;
|
||||
do {
|
||||
digit = (byte) (value % 128);
|
||||
value = value / 128;
|
||||
// if there are more digits to encode, set the top bit of this digit
|
||||
if (value > 0) {
|
||||
digit = (byte) (digit | 0x80);
|
||||
}
|
||||
encoded.put(digit);
|
||||
} while (value > 0);
|
||||
encoded.flip();
|
||||
return encoded;
|
||||
}
|
||||
|
||||
static MessageDecoderResult checkDecodable(byte type, IoBuffer in) {
|
||||
if (in.remaining() < 1) {
|
||||
return MessageDecoderResult.NEED_DATA;
|
||||
}
|
||||
byte h1 = in.get();
|
||||
byte messageType = (byte) ((h1 & 0x00F0) >> 4);
|
||||
|
||||
int remainingLength = Utils.decodeRemainingLenght(in);
|
||||
if (remainingLength == -1) {
|
||||
return MessageDecoderResult.NEED_DATA;
|
||||
}
|
||||
|
||||
//check remaining length
|
||||
if (in.remaining() < remainingLength) {
|
||||
return MessageDecoderResult.NEED_DATA;
|
||||
}
|
||||
|
||||
return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded
|
||||
* string content.
|
||||
*/
|
||||
static IoBuffer encodeString(String str) {
|
||||
IoBuffer out = IoBuffer.allocate(2).setAutoExpand(true);
|
||||
byte[] raw;
|
||||
try {
|
||||
raw = str.getBytes("UTF-8");
|
||||
//NB every Java platform has got UTF-8 encoding by default, so this
|
||||
//exception are never raised.
|
||||
} catch (UnsupportedEncodingException ex) {
|
||||
LoggerFactory.getLogger(ConnectEncoder.class).error(null, ex);
|
||||
return null;
|
||||
}
|
||||
Utils.writeWord(out, raw.length);
|
||||
out.put(raw).flip();
|
||||
return out;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Load a string from the given buffer, reading first the two bytes of len
|
||||
* and then the UTF-8 bytes of the string.
|
||||
*
|
||||
* @return the decoded string or null if NEED_DATA
|
||||
*/
|
||||
static String decodeString(IoBuffer in) throws UnsupportedEncodingException {
|
||||
if (in.remaining() < 2) {
|
||||
return null;
|
||||
}
|
||||
int strLen = Utils.readWord(in);
|
||||
if (in.remaining() < strLen) {
|
||||
return null;
|
||||
}
|
||||
byte[] strRaw = new byte[strLen];
|
||||
in.get(strRaw);
|
||||
|
||||
return new String(strRaw, "UTF-8");
|
||||
}
|
||||
|
||||
static byte encodeFlags(AbstractMessage message) {
|
||||
byte flags = 0;
|
||||
if (message.isDupFlag()) {
|
||||
flags |= 0x08;
|
||||
}
|
||||
if (message.isRetainFlag()) {
|
||||
flags |= 0x01;
|
||||
}
|
||||
|
||||
flags |= ((message.getQos().ordinal() & 0x03) << 1);
|
||||
return flags;
|
||||
}
|
||||
|
||||
/**
|
||||
* Converts MQTT message type to a textual description.
|
||||
* */
|
||||
public static String msgType2String(int type) {
|
||||
switch (type) {
|
||||
case AbstractMessage.CONNECT: return "CONNECT";
|
||||
case AbstractMessage.CONNACK: return "CONNACK";
|
||||
case AbstractMessage.PUBLISH: return "PUBLISH";
|
||||
case AbstractMessage.PUBACK: return "PUBACK";
|
||||
case AbstractMessage.PUBREC: return "PUBREC";
|
||||
case AbstractMessage.PUBREL: return "PUBREL";
|
||||
case AbstractMessage.PUBCOMP: return "PUBCOMP";
|
||||
case AbstractMessage.SUBSCRIBE: return "SUBSCRIBE";
|
||||
case AbstractMessage.SUBACK: return "SUBACK";
|
||||
case AbstractMessage.UNSUBSCRIBE: return "UNSUBSCRIBE";
|
||||
case AbstractMessage.UNSUBACK: return "UNSUBACK";
|
||||
case AbstractMessage.PINGREQ: return "PINGREQ";
|
||||
case AbstractMessage.PINGRESP: return "PINGRESP";
|
||||
case AbstractMessage.DISCONNECT: return "DISCONNECT";
|
||||
default: throw new RuntimeException("Can't decode message type " + type);
|
||||
}
|
||||
}
|
||||
}
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import java.io.UnsupportedEncodingException;
|
||||
import org.apache.mina.core.buffer.IoBuffer;
|
||||
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* Common utils methodd used in codecs.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class Utils {
|
||||
|
||||
public static final int MAX_LENGTH_LIMIT = 268435455;
|
||||
|
||||
/**
|
||||
* Read 2 bytes from in buffer first MSB, and then LSB returning as int.
|
||||
*/
|
||||
static int readWord(IoBuffer in) {
|
||||
int msb = in.get() & 0x00FF; //remove sign extension due to casting
|
||||
int lsb = in.get() & 0x00FF;
|
||||
msb = (msb << 8) | lsb ;
|
||||
return msb;
|
||||
}
|
||||
|
||||
/**
|
||||
* Writes as 2 bytes the int value into buffer first MSB, and then LSB.
|
||||
*/
|
||||
static void writeWord(IoBuffer out, int value) {
|
||||
out.put((byte) ((value & 0xFF00) >> 8)); //msb
|
||||
out.put((byte) (value & 0x00FF)); //lsb
|
||||
}
|
||||
|
||||
/**
|
||||
* Decode the variable remaining lenght as defined in MQTT v3.1 specification
|
||||
* (section 2.1).
|
||||
*
|
||||
* @return the decoded length or -1 if needed more data to decode the length field.
|
||||
*/
|
||||
static int decodeRemainingLenght(IoBuffer in) {
|
||||
int multiplier = 1;
|
||||
int value = 0;
|
||||
byte digit;
|
||||
do {
|
||||
if (in.remaining() < 1) {
|
||||
return -1;
|
||||
}
|
||||
digit = in.get();
|
||||
value += (digit & 0x7F) * multiplier;
|
||||
multiplier *= 128;
|
||||
} while ((digit & 0x80) != 0);
|
||||
return value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the number of bytes to encode the gicen remaining length value
|
||||
*/
|
||||
static int numBytesToEncode(int len) {
|
||||
if (0 <= len && len <= 127) return 1;
|
||||
if (128 <= len && len <= 16383) return 2;
|
||||
if (16384 <= len && len <= 2097151) return 3;
|
||||
if (2097152 <= len && len <= 268435455) return 4;
|
||||
throw new IllegalArgumentException("value shoul be in the range [0..268435455]");
|
||||
}
|
||||
|
||||
/**
|
||||
* Encode the value in the format defined in specification as variable length
|
||||
* array.
|
||||
*
|
||||
* @throws IllegalArgumentException if the value is not in the specification bounds
|
||||
* [0..268435455].
|
||||
*/
|
||||
static IoBuffer encodeRemainingLength(int value) throws IllegalAccessException {
|
||||
if (value > MAX_LENGTH_LIMIT || value < 0) {
|
||||
throw new IllegalAccessException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value);
|
||||
}
|
||||
|
||||
IoBuffer encoded = IoBuffer.allocate(4);
|
||||
byte digit;
|
||||
do {
|
||||
digit = (byte) (value % 128);
|
||||
value = value / 128;
|
||||
// if there are more digits to encode, set the top bit of this digit
|
||||
if (value > 0) {
|
||||
digit = (byte) (digit | 0x80);
|
||||
}
|
||||
encoded.put(digit);
|
||||
} while (value > 0);
|
||||
encoded.flip();
|
||||
return encoded;
|
||||
}
|
||||
|
||||
static MessageDecoderResult checkDecodable(byte type, IoBuffer in) {
|
||||
if (in.remaining() < 1) {
|
||||
return MessageDecoderResult.NEED_DATA;
|
||||
}
|
||||
byte h1 = in.get();
|
||||
byte messageType = (byte) ((h1 & 0x00F0) >> 4);
|
||||
|
||||
int remainingLength = Utils.decodeRemainingLenght(in);
|
||||
if (remainingLength == -1) {
|
||||
return MessageDecoderResult.NEED_DATA;
|
||||
}
|
||||
|
||||
//check remaining length
|
||||
if (in.remaining() < remainingLength) {
|
||||
return MessageDecoderResult.NEED_DATA;
|
||||
}
|
||||
|
||||
return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded
|
||||
* string content.
|
||||
*/
|
||||
static IoBuffer encodeString(String str) {
|
||||
IoBuffer out = IoBuffer.allocate(2).setAutoExpand(true);
|
||||
byte[] raw;
|
||||
try {
|
||||
raw = str.getBytes("UTF-8");
|
||||
//NB every Java platform has got UTF-8 encoding by default, so this
|
||||
//exception are never raised.
|
||||
} catch (UnsupportedEncodingException ex) {
|
||||
LoggerFactory.getLogger(ConnectEncoder.class).error(null, ex);
|
||||
return null;
|
||||
}
|
||||
Utils.writeWord(out, raw.length);
|
||||
out.put(raw).flip();
|
||||
return out;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Load a string from the given buffer, reading first the two bytes of len
|
||||
* and then the UTF-8 bytes of the string.
|
||||
*
|
||||
* @return the decoded string or null if NEED_DATA
|
||||
*/
|
||||
static String decodeString(IoBuffer in) throws UnsupportedEncodingException {
|
||||
if (in.remaining() < 2) {
|
||||
return null;
|
||||
}
|
||||
int strLen = Utils.readWord(in);
|
||||
if (in.remaining() < strLen) {
|
||||
return null;
|
||||
}
|
||||
byte[] strRaw = new byte[strLen];
|
||||
in.get(strRaw);
|
||||
|
||||
return new String(strRaw, "UTF-8");
|
||||
}
|
||||
|
||||
static byte encodeFlags(AbstractMessage message) {
|
||||
byte flags = 0;
|
||||
if (message.isDupFlag()) {
|
||||
flags |= 0x08;
|
||||
}
|
||||
if (message.isRetainFlag()) {
|
||||
flags |= 0x01;
|
||||
}
|
||||
|
||||
flags |= ((message.getQos().ordinal() & 0x03) << 1);
|
||||
return flags;
|
||||
}
|
||||
|
||||
/**
|
||||
* Converts MQTT message type to a textual description.
|
||||
* */
|
||||
public static String msgType2String(int type) {
|
||||
switch (type) {
|
||||
case AbstractMessage.CONNECT: return "CONNECT";
|
||||
case AbstractMessage.CONNACK: return "CONNACK";
|
||||
case AbstractMessage.PUBLISH: return "PUBLISH";
|
||||
case AbstractMessage.PUBACK: return "PUBACK";
|
||||
case AbstractMessage.PUBREC: return "PUBREC";
|
||||
case AbstractMessage.PUBREL: return "PUBREL";
|
||||
case AbstractMessage.PUBCOMP: return "PUBCOMP";
|
||||
case AbstractMessage.SUBSCRIBE: return "SUBSCRIBE";
|
||||
case AbstractMessage.SUBACK: return "SUBACK";
|
||||
case AbstractMessage.UNSUBSCRIBE: return "UNSUBSCRIBE";
|
||||
case AbstractMessage.UNSUBACK: return "UNSUBACK";
|
||||
case AbstractMessage.PINGREQ: return "PINGREQ";
|
||||
case AbstractMessage.PINGRESP: return "PINGRESP";
|
||||
case AbstractMessage.DISCONNECT: return "DISCONNECT";
|
||||
default: throw new RuntimeException("Can't decode message type " + type);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,17 +1,19 @@
|
||||
# Set root logger level to DEBUG and its only appender to A1.
|
||||
log4j.rootLogger=ERROR, A1
|
||||
#log4j.rootLogger=DEBUG, A1
|
||||
#log4j.rootLogger=WARN, A1
|
||||
#log4j.logger.org.dna.mqtt.moquette.client=WARN
|
||||
#log4j.logger.org.mina=WARN
|
||||
#log4j.logger.org.dna.mqtt.moquette.server=DEBUG
|
||||
#log4j.logger.org.dna.mqtt.moquette.messaging=DEBUG
|
||||
|
||||
log4j.logger.org.dna.mqtt.bechnmark=INFO
|
||||
|
||||
# A1 is set to be a ConsoleAppender.
|
||||
log4j.appender.A1=org.apache.log4j.ConsoleAppender
|
||||
|
||||
# A1 uses PatternLayout.
|
||||
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
|
||||
# Set root logger level to DEBUG and its only appender to A1.
|
||||
#log4j.rootLogger=ERROR, A1
|
||||
log4j.rootLogger=INFO, A1
|
||||
#log4j.rootLogger=WARN, A1
|
||||
#log4j.logger.org.dna.mqtt.moquette.client=WARN
|
||||
#log4j.logger.org.mina=WARN
|
||||
#log4j.logger.org.dna.mqtt.moquette.server=DEBUG
|
||||
#log4j.logger.org.dna.mqtt.moquette.messaging=DEBUG
|
||||
#log4j.logger.org.dna.mqtt.moquette.proto.MQTTLoggingFilter=DEBUG
|
||||
#log4j.logger.org.dna.mqtt.moquette.proto.PublishDecoder=DEBUG
|
||||
|
||||
log4j.logger.org.dna.mqtt.bechnmark=INFO
|
||||
|
||||
# A1 is set to be a ConsoleAppender.
|
||||
log4j.appender.A1=org.apache.log4j.ConsoleAppender
|
||||
|
||||
# A1 uses PatternLayout.
|
||||
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
|
||||
log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n
|
||||
@@ -1,132 +1,180 @@
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import org.apache.mina.core.buffer.IoBuffer;
|
||||
import org.apache.mina.filter.codec.demux.MessageDecoder;
|
||||
import org.apache.mina.filter.codec.demux.MessageDecoderResult;
|
||||
import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput;
|
||||
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.Test;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PublishDecoderTest {
|
||||
|
||||
IoBuffer m_buff;
|
||||
PublishDecoder m_msgdec;
|
||||
MockProtocolDecoderOutput<PublishMessage> m_mockProtoDecoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_buff = IoBuffer.allocate(4);
|
||||
m_msgdec = new PublishDecoder();
|
||||
m_mockProtoDecoder = new MockProtocolDecoderOutput<PublishMessage>();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDecodable_OK() {
|
||||
m_buff.put((byte) (AbstractMessage.PUBLISH << 4))
|
||||
.put((byte)0) //0 length
|
||||
.flip();
|
||||
assertEquals(MessageDecoder.OK, m_msgdec.decodable(null, m_buff));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHeader() throws Exception {
|
||||
m_buff = IoBuffer.allocate(14);
|
||||
initHeader(m_buff);
|
||||
m_buff.flip();
|
||||
|
||||
//Exercise
|
||||
MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder);
|
||||
|
||||
assertNotNull(m_mockProtoDecoder.getMessage());
|
||||
assertEquals(MessageDecoder.OK, res);
|
||||
assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName());
|
||||
assertNull(m_mockProtoDecoder.getMessage().getMessageID());
|
||||
assertEquals(AbstractMessage.PUBLISH, m_mockProtoDecoder.getMessage().getMessageType());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHeaderWithMessageID() throws Exception {
|
||||
m_buff = IoBuffer.allocate(14).setAutoExpand(true);
|
||||
int messageID = 123;
|
||||
initHeaderWithMessageID(m_buff, messageID);
|
||||
m_buff.flip();
|
||||
|
||||
//Exercise
|
||||
MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder);
|
||||
|
||||
assertNotNull(m_mockProtoDecoder.getMessage());
|
||||
assertEquals(MessageDecoder.OK, res);
|
||||
assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName());
|
||||
assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHeaderWithMessageID_Payload() throws Exception {
|
||||
m_buff = IoBuffer.allocate(14).setAutoExpand(true);
|
||||
int messageID = 123;
|
||||
byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
|
||||
initHeaderWithMessageID_Payload(m_buff, messageID, payload);
|
||||
m_buff.flip();
|
||||
|
||||
//Exercise
|
||||
MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder);
|
||||
|
||||
assertNotNull(m_mockProtoDecoder.getMessage());
|
||||
assertEquals(MessageDecoder.OK, res);
|
||||
assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName());
|
||||
assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID());
|
||||
TestUtils.verifyEquals(payload, m_mockProtoDecoder.getMessage().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 = IoBuffer.allocate(overallMessage.length).setAutoExpand(true);
|
||||
m_buff.put(overallMessage);
|
||||
m_buff.flip();
|
||||
|
||||
//Exercise
|
||||
MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder);
|
||||
|
||||
assertNotNull(m_mockProtoDecoder.getMessage());
|
||||
assertEquals(MessageDecoder.OK, res);
|
||||
}
|
||||
|
||||
private void initHeader(IoBuffer buff) throws IllegalAccessException {
|
||||
IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")).flip();
|
||||
buff.clear().put((byte) (AbstractMessage.PUBLISH << 4)).put(Utils.encodeRemainingLength(tmp.remaining()));
|
||||
//topic name
|
||||
buff.put(tmp);
|
||||
}
|
||||
|
||||
private void initHeaderWithMessageID(IoBuffer buff, int messageID) throws IllegalAccessException {
|
||||
IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic"));
|
||||
Utils.writeWord(tmp, messageID);
|
||||
tmp.flip();
|
||||
buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1
|
||||
.put(Utils.encodeRemainingLength(tmp.remaining()));
|
||||
//topic name
|
||||
buff.put(tmp);
|
||||
}
|
||||
|
||||
private void initHeaderWithMessageID_Payload(IoBuffer buff, int messageID, byte[] payload) throws IllegalAccessException {
|
||||
IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic"));
|
||||
Utils.writeWord(tmp, messageID);
|
||||
tmp.put(payload);
|
||||
tmp.flip();
|
||||
buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1
|
||||
.put(Utils.encodeRemainingLength(tmp.remaining()));
|
||||
//topic name
|
||||
buff.put(tmp);
|
||||
}
|
||||
}
|
||||
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;
|
||||
import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput;
|
||||
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;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PublishDecoderTest {
|
||||
|
||||
IoBuffer m_buff;
|
||||
PublishDecoder m_msgdec;
|
||||
MockProtocolDecoderOutput<PublishMessage> m_mockProtoDecoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_buff = IoBuffer.allocate(4);
|
||||
m_msgdec = new PublishDecoder();
|
||||
m_mockProtoDecoder = new MockProtocolDecoderOutput<PublishMessage>();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDecodable_OK() {
|
||||
m_buff.put((byte) (AbstractMessage.PUBLISH << 4))
|
||||
.put((byte) 0) //0 length
|
||||
.flip();
|
||||
assertEquals(MessageDecoder.OK, m_msgdec.decodable(null, m_buff));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHeader() throws Exception {
|
||||
m_buff = IoBuffer.allocate(14);
|
||||
initHeader(m_buff);
|
||||
m_buff.flip();
|
||||
|
||||
//Exercise
|
||||
MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder);
|
||||
|
||||
assertNotNull(m_mockProtoDecoder.getMessage());
|
||||
assertEquals(MessageDecoder.OK, res);
|
||||
assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName());
|
||||
assertNull(m_mockProtoDecoder.getMessage().getMessageID());
|
||||
assertEquals(AbstractMessage.PUBLISH, m_mockProtoDecoder.getMessage().getMessageType());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHeaderWithMessageID() throws Exception {
|
||||
m_buff = IoBuffer.allocate(14).setAutoExpand(true);
|
||||
int messageID = 123;
|
||||
initHeaderWithMessageID(m_buff, messageID);
|
||||
m_buff.flip();
|
||||
|
||||
//Exercise
|
||||
MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder);
|
||||
|
||||
assertNotNull(m_mockProtoDecoder.getMessage());
|
||||
assertEquals(MessageDecoder.OK, res);
|
||||
assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName());
|
||||
assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testHeaderWithMessageID_Payload() throws Exception {
|
||||
m_buff = IoBuffer.allocate(14).setAutoExpand(true);
|
||||
int messageID = 123;
|
||||
byte[] payload = new byte[]{0x0A, 0x0B, 0x0C};
|
||||
initHeaderWithMessageID_Payload(m_buff, messageID, payload);
|
||||
m_buff.flip();
|
||||
|
||||
//Exercise
|
||||
MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder);
|
||||
|
||||
assertNotNull(m_mockProtoDecoder.getMessage());
|
||||
assertEquals(MessageDecoder.OK, res);
|
||||
assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName());
|
||||
assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID());
|
||||
TestUtils.verifyEquals(payload, m_mockProtoDecoder.getMessage().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 = IoBuffer.allocate(overallMessage.length).setAutoExpand(true);
|
||||
m_buff.put(overallMessage);
|
||||
m_buff.flip();
|
||||
|
||||
//Exercise
|
||||
MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder);
|
||||
|
||||
assertNotNull(m_mockProtoDecoder.getMessage());
|
||||
assertEquals(MessageDecoder.OK, res);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDecodeBigContent() throws Exception {
|
||||
int size = 129;
|
||||
IoBuffer payload = TestUtils.generateRandomPayload(size);
|
||||
|
||||
IoBuffer firstPublish = generatePublishQoS0(payload);
|
||||
IoBuffer secondPublish = generatePublishQoS0(TestUtils.generateRandomPayload(size));
|
||||
|
||||
IoBuffer doubleMessageBuf = IoBuffer.allocate(size * 2).setAutoExpand(true);
|
||||
doubleMessageBuf.put(firstPublish).put(secondPublish).flip();
|
||||
|
||||
|
||||
//Exercise
|
||||
MessageDecoderResult res = m_msgdec.decode(null, doubleMessageBuf, m_mockProtoDecoder);
|
||||
|
||||
assertEquals(MessageDecoder.OK, res);
|
||||
PublishMessage pubMsg = m_mockProtoDecoder.getMessage();
|
||||
assertNotNull(pubMsg);
|
||||
|
||||
res = m_msgdec.decode(null, doubleMessageBuf, m_mockProtoDecoder);
|
||||
|
||||
assertNotNull(m_mockProtoDecoder.getMessage());
|
||||
assertEquals(MessageDecoder.OK, res);
|
||||
|
||||
m_buff.flip();
|
||||
}
|
||||
|
||||
private void initHeader(IoBuffer buff) throws IllegalAccessException {
|
||||
IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")).flip();
|
||||
buff.clear().put((byte) (AbstractMessage.PUBLISH << 4)).put(Utils.encodeRemainingLength(tmp.remaining()));
|
||||
//topic name
|
||||
buff.put(tmp);
|
||||
}
|
||||
|
||||
private void initHeaderWithMessageID(IoBuffer buff, int messageID) throws IllegalAccessException {
|
||||
IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic"));
|
||||
Utils.writeWord(tmp, messageID);
|
||||
tmp.flip();
|
||||
buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1
|
||||
.put(Utils.encodeRemainingLength(tmp.remaining()));
|
||||
//topic name
|
||||
buff.put(tmp);
|
||||
}
|
||||
|
||||
private void initHeaderWithMessageID_Payload(IoBuffer buff, int messageID, byte[] payload) throws IllegalAccessException {
|
||||
IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic"));
|
||||
Utils.writeWord(tmp, messageID);
|
||||
tmp.put(payload);
|
||||
tmp.flip();
|
||||
buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1
|
||||
.put(Utils.encodeRemainingLength(tmp.remaining()));
|
||||
//topic name
|
||||
buff.put(tmp);
|
||||
}
|
||||
|
||||
private IoBuffer generatePublishQoS0(IoBuffer payload) throws IllegalAccessException {
|
||||
int size = payload.capacity();
|
||||
IoBuffer messageBody = IoBuffer.allocate(size).setAutoExpand(true);
|
||||
messageBody.put(Utils.encodeString("/topic"));
|
||||
|
||||
//ONLY for QoS > 1 Utils.writeWord(messageBody, messageID);
|
||||
messageBody.put(payload).flip();
|
||||
|
||||
IoBuffer completeMsg = IoBuffer.allocate(size).setAutoExpand(true);
|
||||
completeMsg.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x00)) //set Qos to 0
|
||||
.put(Utils.encodeRemainingLength(messageBody.remaining()));
|
||||
completeMsg.put(messageBody).flip();
|
||||
|
||||
return completeMsg;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,145 +1,167 @@
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import org.dna.mqtt.moquette.proto.PublishEncoder;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
|
||||
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import static org.junit.Assert.*;
|
||||
import static org.dna.mqtt.moquette.proto.TestUtils.*;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PublishEncoderTest {
|
||||
|
||||
PublishEncoder m_encoder;
|
||||
TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_encoder = new PublishEncoder();
|
||||
m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput();
|
||||
}
|
||||
|
||||
@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(null, msg, m_mockProtoEncoder);
|
||||
|
||||
//Verify
|
||||
assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte
|
||||
//(2+7) topic + 3 payload
|
||||
assertEquals(12, m_mockProtoEncoder.getBuffer().get()); //remaining length
|
||||
|
||||
//Variable part
|
||||
verifyString(topic, m_mockProtoEncoder.getBuffer());
|
||||
verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer());
|
||||
}
|
||||
|
||||
@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(null, msg, m_mockProtoEncoder);
|
||||
|
||||
//Verify
|
||||
assertEquals(0x32, m_mockProtoEncoder.getBuffer().get()); //1 byte
|
||||
//(2+7) topic + 2 messageID + 3 payload
|
||||
assertEquals(14, m_mockProtoEncoder.getBuffer().get()); //remaining length
|
||||
|
||||
//Variable part
|
||||
verifyString(topic, m_mockProtoEncoder.getBuffer());
|
||||
assertEquals(0, m_mockProtoEncoder.getBuffer().get()); //MessageID MSB
|
||||
assertEquals(1, m_mockProtoEncoder.getBuffer().get()); //MessageID LSB
|
||||
verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer());
|
||||
}
|
||||
|
||||
@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(null, msg, m_mockProtoEncoder);
|
||||
}
|
||||
|
||||
@Test(expected = IllegalArgumentException.class)
|
||||
public void testEncode_empty_topic() throws Exception {
|
||||
PublishMessage msg = new PublishMessage();
|
||||
msg.setQos(QOSType.LEAST_ONE);
|
||||
msg.setMessageID(1);
|
||||
|
||||
//Exercise
|
||||
m_encoder.encode(null, msg, m_mockProtoEncoder);
|
||||
}
|
||||
|
||||
@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(null, msg, m_mockProtoEncoder);
|
||||
}
|
||||
|
||||
@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);
|
||||
|
||||
//Exercise
|
||||
m_encoder.encode(null, msg, m_mockProtoEncoder);
|
||||
}
|
||||
|
||||
@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(null, msg, m_mockProtoEncoder);
|
||||
|
||||
//Verify
|
||||
//2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload
|
||||
int size = m_mockProtoEncoder.getBuffer().remaining();
|
||||
assertEquals(10+payload.length, size);
|
||||
assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte
|
||||
assertEquals(23, m_mockProtoEncoder.getBuffer().get()); //1 byte the length
|
||||
}
|
||||
}
|
||||
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;
|
||||
import org.junit.Test;
|
||||
import static org.junit.Assert.*;
|
||||
import static org.dna.mqtt.moquette.proto.TestUtils.*;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PublishEncoderTest {
|
||||
|
||||
PublishEncoder m_encoder;
|
||||
TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_encoder = new PublishEncoder();
|
||||
m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput();
|
||||
}
|
||||
|
||||
@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(null, msg, m_mockProtoEncoder);
|
||||
|
||||
//Verify
|
||||
assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte
|
||||
//(2+7) topic + 3 payload
|
||||
assertEquals(12, m_mockProtoEncoder.getBuffer().get()); //remaining length
|
||||
|
||||
//Variable part
|
||||
verifyString(topic, m_mockProtoEncoder.getBuffer());
|
||||
verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer());
|
||||
}
|
||||
|
||||
@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(null, msg, m_mockProtoEncoder);
|
||||
|
||||
//Verify
|
||||
assertEquals(0x32, m_mockProtoEncoder.getBuffer().get()); //1 byte
|
||||
//(2+7) topic + 2 messageID + 3 payload
|
||||
assertEquals(14, m_mockProtoEncoder.getBuffer().get()); //remaining length
|
||||
|
||||
//Variable part
|
||||
verifyString(topic, m_mockProtoEncoder.getBuffer());
|
||||
assertEquals(0, m_mockProtoEncoder.getBuffer().get()); //MessageID MSB
|
||||
assertEquals(1, m_mockProtoEncoder.getBuffer().get()); //MessageID LSB
|
||||
verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer());
|
||||
}
|
||||
|
||||
@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(null, msg, m_mockProtoEncoder);
|
||||
}
|
||||
|
||||
@Test(expected = IllegalArgumentException.class)
|
||||
public void testEncode_empty_topic() throws Exception {
|
||||
PublishMessage msg = new PublishMessage();
|
||||
msg.setQos(QOSType.LEAST_ONE);
|
||||
msg.setMessageID(1);
|
||||
|
||||
//Exercise
|
||||
m_encoder.encode(null, msg, m_mockProtoEncoder);
|
||||
}
|
||||
|
||||
@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(null, msg, m_mockProtoEncoder);
|
||||
}
|
||||
|
||||
@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);
|
||||
|
||||
//Exercise
|
||||
m_encoder.encode(null, msg, m_mockProtoEncoder);
|
||||
}
|
||||
|
||||
@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(null, msg, m_mockProtoEncoder);
|
||||
|
||||
//Verify
|
||||
//2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload
|
||||
int size = m_mockProtoEncoder.getBuffer().remaining();
|
||||
assertEquals(10+payload.length, size);
|
||||
assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte
|
||||
assertEquals(23, m_mockProtoEncoder.getBuffer().get()); //1 byte the length
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testEncodeBigMessage() throws Exception {
|
||||
byte[] payload = new byte[256];
|
||||
|
||||
TestUtils.generateRandomPayload(256).get(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(null, msg, m_mockProtoEncoder);
|
||||
|
||||
int size = m_mockProtoEncoder.getBuffer().remaining();
|
||||
assertEquals(11+payload.length, size);
|
||||
assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,97 +1,124 @@
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import java.io.UnsupportedEncodingException;
|
||||
import org.apache.mina.core.buffer.IoBuffer;
|
||||
import org.apache.mina.core.filterchain.IoFilter.NextFilter;
|
||||
import org.apache.mina.core.future.WriteFuture;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.apache.mina.filter.codec.ProtocolDecoderOutput;
|
||||
import org.apache.mina.filter.codec.ProtocolEncoderOutput;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class TestUtils {
|
||||
|
||||
static final class MockProtocolDecoderOutput<T extends AbstractMessage> implements ProtocolDecoderOutput {
|
||||
|
||||
protected T m_connMessage;
|
||||
|
||||
public T getMessage() {
|
||||
return m_connMessage;
|
||||
}
|
||||
|
||||
public void write(Object message) {
|
||||
m_connMessage = (T) message;
|
||||
}
|
||||
|
||||
public void flush(NextFilter nextFilter, IoSession session) {
|
||||
throw new UnsupportedOperationException("Not supported yet.");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
static final class MockProtocolEncoderOutput implements ProtocolEncoderOutput {
|
||||
|
||||
protected IoBuffer m_buffer;
|
||||
|
||||
public IoBuffer getBuffer() {
|
||||
return m_buffer;
|
||||
}
|
||||
|
||||
public void write(Object buffer) {
|
||||
m_buffer = (IoBuffer) buffer;
|
||||
}
|
||||
|
||||
public void mergeAll() {
|
||||
throw new UnsupportedOperationException("Not supported yet.");
|
||||
}
|
||||
|
||||
public WriteFuture flush() {
|
||||
throw new UnsupportedOperationException("Not supported yet.");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Verify the presence of the given string starting from the current position
|
||||
* inside the buffer.
|
||||
*/
|
||||
static void verifyString(String str, IoBuffer buff) throws UnsupportedEncodingException {
|
||||
IoBuffer tmpBuff = IoBuffer.allocate(2).setAutoExpand(true);
|
||||
byte[] raw = str.getBytes("UTF-8");
|
||||
Utils.writeWord(tmpBuff, raw.length);
|
||||
tmpBuff.put(raw).flip();
|
||||
|
||||
verifyIoBuffer(tmpBuff, buff);
|
||||
}
|
||||
|
||||
static void verifyIoBuffer(IoBuffer expected, IoBuffer found) {
|
||||
while(expected.hasRemaining()) {
|
||||
assertEquals(expected.get(), found.get());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Verify that the given bytes buffer og the given numBytes length is present
|
||||
* in the buff starting from the current position.
|
||||
*/
|
||||
static void verifyBuff(int numBytes, byte[] bytes, IoBuffer buff) {
|
||||
assertTrue(numBytes <= buff.remaining());
|
||||
byte[] toCheck = new byte[numBytes];
|
||||
buff.get(toCheck);
|
||||
|
||||
for (int i = 0; i < numBytes; i++) {
|
||||
assertEquals(bytes[i], toCheck[i]);
|
||||
}
|
||||
}
|
||||
|
||||
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]);
|
||||
}
|
||||
}
|
||||
}
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.FileInputStream;
|
||||
import java.io.FileNotFoundException;
|
||||
import java.io.IOException;
|
||||
import java.io.UnsupportedEncodingException;
|
||||
import org.apache.mina.core.buffer.IoBuffer;
|
||||
import org.apache.mina.core.filterchain.IoFilter.NextFilter;
|
||||
import org.apache.mina.core.future.WriteFuture;
|
||||
import org.apache.mina.core.session.IoSession;
|
||||
import org.apache.mina.filter.codec.ProtocolDecoderOutput;
|
||||
import org.apache.mina.filter.codec.ProtocolEncoderOutput;
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class TestUtils {
|
||||
|
||||
static final class MockProtocolDecoderOutput<T extends AbstractMessage> implements ProtocolDecoderOutput {
|
||||
|
||||
protected T m_connMessage;
|
||||
|
||||
public T getMessage() {
|
||||
return m_connMessage;
|
||||
}
|
||||
|
||||
public void write(Object message) {
|
||||
m_connMessage = (T) message;
|
||||
}
|
||||
|
||||
public void flush(NextFilter nextFilter, IoSession session) {
|
||||
throw new UnsupportedOperationException("Not supported yet.");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
static final class MockProtocolEncoderOutput implements ProtocolEncoderOutput {
|
||||
|
||||
protected IoBuffer m_buffer;
|
||||
|
||||
public IoBuffer getBuffer() {
|
||||
return m_buffer;
|
||||
}
|
||||
|
||||
public void write(Object buffer) {
|
||||
m_buffer = (IoBuffer) buffer;
|
||||
}
|
||||
|
||||
public void mergeAll() {
|
||||
throw new UnsupportedOperationException("Not supported yet.");
|
||||
}
|
||||
|
||||
public WriteFuture flush() {
|
||||
throw new UnsupportedOperationException("Not supported yet.");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Verify the presence of the given string starting from the current position
|
||||
* inside the buffer.
|
||||
*/
|
||||
static void verifyString(String str, IoBuffer buff) throws UnsupportedEncodingException {
|
||||
IoBuffer tmpBuff = IoBuffer.allocate(2).setAutoExpand(true);
|
||||
byte[] raw = str.getBytes("UTF-8");
|
||||
Utils.writeWord(tmpBuff, raw.length);
|
||||
tmpBuff.put(raw).flip();
|
||||
|
||||
verifyIoBuffer(tmpBuff, buff);
|
||||
}
|
||||
|
||||
static void verifyIoBuffer(IoBuffer expected, IoBuffer found) {
|
||||
while(expected.hasRemaining()) {
|
||||
assertEquals(expected.get(), found.get());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Verify that the given bytes buffer og the given numBytes length is present
|
||||
* in the buff starting from the current position.
|
||||
*/
|
||||
static void verifyBuff(int numBytes, byte[] bytes, IoBuffer buff) {
|
||||
assertTrue(numBytes <= buff.remaining());
|
||||
byte[] toCheck = new byte[numBytes];
|
||||
buff.get(toCheck);
|
||||
|
||||
for (int i = 0; i < numBytes; i++) {
|
||||
assertEquals(bytes[i], toCheck[i]);
|
||||
}
|
||||
}
|
||||
|
||||
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 IoBuffer generateRandomPayload(int size) {
|
||||
IoBuffer payloadBuffer = IoBuffer.allocate(size);
|
||||
for (int i = 0; i < size; i++) {
|
||||
payloadBuffer.put((byte) (Math.random() * 255));
|
||||
}
|
||||
payloadBuffer.flip();
|
||||
return payloadBuffer;
|
||||
}
|
||||
|
||||
public static IoBuffer loadFile(String path) throws FileNotFoundException, IOException {
|
||||
File f = new File(path);
|
||||
long size = f.length();
|
||||
byte[] payload = new byte[(int) size];
|
||||
|
||||
FileInputStream fr = new FileInputStream(f);
|
||||
fr.read(payload);
|
||||
|
||||
IoBuffer payloadBuffer = IoBuffer.allocate((int) size);
|
||||
payloadBuffer.put(payload).flip();
|
||||
return payloadBuffer;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,26 +1,25 @@
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class UtilsTests {
|
||||
|
||||
@Test
|
||||
public void testEncodeFlags() {
|
||||
UnsubscribeMessage msg = new UnsubscribeMessage();
|
||||
msg.setRetainFlag(true);
|
||||
msg.setQos(AbstractMessage.QOSType.MOST_ONE);
|
||||
|
||||
//Exercise
|
||||
assertEquals(1, Utils.encodeFlags(msg));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
package org.dna.mqtt.moquette.proto;
|
||||
|
||||
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
|
||||
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
|
||||
import static org.junit.Assert.*;
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class UtilsTests {
|
||||
|
||||
@Test
|
||||
public void testEncodeFlags() {
|
||||
UnsubscribeMessage msg = new UnsubscribeMessage();
|
||||
msg.setRetainFlag(true);
|
||||
msg.setQos(AbstractMessage.QOSType.MOST_ONE);
|
||||
|
||||
//Exercise
|
||||
assertEquals(1, Utils.encodeFlags(msg));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user