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

Minor change, remove unused (and without any meaning) session parameter from PublishEvent

Browse Source
This commit is contained in:
selva.andre
2014-06-15 18:47:17 +02:00
parent fdc7f6237e
commit d22ec1cf7a
5 changed files with 14 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+1 -1
View File
@@ -307,7 +307,7 @@ public class HawtDBStorageService implements IStorageService {
ByteBuffer bbmessage = ByteBuffer.wrap(message); ByteBuffer bbmessage = ByteBuffer.wrap(message);
//bbmessage.flip(); //bbmessage.flip();
PublishEvent liveEvt = new PublishEvent(evt.getTopic(), evt.getQos(), PublishEvent liveEvt = new PublishEvent(evt.getTopic(), evt.getQos(),
bbmessage, evt.isRetain(), evt.getClientID(), evt.getMessageID(), null); bbmessage, evt.isRetain(), evt.getClientID(), evt.getMessageID());
return liveEvt; return liveEvt;
} }
} }
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+3 -3
View File
@@ -318,13 +318,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
//if the target subscription is not clean session and is not connected => store it //if the target subscription is not clean session and is not connected => store it
if (!sub.isCleanSession() && !sub.isActive()) { if (!sub.isCleanSession() && !sub.isActive()) {
//clone the event with matching clientID //clone the event with matching clientID
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID);
m_storageService.storePublishForFuture(newPublishEvt); m_storageService.storePublishForFuture(newPublishEvt);
} else { } else {
//if QoS 2 then store it in temp memory //if QoS 2 then store it in temp memory
if (qos ==AbstractMessage.QOSType.EXACTLY_ONCE) { if (qos ==AbstractMessage.QOSType.EXACTLY_ONCE) {
String publishKey = String.format("%s%d", sub.getClientId(), messageID); String publishKey = String.format("%s%d", sub.getClientId(), messageID);
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID, null); PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID);
m_storageService.addInFlight(newPublishEvt, publishKey); m_storageService.addInFlight(newPublishEvt, publishKey);
} }
//publish //publish
@@ -482,7 +482,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
if (m_willStore.containsKey(clientID)) { if (m_willStore.containsKey(clientID)) {
WillMessage will = m_willStore.get(clientID); WillMessage will = m_willStore.get(clientID);
PublishEvent pubEvt = new PublishEvent(will.getTopic(), will.getQos(), PublishEvent pubEvt = new PublishEvent(will.getTopic(), will.getQos(),
will.getPayload(), will.isRetained(), clientID, null); will.getPayload(), will.isRetained(), clientID);
processPublish(pubEvt); processPublish(pubEvt);
m_willStore.remove(clientID); m_willStore.remove(clientID);
} }
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+1 -1
View File
@@ -145,7 +145,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} else if (message instanceof PublishMessage) { } else if (message instanceof PublishMessage) {
PublishEvent pubEvt; PublishEvent pubEvt;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
pubEvt = new PublishEvent((PublishMessage) message, clientID, session); pubEvt = new PublishEvent((PublishMessage) message, clientID);
// if (message.getQos() == QOSType.MOST_ONE) { // if (message.getQos() == QOSType.MOST_ONE) {
// pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session); // pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session);
// //
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
+4 -13
View File
@@ -35,33 +35,29 @@ public class PublishEvent extends MessagingEvent {
//Optional attribute, available only fo QoS 1 and 2 //Optional attribute, available only fo QoS 1 and 2
int m_msgID; int m_msgID;
transient ServerChannel m_session; public PublishEvent(PublishMessage pubMsg, String clientID) {
public PublishEvent(PublishMessage pubMsg, String clientID, ServerChannel session) {
m_topic = pubMsg.getTopicName(); m_topic = pubMsg.getTopicName();
m_qos = pubMsg.getQos(); m_qos = pubMsg.getQos();
m_message = pubMsg.getPayload(); m_message = pubMsg.getPayload();
m_retain = pubMsg.isRetainFlag(); m_retain = pubMsg.isRetainFlag();
m_clientID = clientID; m_clientID = clientID;
m_session = session;
if (pubMsg.getQos() != QOSType.MOST_ONE) { if (pubMsg.getQos() != QOSType.MOST_ONE) {
m_msgID = pubMsg.getMessageID(); m_msgID = pubMsg.getMessageID();
} }
} }
public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain,
String clientID, ServerChannel session) { String clientID) {
m_topic = topic; m_topic = topic;
m_qos = qos; m_qos = qos;
m_message = message; m_message = message;
m_retain = retain; m_retain = retain;
m_clientID = clientID; m_clientID = clientID;
m_session = session;
} }
public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain,
String clientID, int msgID, ServerChannel session) { String clientID, int msgID) {
this(topic, qos, message, retain, clientID, session); this(topic, qos, message, retain, clientID);
m_msgID = msgID; m_msgID = msgID;
} }
@@ -89,11 +85,6 @@ public class PublishEvent extends MessagingEvent {
return m_msgID; return m_msgID;
} }
//TODO remove the session it's never used!!
public ServerChannel getSession() {
return m_session;
}
@Override @Override
public String toString() { public String toString() {
return "PublishEvent{" + return "PublishEvent{" +
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+5 -5
View File
@@ -272,7 +272,7 @@ public class ProtocolProcessorTest {
//Exercise //Exercise
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, false, "FakeCLI", null); PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, false, "FakeCLI");
m_processor.processPublish(pubEvt); m_processor.processPublish(pubEvt);
//Verify //Verify
@@ -321,7 +321,7 @@ public class ProtocolProcessorTest {
//Exercise //Exercise
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
buffer.rewind(); buffer.rewind();
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, false, "FakeCLI", null); PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, false, "FakeCLI");
m_processor.processPublish(pubEvt); m_processor.processPublish(pubEvt);
//Verify //Verify
@@ -418,7 +418,7 @@ public class ProtocolProcessorTest {
connectMessage.setCleanSession(subscription.isCleanSession()); connectMessage.setCleanSession(subscription.isCleanSession());
m_processor.processConnect(m_session, connectMessage); m_processor.processConnect(m_session, connectMessage);
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher", null); PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher");
m_processor.processPublish(pubEvt); m_processor.processPublish(pubEvt);
//Exercise //Exercise
@@ -447,7 +447,7 @@ public class ProtocolProcessorTest {
//Exercise //Exercise
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
PublishEvent pubEvt = new PublishEvent("/topic", AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher", null); PublishEvent pubEvt = new PublishEvent("/topic", AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher");
m_processor.processPublish(pubEvt); m_processor.processPublish(pubEvt);
//Verify no message is received //Verify no message is received
@@ -467,7 +467,7 @@ public class ProtocolProcessorTest {
//Exercise //Exercise
ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes());
PublishEvent pubEvt = new PublishEvent("/topic", AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher", null); PublishEvent pubEvt = new PublishEvent("/topic", AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher");
m_processor.processPublish(pubEvt); m_processor.processPublish(pubEvt);
//Verify no message is received //Verify no message is received