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

Added QoS 1 pub ack to the server part

Browse Source
This commit is contained in:
selva.andre
2012-08-31 19:23:53 +02:00
parent cb95ecc7cb
commit 65e34ddc40
10 changed files with 152 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+6
View File
@@ -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
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java
+7 -2
View File
@@ -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);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+17 -4
View File
@@ -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) {
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java
+13
View File
@@ -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;
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PubAckEvent.java
+24
View File
@@ -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;
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
+6 -4
View File
@@ -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;
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+47 -18
View File
@@ -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();
}
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/Notifier.java
+4 -5
View File
@@ -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;
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+3 -17
View File
@@ -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());
parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java
+25
View File
@@ -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);
}
}