You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2020/07/31 12:48:56 UTC
[activemq-artemis] 02/04: ARTEMIS-2844 Save additional copies and
use pooled direct ByteBufs
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 70068a06596492d42c57246d6b27a3d6817a107f
Author: Francesco Nigro <ni...@gmail.com>
AuthorDate: Sat Jul 11 09:10:25 2020 +0200
ARTEMIS-2844 Save additional copies and use pooled direct ByteBufs
---
.../artemis/core/protocol/mqtt/MQTTPublishManager.java | 14 +++++++-------
1 file changed, 7 insertions(+), 7 deletions(-)
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
index adf7b1a..4b89636 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
@@ -17,10 +17,9 @@
package org.apache.activemq.artemis.core.protocol.mqtt;
-import java.nio.charset.StandardCharsets;
-
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.EmptyByteBuf;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException;
@@ -267,20 +266,21 @@ public class MQTTPublishManager {
}
private void sendServerMessage(int messageId, ICoreMessage message, int deliveryCount, int qos) {
- String address = MQTTUtil.convertCoreAddressFilterToMQTT(message.getAddress() == null ? "" : message.getAddress().toString(), session.getWildcardConfiguration());
+ String address = MQTTUtil.convertCoreAddressFilterToMQTT(message.getAddress() == null ? "" : message.getAddress(), session.getWildcardConfiguration());
boolean isRetain = message.getBooleanProperty(MQTT_MESSAGE_RETAIN_KEY);
ByteBuf payload;
switch (message.getType()) {
case Message.TEXT_TYPE:
SimpleString text = message.getDataBuffer().readNullableSimpleString();
- byte[] stringPayload = text.toString().getBytes(StandardCharsets.UTF_8);
- payload = ByteBufAllocator.DEFAULT.buffer(stringPayload.length);
- payload.writeBytes(stringPayload);
+ final int utf8Bytes = ByteBufUtil.utf8Bytes(text);
+ payload = ByteBufAllocator.DEFAULT.directBuffer(utf8Bytes);
+ // IMPORTANT: this one won't enlarge ByteBuf by ByteBufUtil.maxUtf8Bytes(text), but just utf8Bytes
+ ByteBufUtil.reserveAndWriteUtf8(payload, text, utf8Bytes);
break;
default:
ActiveMQBuffer bodyBuffer = message.getDataBuffer();
- payload = ByteBufAllocator.DEFAULT.buffer(bodyBuffer.writerIndex());
+ payload = ByteBufAllocator.DEFAULT.directBuffer(bodyBuffer.writerIndex());
payload.writeBytes(bodyBuffer.byteBuf());
break;
}