From 817ca8e1761c57a3a6642fce9f2970cd1e22f962 Mon Sep 17 00:00:00 2001 From: andrea Date: Thu, 14 Nov 2013 16:43:03 +0100 Subject: [PATCH] Added output reactor to isolate the writing of messages --- .../messaging/spi/impl/ProtocolProcessor.java | 61 ++++++++++++++++--- .../messaging/spi/impl/SimpleMessaging.java | 5 +- .../spi/impl/events/OutputMessagingEvent.java | 28 +++++++++ .../server/ServerIntegrationPahoTest.java | 2 + 4 files changed, 83 insertions(+), 13 deletions(-) create mode 100644 broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/OutputMessagingEvent.java 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 69e2ace7..a478a993 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 @@ -1,11 +1,19 @@ 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.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.OutputMessagingEvent; 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.subscriptions.Subscription; @@ -36,7 +44,7 @@ import org.slf4j.LoggerFactory; * * @author andrea */ -class ProtocolProcessor { +class ProtocolProcessor implements EventHandler { private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); @@ -44,16 +52,29 @@ class ProtocolProcessor { private SubscriptionsStore subscriptions; private IStorageService m_storageService; private IAuthenticator m_authenticator; + + private ExecutorService m_executor; + BatchEventProcessor m_eventProcessor; + private RingBuffer m_ringBuffer; - ProtocolProcessor() { - - } + ProtocolProcessor() {} void init(/*Map clientIDs, */SubscriptionsStore subscriptions, IStorageService storageService) { //m_clientIDs = clientIDs; this.subscriptions = subscriptions; m_storageService = storageService; + + //init the output ringbuffer + 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); } void processConnect(ServerChannel session, ConnectMessage msg) { @@ -253,7 +274,8 @@ class ProtocolProcessor { throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs)); } LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); - m_clientIDs.get(clientId).getSession().write(pubMessage); +// m_clientIDs.get(clientId).getSession().write(pubMessage); + disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubMessage)); }catch(Throwable t) { LOG.error(null, t); } @@ -264,7 +286,8 @@ class ProtocolProcessor { PubRecMessage pubRecMessage = new PubRecMessage(); pubRecMessage.setMessageID(messageID); - m_clientIDs.get(clientID).getSession().write(pubRecMessage); +// m_clientIDs.get(clientID).getSession().write(pubRecMessage); + disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRecMessage)); } private void sendPubAck(PubAckEvent evt) { @@ -284,7 +307,8 @@ class ProtocolProcessor { throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs)); } LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); - m_clientIDs.get(clientId).getSession().write(pubAckMessage); +// m_clientIDs.get(clientId).getSession().write(pubAckMessage); + disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubAckMessage)); }catch(Throwable t) { LOG.error(null, t); } @@ -319,7 +343,8 @@ class ProtocolProcessor { PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); - m_clientIDs.get(clientID).getSession().write(pubCompMessage); +// m_clientIDs.get(clientID).getSession().write(pubCompMessage); + disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubCompMessage)); } void processPubRec(String clientID, int messageID) { @@ -328,7 +353,8 @@ class ProtocolProcessor { PubRelMessage pubRelMessage = new PubRelMessage(); pubRelMessage.setMessageID(messageID); - m_clientIDs.get(clientID).getSession().write(pubRelMessage); +// m_clientIDs.get(clientID).getSession().write(pubRelMessage); + disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage)); } void processPubComp(String clientID, int messageID) { @@ -407,5 +433,22 @@ class ProtocolProcessor { sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true); } } + + private void disruptorPublish(OutputMessagingEvent msgEvent) { + LOG.debug("disruptorPublish publishing event on output " + msgEvent); + long sequence = m_ringBuffer.next(); + ValueEvent event = m_ringBuffer.get(sequence); + + event.setEvent(msgEvent); + + m_ringBuffer.publish(sequence); + } + + public void onEvent(ValueEvent t, long l, boolean bln) throws Exception { + MessagingEvent evt = t.getEvent(); + //It's always of type OutputMessagingEvent + OutputMessagingEvent outEvent = (OutputMessagingEvent) evt; + outEvent.getChannel().write(outEvent.getMessage()); + } } 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 762e0988..a876d165 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 @@ -4,8 +4,6 @@ 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; @@ -15,7 +13,6 @@ 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; @@ -96,7 +93,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { public void onEvent(ValueEvent t, long l, boolean bln) throws Exception { MessagingEvent evt = t.getEvent(); - LOG.info("onEvent processing messaging event " + evt); + LOG.info("onEvent processing messaging event from input ringbuffer" + evt); if (evt instanceof PublishEvent) { m_processor.processPublish((PublishEvent) evt); } else if (evt instanceof StopEvent) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/OutputMessagingEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/OutputMessagingEvent.java new file mode 100644 index 00000000..25a54512 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/OutputMessagingEvent.java @@ -0,0 +1,28 @@ +package org.dna.mqtt.moquette.messaging.spi.impl.events; + +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.server.ServerChannel; + +/** + * + * @author andrea + */ +public class OutputMessagingEvent extends MessagingEvent { + private ServerChannel m_channel; + private AbstractMessage m_message; + + public OutputMessagingEvent(ServerChannel channel, AbstractMessage message) { + m_channel = channel; + m_message = message; + } + + public ServerChannel getChannel() { + return m_channel; + } + + public AbstractMessage getMessage() { + return m_message; + } + + +} diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java index f0e2b8aa..7703813b 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java @@ -315,6 +315,8 @@ public class ServerIntegrationPahoTest { m_client.connect(options); + assertNotNull(m_callback); + assertNotNull(m_callback.getMessage()); assertEquals("Hello MQTT", m_callback.getMessage().toString()); }