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

Initial try to remove buffer copies

Browse Source
This commit is contained in:
andrea
2013-11-25 09:34:36 +01:00
parent 9603310b4e
commit a1b1417acb
6 changed files with 49 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java
+2 -1
View File
@@ -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<StoredMessage> searchMatching(IMatchingCondition condition);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+4 -3
View File
@@ -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<String, PublishEvent>) 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);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+12 -11
View File
@@ -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<ValueEvent> {
//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<ValueEvent> {
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<ValueEvent> {
/**
* 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<ValueEvent> {
}
}
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<ValueEvent> {
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);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+7 -9
View File
@@ -108,17 +108,15 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
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);
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/PublishEvent.java
+19 -4
View File
@@ -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;
}