From f09fcd6cd5908297655226acd7e0da3ce2b2a879 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Thu, 19 Sep 2013 20:11:52 +0200 Subject: [PATCH] Minor fixes to typos --- .../messaging/spi/impl/ProtocolProcessor.java | 13 +- .../messaging/spi/impl/SimpleMessaging.java | 356 +++++++++--------- .../subscriptions/SubscriptionsStore.java | 2 +- .../spi/impl/subscriptions/TreeNode.java | 4 +- 4 files changed, 190 insertions(+), 185 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index ecfa8d7f..69e2ace7 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -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); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index bcedb5ea..762e0988 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -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 { - - private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class); - - private SubscriptionsStore subscriptions = new SubscriptionsStore(); - - private RingBuffer m_ringBuffer; - - private IStorageService m_storageService; - - private ExecutorService m_executor; - BatchEventProcessor 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.EVENT_FACTORY, 1024 * 32); - - SequenceBarrier barrier = m_ringBuffer.newBarrier(); - m_eventProcessor = new BatchEventProcessor(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 { + + private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class); + + private SubscriptionsStore subscriptions = new SubscriptionsStore(); + + private RingBuffer m_ringBuffer; + + private IStorageService m_storageService; + + private ExecutorService m_executor; + BatchEventProcessor 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.EVENT_FACTORY, 1024 * 32); + + SequenceBarrier barrier = m_ringBuffer.newBarrier(); + m_eventProcessor = new BatchEventProcessor(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(); + } +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java index c1d9fa3a..ceb34f0a 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java @@ -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) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java index aeb6bd91..dcf1ce59 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java @@ -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(tokens), matchingSubs); }