From a1b1417acb988e1c3c1bb94605be62ced7c20b8d Mon Sep 17 00:00:00 2001 From: andrea Date: Mon, 25 Nov 2013 09:34:36 +0100 Subject: [PATCH] Initial try to remove buffer copies --- .../messaging/spi/IStorageService.java | 3 ++- .../spi/impl/HawtDBStorageService.java | 7 +++--- .../messaging/spi/impl/ProtocolProcessor.java | 23 ++++++++++--------- .../messaging/spi/impl/SimpleMessaging.java | 16 ++++++------- .../spi/impl/events/PublishEvent.java | 23 +++++++++++++++---- .../proto/messages/PublishMessage.java | 8 ++++--- 6 files changed, 49 insertions(+), 31 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java index 16b346af..500a8a88 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java @@ -1,5 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi; +import java.nio.ByteBuffer; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; @@ -18,7 +19,7 @@ public interface IStorageService { * */ void initStore(); - void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos); + void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos); Collection searchMatching(IMatchingCondition condition); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index a89edea7..cc3c126c 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -15,6 +15,7 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.io.Serializable; +import java.nio.ByteBuffer; import java.util.*; @@ -38,8 +39,8 @@ public class HawtDBStorageService implements IStorageService { return m_qos; } - byte[] getPayload() { - return m_payload; + ByteBuffer getPayload() { + return (ByteBuffer) ByteBuffer.allocate(m_payload.length).put(m_payload).flip(); } String getTopic() { @@ -132,7 +133,7 @@ public class HawtDBStorageService implements IStorageService { m_qos2Store = (SortedIndex) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory); } - public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { + public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) { if (message.length == 0) { //clean the message from topic m_retainedStore.remove(topic); 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 a478a993..95d39ae5 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 @@ -4,6 +4,7 @@ import com.lmax.disruptor.BatchEventProcessor; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.SequenceBarrier; +import java.nio.ByteBuffer; import java.util.Collection; import java.util.HashMap; import java.util.List; @@ -123,8 +124,10 @@ class ProtocolProcessor implements EventHandler { //Handle will flag if (msg.isWillFlag()) { AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()]; - PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, msg.getWillMessage().getBytes(), - msg.isWillRetain(), msg.getClientID(), session); + byte[] willPayload = msg.getWillMessage().getBytes(); + ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); + PublishEvent pubEvt = new PublishEvent(msg.getWillTopic(), willQos, + bb, msg.isWillRetain(), msg.getClientID(), session); processPublish(pubEvt); } @@ -188,7 +191,7 @@ class ProtocolProcessor implements EventHandler { LOG.debug("processPublish invoked with " + evt); final String topic = evt.getTopic(); final AbstractMessage.QOSType qos = evt.getQos(); - final byte[] message = evt.getMessage(); + final ByteBuffer message = evt.getMessage(); boolean retain = evt.isRetain(); String publishKey = null; @@ -222,7 +225,7 @@ class ProtocolProcessor implements EventHandler { /** * 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) { + private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer 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()); @@ -250,11 +253,11 @@ class ProtocolProcessor implements EventHandler { } } - private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, byte[] message, boolean retained) { + private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer 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) { + private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, int messageID) { LOG.debug("notify invoked with event "); PublishMessage pubMessage = new PublishMessage(); pubMessage.setRetainFlag(retained); @@ -324,15 +327,13 @@ class ProtocolProcessor implements EventHandler { final String topic = evt.getTopic(); final AbstractMessage.QOSType qos = evt.getQos(); - final byte[] message = evt.getMessage(); - boolean retain = evt.isRetain(); - publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + publish2Subscribers(topic, qos, evt.getMessage(), evt.isRetain(), evt.getMessageID()); m_storageService.removeQoS2Message(publishKey); - if (retain) { - m_storageService.storeRetained(topic, message, qos); + if (evt.isRetain()) { + m_storageService.storeRetained(topic, evt.getMessage(), qos); } sendPubComp(clientID, messageID); 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 a876d165..354671fe 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 @@ -108,17 +108,15 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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); - } + pubEvt = new PublishEvent((PublishMessage) message, clientID, session); +// 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); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java index 7e2c2b5b..750cb374 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java @@ -3,6 +3,8 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import java.io.Serializable; +import java.nio.ByteBuffer; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.server.ServerChannel; /** @@ -12,15 +14,28 @@ import org.dna.mqtt.moquette.server.ServerChannel; public class PublishEvent extends MessagingEvent implements Serializable { String m_topic; QOSType m_qos; - byte[] m_message; + //byte[] m_message; + ByteBuffer m_message; boolean m_retain; String m_clientID; //Optional attribute, available only fo QoS 1 and 2 int m_msgID; transient ServerChannel m_session; + + public PublishEvent(PublishMessage pubMsg, String clientID, ServerChannel session) { + m_topic = pubMsg.getTopicName(); + m_qos = pubMsg.getQos(); + m_message = pubMsg.getPayload(); + m_retain = pubMsg.isRetainFlag(); + m_clientID = clientID; + m_session = session; + if (pubMsg.getQos() != QOSType.MOST_ONE) { + m_msgID = pubMsg.getMessageID(); + } + } - public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, + public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, String clientID, ServerChannel session) { m_topic = topic; m_qos = qos; @@ -30,7 +45,7 @@ public class PublishEvent extends MessagingEvent implements Serializable { m_session = session; } - public PublishEvent(String topic, QOSType qos, byte[] message, boolean retain, + public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, String clientID, int msgID, ServerChannel session) { this(topic, qos, message, retain, clientID, session); m_msgID = msgID; @@ -44,7 +59,7 @@ public class PublishEvent extends MessagingEvent implements Serializable { return m_qos; } - public byte[] getMessage() { + public ByteBuffer getMessage() { return m_message; } diff --git a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java index 67a5da3f..24818db9 100644 --- a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java +++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java @@ -1,5 +1,7 @@ package org.dna.mqtt.moquette.proto.messages; +import java.nio.ByteBuffer; + /** * * @author andrea @@ -8,7 +10,7 @@ public class PublishMessage extends MessageIDMessage { private String m_topicName; // private Integer m_messageID; //could be null if Qos is == 0 - private byte[] m_payload; + private ByteBuffer m_payload; /*public Integer getMessageID() { return m_messageID; @@ -30,11 +32,11 @@ public class PublishMessage extends MessageIDMessage { this.m_topicName = topicName; } - public byte[] getPayload() { + public ByteBuffer getPayload() { return m_payload; } - public void setPayload(byte[] payload) { + public void setPayload(ByteBuffer payload) { this.m_payload = payload; } }