From 65e34ddc40d4157418d7b8cc01c1db05fec611a6 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Fri, 31 Aug 2012 19:23:53 +0200 Subject: [PATCH] Added QoS 1 pub ack to the server part --- .../moquette/messaging/spi/IMessaging.java | 6 ++ .../moquette/messaging/spi/INotifier.java | 9 ++- .../messaging/spi/impl/SimpleMessaging.java | 21 ++++-- .../spi/impl/events/NotifyEvent.java | 13 ++++ .../spi/impl/events/PubAckEvent.java | 24 +++++++ .../spi/impl/events/PublishEvent.java | 10 +-- .../dna/mqtt/moquette/server/MQTTHandler.java | 65 ++++++++++++++----- .../dna/mqtt/moquette/server/Notifier.java | 9 ++- .../org/dna/mqtt/moquette/server/Server.java | 20 +----- .../server/ServerIntegrationTest.java | 25 +++++++ 10 files changed, 152 insertions(+), 50 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index fe762252..cacf9da1 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -2,6 +2,7 @@ package org.dna.mqtt.moquette.messaging.spi; import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; /** @@ -18,6 +19,11 @@ public interface IMessaging { * been notified. */ void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID); + + /** + * Used to publish messages with QoS > 0 + * */ + void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID); /** * Subscribe a client to a specified topic with a defined level diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java index 13685f38..5196ca2e 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java @@ -1,7 +1,10 @@ package org.dna.mqtt.moquette.messaging.spi; import org.apache.mina.core.session.IoSession; +import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PubAckMessage; /** * @@ -9,8 +12,10 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; */ public interface INotifier { - public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained); + public void notify(NotifyEvent evt/*String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained*/); public void disconnect(IoSession session); - + + void sendPubAck(PubAckEvent evt); + } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 44a69fb4..05c27b8c 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -119,6 +119,14 @@ public class SimpleMessaging implements IMessaging, Runnable { } } + public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID) { + try { + m_inboundQueue.put(new PublishEvent(topic, qos, message, retain, clientID, messageID)); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } + } + public void subscribe(String clientId, String topic, QOSType qos) { Subscription newSubscription = new Subscription(clientId, topic, qos); try { @@ -151,8 +159,7 @@ public class SimpleMessaging implements IMessaging, Runnable { public void refill(MessagingEvent evt) throws InterruptedException { m_inboundQueue.put(evt); } - - + /** * Verify if the 2 topics matching respecting the rules of MQTT Appendix A */ @@ -233,6 +240,7 @@ public class SimpleMessaging implements IMessaging, Runnable { } else if (evt instanceof CleanInFlightEvent) { //remove the message from inflight storage m_inflightStore.remove(((CleanInFlightEvent)evt).getMsgId()); + } } catch (InterruptedException ex) { processClose(); @@ -252,17 +260,22 @@ public class SimpleMessaging implements IMessaging, Runnable { if (qos == QOSType.LEAST_ONE) { //store the temporary message - String publishKey = evt.getClientID() + evt.getMessageID(); + String publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); m_inflightStore.put(publishKey, evt); cleanEvt = new CleanInFlightEvent(publishKey); } for (final Subscription sub : subscriptions.matches(topic)) { - m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); + if (qos == QOSType.MOST_ONE) { + m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false)); + } else { + m_outboundQueue.put(new NotifyEvent(sub.clientId, topic, qos, message, false, evt.getMessageID())); + } } if (cleanEvt != null) { m_outboundQueue.put(cleanEvt); + m_outboundQueue.put(new PubAckEvent(evt.getMessageID(), evt.getClientID())); } if (retain) { diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java index 4fe6dc2e..ddbb138a 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java @@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; /** + * TODO it's identical to PublishEvent * * @author andrea */ @@ -13,6 +14,9 @@ public class NotifyEvent extends MessagingEvent { private QOSType m_qos; private byte[] m_message; private boolean m_retaned; + + //Optional attribute, available only fo QoS 1 and 2 + int m_msgID; public NotifyEvent(String clientId, String topic, QOSType qos, byte[] message, boolean retained) { @@ -23,6 +27,11 @@ public class NotifyEvent extends MessagingEvent { m_retaned = retained; } + public NotifyEvent(String clientId, String topic, QOSType qos, byte[] message, boolean retained, int msgID) { + this(clientId, topic, qos, message, retained); + m_msgID = msgID; + } + public String getClientId() { return m_clientId; } @@ -42,5 +51,9 @@ public class NotifyEvent extends MessagingEvent { public boolean isRetained() { return m_retaned; } + + public int getMessageID() { + return m_msgID; + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java new file mode 100644 index 00000000..2197775d --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java @@ -0,0 +1,24 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +/** + * Used to send the ack message back to the client after a publish + */ +public class PubAckEvent extends MessagingEvent { + + int m_messageId; + + String m_clientID; + + public PubAckEvent(int messageID, String clientID) { + m_messageId = messageID ; + m_clientID = clientID; + } + + public int getMessageId() { + return m_messageId; + } + + public String getClientID() { + return m_clientID; + } +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java index 86a9954b..6dbf1e5a 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java @@ -2,18 +2,20 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import java.io.Serializable; + /** * * @author andrea */ -public class PublishEvent extends MessagingEvent { +public class PublishEvent extends MessagingEvent implements Serializable { String m_topic; QOSType m_qos; byte[] m_message; boolean m_retain; String m_clientID; //Optional attribute, available only fo QoS 1 and 2 - String m_msgID; + int m_msgID; public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, String clientID) { @@ -25,7 +27,7 @@ public class PublishEvent extends MessagingEvent { } public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, - String clientID, String msgID) { + String clientID, int msgID) { this(topic, qos, message, retain, clientID); m_msgID = msgID; } @@ -50,7 +52,7 @@ public class PublishEvent extends MessagingEvent { return m_clientID; } - public String getMessageID() { + public int getMessageID() { return m_msgID; } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index 8ae838ba..d27cdb59 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -9,18 +9,12 @@ 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.messaging.spi.INotifier; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; +import org.dna.mqtt.moquette.proto.messages.*; + import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; -import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; -import org.dna.mqtt.moquette.proto.messages.ConnectMessage; -import org.dna.mqtt.moquette.proto.messages.DisconnectMessage; -import org.dna.mqtt.moquette.proto.messages.PingRespMessage; -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.proto.messages.UnsubAckMessage; -import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -179,14 +173,20 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { UnsubAckMessage ackMessage = new UnsubAckMessage(); ackMessage.setMessageID(msg.getMessageID()); - LOG.info("replying with UnsubAct to MSG ID {0}", msg.getMessageID()); + LOG.info("replying with UnsubAck to MSG ID {0}", msg.getMessageID()); session.write(ackMessage); } protected void handlePublish(IoSession session, PublishMessage message) { String clientID = (String) session.getAttribute(ATTR_CLIENTID); - m_messaging.publish(message.getTopicName(), message.getPayload(), - message.getQos(), message.isRetainFlag(), clientID); + + if (message.getQos() == QOSType.MOST_ONE) { + m_messaging.publish(message.getTopicName(), message.getPayload(), + message.getQos(), message.isRetainFlag(), clientID); + } else { + m_messaging.publish(message.getTopicName(), message.getPayload(), + message.getQos(), message.isRetainFlag(), clientID, message.getMessageID()); + } } protected void handleDisconnect(IoSession session, DisconnectMessage disconnectMessage) { @@ -222,12 +222,19 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { m_authenticator = authenticator; } - public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained) { + + + public void notify(NotifyEvent evt) { + LOG.debug("notify invoked with event " + evt); + String clientId = evt.getClientId(); PublishMessage pubMessage = new PublishMessage(); - pubMessage.setRetainFlag(retained); - pubMessage.setTopicName(topic); - pubMessage.setQos(qOSType); - pubMessage.setPayload(payload); + pubMessage.setRetainFlag(evt.isRetained()); + pubMessage.setTopicName(evt.getTopic()); + pubMessage.setQos(evt.getQos()); + pubMessage.setPayload(evt.getMessage()); + if (pubMessage.getQos() != QOSType.MOST_ONE) { + pubMessage.setMessageID(evt.getMessageID()); + } LOG.debug("notify invoked"); m_clientIDsLock.lock(); @@ -254,4 +261,26 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { } session.close(true); } + + public void sendPubAck(PubAckEvent evt) { + LOG.debug("sendPubAck invoked"); + + String clientId = evt.getClientID(); + + PubAckMessage pubAckMessage = new PubAckMessage(); + pubAckMessage.setMessageID(evt.getMessageId()); + + m_clientIDsLock.lock(); + try { + assert m_clientIDs != null; + LOG.debug("clientIDs are " + m_clientIDs); + assert m_clientIDs.get(clientId) != null; + LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); + m_clientIDs.get(clientId).getSession().write(pubAckMessage); + }catch(Throwable t) { + LOG.error(null, t); + } finally { + m_clientIDsLock.unlock(); + } + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java index 850b58b2..ed331543 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java @@ -4,10 +4,7 @@ import java.util.concurrent.BlockingQueue; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.INotifier; -import org.dna.mqtt.moquette.messaging.spi.impl.events.CleanInFlightEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.DisconnectEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,12 +35,14 @@ class Notifier implements Runnable { LOG.debug("Notifing event " + evt); if (evt instanceof NotifyEvent) { NotifyEvent evtN = (NotifyEvent) evt; - m_notifier.notify(evtN.getClientId(), evtN.getTopic(), evtN.getQos(), evtN.getMessage(), evtN.isRetained()); + m_notifier.notify(evtN); } else if (evt instanceof DisconnectEvent) { DisconnectEvent evtD = (DisconnectEvent) evt; m_notifier.disconnect(evtD.getSession()); } else if (evt instanceof CleanInFlightEvent) { m_messaging.refill(evt); + } else if (evt instanceof PubAckEvent) { + m_notifier.sendPubAck((PubAckEvent) evt); } } catch (InterruptedException ex) { interrupted = true; diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java index e1ebefc7..ca9cb9e8 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -15,23 +15,8 @@ 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.messaging.spi.impl.events.MessagingEvent; -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.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.PublishMessage; -import org.dna.mqtt.moquette.proto.messages.SubAckMessage; -import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; +import org.dna.mqtt.moquette.proto.*; +import org.dna.mqtt.moquette.proto.messages.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** @@ -73,6 +58,7 @@ public class Server { encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder()); encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder()); encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder()); + encoder.addMessageEncoder(PubAckMessage.class, new PubAckEncoder()); encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder()); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java index 01b952cd..28aaa6df 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java @@ -6,6 +6,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.dna.mqtt.moquette.client.Client; import org.dna.mqtt.moquette.client.IPublishCallback; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.junit.After; import static org.junit.Assert.*; import org.junit.Before; @@ -313,5 +314,29 @@ public class ServerIntegrationTest { client.close(); client.shutdown(); } + + /*-------------- QoS 1 tests ----*/ + @Test + public void testPublishWithQoS1() throws IOException, InterruptedException { +// startServer(); + Client client = new Client("localhost", Server.PORT); +// Client client = new Client("test.mosquitto.org", 1883); + client.connect(); + + + client.subscribe("/topic", new IPublishCallback() { + + public void published(String topic, byte[] message) { + received = true; + } + }); + + client.publish("/topic", "Test my payload".getBytes(), AbstractMessage.QOSType.LEAST_ONE, false); + + client.close(); + client.shutdown(); + + assertTrue(received); + } }