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

Simplified APIs

Browse Source
This commit is contained in:
andrea
2013-08-30 19:21:11 +02:00
parent 2c11cade06
commit 0694f28f32
6 changed files with 23 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
-2
View File
@@ -14,7 +14,5 @@ public interface IMessaging {
void disconnect(IoSession session);
// void republishStored(String clientID);
void handleProtocolMessage(IoSession session, AbstractMessage msg);
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java
-21
View File
@@ -1,21 +0,0 @@
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;
/**
*
* @author andrea
*/
public interface INotifier {
public void notify(NotifyEvent evt/*String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained*/);
public void disconnect(IoSession session);
void sendPubAck(PubAckEvent evt);
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+18 -19
View File
@@ -8,10 +8,8 @@ import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
import org.dna.mqtt.moquette.messaging.spi.impl.events.NotifyEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.RepublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
import org.dna.mqtt.moquette.proto.PubCompMessage;
@@ -47,18 +45,16 @@ class ProtocolProcessor {
private SubscriptionsStore subscriptions;
private IStorageService m_storageService;
private IAuthenticator m_authenticator;
private IMessaging m_messaging;
ProtocolProcessor() {
}
void init(Map<String, ConnectionDescriptor> clientIDs, SubscriptionsStore subscriptions,
IStorageService storageService, IMessaging messaging) {
IStorageService storageService) {
m_clientIDs = clientIDs;
this.subscriptions = subscriptions;
m_storageService = storageService;
m_messaging = messaging;
}
void processConnect(IoSession session, ConnectMessage msg) {
@@ -152,8 +148,8 @@ class ProtocolProcessor {
}
for (PublishEvent pubEvt : publishedEvents) {
notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID()));
sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID());
}
}
@@ -206,7 +202,7 @@ class ProtocolProcessor {
for (final Subscription sub : subscriptions.matches(topic)) {
if (qos == AbstractMessage.QOSType.MOST_ONE) {
//QoS 0
notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false));
sendPublish(sub.getClientId(), topic, qos, message, false);
} else {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
@@ -222,22 +218,25 @@ class ProtocolProcessor {
m_storageService.addInFlight(newPublishEvt, publishKey);
}
//publish
notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false));
sendPublish(sub.getClientId(), topic, qos, message, false);
}
}
}
}
private void notify(NotifyEvent evt) {
LOG.debug("notify invoked with event " + evt);
String clientId = evt.getClientId();
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained) {
sendPublish(clientId, topic, qos, message, retained, 0);
}
private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained, int messageID) {
LOG.debug("notify invoked with event ");
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(evt.isRetained());
pubMessage.setTopicName(evt.getTopic());
pubMessage.setQos(evt.getQos());
pubMessage.setPayload(evt.getMessage());
pubMessage.setRetainFlag(retained);
pubMessage.setTopicName(topic);
pubMessage.setQos(qos);
pubMessage.setPayload(message);
if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) {
pubMessage.setMessageID(evt.getMessageID());
pubMessage.setMessageID(messageID);
}
try {
@@ -399,8 +398,8 @@ class ProtocolProcessor {
for (HawtDBStorageService.StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("Inserting NotifyEvent into outbound for topic " + topic);
notify(new NotifyEvent(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true));
LOG.debug("send publish message for topic " + topic);
sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
}
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+2 -29
View File
@@ -87,17 +87,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
disruptorPublish(new DisconnectEvent(session));
}
//method used by hte Notifier to re-put an event on the inbound queue
// private void refill(MessagingEvent evt) {
// disruptorPublish(evt);
// }
/*public void republishStored(String clientID) {
//create the event to push
LOG.debug("republishStored invoked to publish soterd messages for clientID " + clientID);
disruptorPublish(new RepublishEvent(clientID));
}*/
public void handleProtocolMessage(IoSession session, AbstractMessage msg) {
disruptorPublish(new ProtocolEvent(session, msg));
}
@@ -117,9 +106,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
DisconnectEvent disEvt = (DisconnectEvent) evt;
String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID);
m_processor.processDisconnect(disEvt.getSession(), clientID, false);
} /*else if (evt instanceof RepublishEvent) {
processRepublish((RepublishEvent) evt);
} */else if (evt instanceof ProtocolEvent) {
} else if (evt instanceof ProtocolEvent) {
IoSession session = ((ProtocolEvent) evt).getSession();
AbstractMessage message = ((ProtocolEvent) evt).getMessage();
if (message instanceof ConnectMessage) {
@@ -178,7 +165,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
m_storageService.initStore();
subscriptions.init(m_storageService);
m_processor.init(m_clientIDs, subscriptions, m_storageService, this);
m_processor.init(m_clientIDs, subscriptions, m_storageService);
}
@@ -189,18 +176,4 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
// m_eventProcessor.halt();
m_executor.shutdown();
}
// private void processRepublish(RepublishEvent evt) throws InterruptedException {
// LOG.debug("processRepublish invoked");
// List<PublishEvent> publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID());
// if (publishedEvents == null) {
// LOG.debug("processRepublish, no stored publish events");
// return;
// }
//
// for (PublishEvent pubEvt : publishedEvents) {
// m_processor.notify(new NotifyEvent(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
// pubEvt.getMessage(), false, pubEvt.getMessageID()));
// }
// }
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/NotifyEvent.java
-69
View File
@@ -1,69 +0,0 @@
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
*/
public class NotifyEvent extends MessagingEvent {
private String m_clientId;
private String m_topic;
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) {
m_clientId = clientId;
m_topic = topic;
m_qos = qos;
m_message = message;
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;
}
public String getTopic() {
return m_topic;
}
public QOSType getQos() {
return m_qos;
}
public byte[] getMessage() {
return m_message;
}
public boolean isRetained() {
return m_retaned;
}
public int getMessageID() {
return m_msgID;
}
@Override
public String toString() {
return "NotifyEvent{" +
"m_retaned=" + m_retaned +
", m_msgID=" + m_msgID +
", m_qos=" + m_qos +
", m_topic='" + m_topic + '\'' +
", m_clientId='" + m_clientId + '\'' +
'}';
}
}