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

Minor fixes to typos

Browse Source
This commit is contained in:
selva.andre
2013-09-19 20:11:52 +02:00
parent 1dfc6ff77c
commit f09fcd6cd5
4 changed files with 190 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+9 -4
View File
@@ -57,9 +57,11 @@ class ProtocolProcessor {
}
void processConnect(ServerChannel session, ConnectMessage msg) {
LOG.info("processConnect for client " + msg.getClientID());
if (msg.getProcotolVersion() != 0x03) {
ConnAckMessage badProto = new ConnAckMessage();
badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
LOG.info("processConnect sent bad proto ConnAck");
session.write(badProto);
session.close(false);
return;
@@ -130,6 +132,7 @@ class ProtocolProcessor {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
LOG.info("processConnect sent OK ConnAck");
session.write(okResp);
if (!msg.isCleanSession()) {
@@ -187,7 +190,7 @@ class ProtocolProcessor {
}
m_storageService.cleanInFlight(publishKey);
sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID()));
LOG.info("replying with PubAck to MSG ID " + evt.getMessageID());
LOG.debug("replying with PubAck to MSG ID " + evt.getMessageID());
}
if (retain) {
@@ -199,7 +202,9 @@ class ProtocolProcessor {
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */
private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) {
LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic " + topic);
for (final Subscription sub : subscriptions.matches(topic)) {
LOG.debug("found matching subscription on topic " + sub.getTopic());
if (qos == AbstractMessage.QOSType.MOST_ONE) {
//QoS 0
sendPublish(sub.getClientId(), topic, qos, message, false);
@@ -360,13 +365,13 @@ class ProtocolProcessor {
UnsubAckMessage ackMessage = new UnsubAckMessage();
ackMessage.setMessageID(messageID);
LOG.info("replying with UnsubAck to MSG ID {0}", messageID);
LOG.info(String.format("replying with UnsubAck to MSG ID %s", messageID));
session.write(ackMessage);
}
void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) {
LOG.debug("processSubscribe invoked");
LOG.info(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID()));
for (SubscribeMessage.Couple req : msg.subscriptions()) {
AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
@@ -382,7 +387,7 @@ class ProtocolProcessor {
for (int i = 0; i < msg.subscriptions().size(); i++) {
ackMessage.addType(AbstractMessage.QOSType.MOST_ONE);
}
LOG.info("replying with SubAct to MSG ID " + msg.getMessageID());
LOG.info("replying with SubAck to MSG ID " + msg.getMessageID());
session.write(ackMessage);
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+178 -178
View File
@@ -1,178 +1,178 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
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.*;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
import org.dna.mqtt.moquette.proto.messages.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* Singleton class that orchestrate the execution of the protocol.
*
* Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion.
*
* @author andrea
*/
public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class);
private SubscriptionsStore subscriptions = new SubscriptionsStore();
private RingBuffer<ValueEvent> m_ringBuffer;
private IStorageService m_storageService;
private ExecutorService m_executor;
BatchEventProcessor<ValueEvent> m_eventProcessor;
private static SimpleMessaging INSTANCE;
private ProtocolProcessor m_processor = new ProtocolProcessor();
private SimpleMessaging() {
}
public static SimpleMessaging getInstance() {
if (INSTANCE == null) {
INSTANCE = new SimpleMessaging();
}
return INSTANCE;
}
public void init() {
m_executor = Executors.newFixedThreadPool(1);
m_ringBuffer = new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32);
SequenceBarrier barrier = m_ringBuffer.newBarrier();
m_eventProcessor = new BatchEventProcessor<ValueEvent>(m_ringBuffer, barrier, this);
//TODO in a presentation is said to don't do the followinf line!!
m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
m_executor.submit(m_eventProcessor);
disruptorPublish(new InitEvent());
}
private void disruptorPublish(MessagingEvent msgEvent) {
LOG.info("disruptorPublish publishing event " + msgEvent);
long sequence = m_ringBuffer.next();
ValueEvent event = m_ringBuffer.get(sequence);
event.setEvent(msgEvent);
m_ringBuffer.publish(sequence);
}
public void disconnect(ServerChannel session) {
disruptorPublish(new DisconnectEvent(session));
}
public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) {
disruptorPublish(new ProtocolEvent(session, msg));
}
public void stop() {
disruptorPublish(new StopEvent());
}
public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
MessagingEvent evt = t.getEvent();
LOG.info("onEvent processing messaging event " + evt);
if (evt instanceof PublishEvent) {
m_processor.processPublish((PublishEvent) evt);
} else if (evt instanceof StopEvent) {
processStop();
} else if (evt instanceof DisconnectEvent) {
DisconnectEvent disEvt = (DisconnectEvent) evt;
String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID);
m_processor.processDisconnect(disEvt.getSession(), clientID, false);
} else if (evt instanceof ProtocolEvent) {
ServerChannel session = ((ProtocolEvent) evt).getSession();
AbstractMessage message = ((ProtocolEvent) evt).getMessage();
if (message instanceof ConnectMessage) {
m_processor.processConnect(session, (ConnectMessage) message);
} else if (message instanceof PublishMessage) {
PublishMessage pubMsg = (PublishMessage) message;
PublishEvent pubEvt;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
if (message.getQos() == QOSType.MOST_ONE) {
pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session);
} else {
pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session);
}
m_processor.processPublish(pubEvt);
} else if (message instanceof DisconnectMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
//close the TCP connection
//session.close(true);
m_processor.processDisconnect(session, clientID, cleanSession);
} else if (message instanceof UnsubscribeMessage) {
UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
m_processor.processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID());
} else if (message instanceof SubscribeMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
m_processor.processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession);
} else if (message instanceof PubRelMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubRelMessage) message).getMessageID();
m_processor.processPubRel(clientID, messageID);
} else if (message instanceof PubRecMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubRecMessage) message).getMessageID();
m_processor.processPubRec(clientID, messageID);
} else if (message instanceof PubCompMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubCompMessage) message).getMessageID();
m_processor.processPubComp(clientID, messageID);
} else {
throw new RuntimeException("Illegal message received " + message);
}
} else if (evt instanceof InitEvent) {
processInit();
}
}
private void processInit() {
m_storageService = new HawtDBStorageService();
m_storageService.initStore();
subscriptions.init(m_storageService);
m_processor.init(subscriptions, m_storageService);
}
private void processStop() {
LOG.debug("processStop invoked");
m_storageService.close();
// m_eventProcessor.halt();
m_executor.shutdown();
}
}
package org.dna.mqtt.moquette.messaging.spi.impl;
import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
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.*;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
import org.dna.mqtt.moquette.proto.messages.*;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* Singleton class that orchestrate the execution of the protocol.
*
* Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion.
*
* @author andrea
*/
public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class);
private SubscriptionsStore subscriptions = new SubscriptionsStore();
private RingBuffer<ValueEvent> m_ringBuffer;
private IStorageService m_storageService;
private ExecutorService m_executor;
BatchEventProcessor<ValueEvent> m_eventProcessor;
private static SimpleMessaging INSTANCE;
private ProtocolProcessor m_processor = new ProtocolProcessor();
private SimpleMessaging() {
}
public static SimpleMessaging getInstance() {
if (INSTANCE == null) {
INSTANCE = new SimpleMessaging();
}
return INSTANCE;
}
public void init() {
m_executor = Executors.newFixedThreadPool(1);
m_ringBuffer = new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32);
SequenceBarrier barrier = m_ringBuffer.newBarrier();
m_eventProcessor = new BatchEventProcessor<ValueEvent>(m_ringBuffer, barrier, this);
//TODO in a presentation is said to don't do the followinf line!!
m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence());
m_executor.submit(m_eventProcessor);
disruptorPublish(new InitEvent());
}
private void disruptorPublish(MessagingEvent msgEvent) {
LOG.debug("disruptorPublish publishing event " + msgEvent);
long sequence = m_ringBuffer.next();
ValueEvent event = m_ringBuffer.get(sequence);
event.setEvent(msgEvent);
m_ringBuffer.publish(sequence);
}
public void disconnect(ServerChannel session) {
disruptorPublish(new DisconnectEvent(session));
}
public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) {
disruptorPublish(new ProtocolEvent(session, msg));
}
public void stop() {
disruptorPublish(new StopEvent());
}
public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
MessagingEvent evt = t.getEvent();
LOG.info("onEvent processing messaging event " + evt);
if (evt instanceof PublishEvent) {
m_processor.processPublish((PublishEvent) evt);
} else if (evt instanceof StopEvent) {
processStop();
} else if (evt instanceof DisconnectEvent) {
DisconnectEvent disEvt = (DisconnectEvent) evt;
String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID);
m_processor.processDisconnect(disEvt.getSession(), clientID, false);
} else if (evt instanceof ProtocolEvent) {
ServerChannel session = ((ProtocolEvent) evt).getSession();
AbstractMessage message = ((ProtocolEvent) evt).getMessage();
if (message instanceof ConnectMessage) {
m_processor.processConnect(session, (ConnectMessage) message);
} else if (message instanceof PublishMessage) {
PublishMessage pubMsg = (PublishMessage) message;
PublishEvent pubEvt;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
if (message.getQos() == QOSType.MOST_ONE) {
pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, session);
} else {
pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session);
}
m_processor.processPublish(pubEvt);
} else if (message instanceof DisconnectMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
//close the TCP connection
//session.close(true);
m_processor.processDisconnect(session, clientID, cleanSession);
} else if (message instanceof UnsubscribeMessage) {
UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
m_processor.processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID());
} else if (message instanceof SubscribeMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
m_processor.processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession);
} else if (message instanceof PubRelMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubRelMessage) message).getMessageID();
m_processor.processPubRel(clientID, messageID);
} else if (message instanceof PubRecMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubRecMessage) message).getMessageID();
m_processor.processPubRec(clientID, messageID);
} else if (message instanceof PubCompMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubCompMessage) message).getMessageID();
m_processor.processPubComp(clientID, messageID);
} else {
throw new RuntimeException("Illegal message received " + message);
}
} else if (evt instanceof InitEvent) {
processInit();
}
}
private void processInit() {
m_storageService = new HawtDBStorageService();
m_storageService.initStore();
subscriptions.init(m_storageService);
m_processor.init(subscriptions, m_storageService);
}
private void processStop() {
LOG.debug("processStop invoked");
m_storageService.close();
// m_eventProcessor.halt();
m_executor.shutdown();
}
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java
+1 -1
View File
@@ -79,7 +79,7 @@ public class SubscriptionsStore {
protected void addDirect(Subscription newSubscription) {
TreeNode current = findMatchingNode(newSubscription.topic);
current.addSubcription(newSubscription);
current.addSubscription(newSubscription);
}
private TreeNode findMatchingNode(String topic) {
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java
+2 -2
View File
@@ -24,7 +24,7 @@ class TreeNode {
this.m_token = topic;
}
void addSubcription(Subscription s) {
void addSubscription(Subscription s) {
//avoid double registering
if (m_subscriptions.contains(s)) {
return;
@@ -82,7 +82,7 @@ class TreeNode {
for (TreeNode n : m_children) {
if (n.getToken().match(t)) {
//Create a copy of token, alse if navigate 2 sibling it
//Create a copy of token, else if navigate 2 sibling it
//consumes 2 elements on the queue instead of one
n.matches(new LinkedBlockingQueue<Token>(tokens), matchingSubs);
}