This is an automated email from the ASF dual-hosted git repository.
hengyuss pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new ad386882a0 fix(mqtt): preserve binary payloads in MQTT retained
messages (#7537)
ad386882a0 is described below
commit ad386882a0b47a1fcaa803c5d3a510aeb0074959
Author: lymerin <[email protected]>
AuthorDate: Sun Oct 11 01:30:57 2026 +0800
fix(mqtt): preserve binary payloads in MQTT retained messages (#7537)
---
.../org/apache/shenyu/protocol/mqtt/Publish.java | 13 +--
.../org/apache/shenyu/protocol/mqtt/Subscribe.java | 10 +-
.../mqtt/repositories/TopicRepository.java | 8 +-
.../apache/shenyu/protocol/mqtt/PublishTest.java | 101 +++++++++++++++++----
4 files changed, 92 insertions(+), 40 deletions(-)
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
index 279b1e5c04..76f28c08a7 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Publish.java
@@ -55,10 +55,11 @@ public class Publish extends MessageType {
}
String topic = msg.variableHeader().topicName();
ByteBuf payload = msg.payload();
- String message = byteBufToString(payload);
MqttQoS mqttQoS = msg.fixedHeader().qosLevel();
if (msg.fixedHeader().isRetain()) {
if (payload.isReadable()) {
+ byte[] message = new byte[payload.readableBytes()];
+ payload.getBytes(payload.readerIndex(), message);
Singleton.INST.get(TopicRepository.class).add(topic, message);
} else {
Singleton.INST.get(TopicRepository.class).remove(topic);
@@ -114,16 +115,6 @@ public class Publish extends MessageType {
ctx.writeAndFlush(mqttPubRecMessage);
}
- private String byteBufToString(final ByteBuf byteBuf) {
- if (byteBuf.hasArray()) {
- return new String(byteBuf.array(), byteBuf.arrayOffset() +
byteBuf.readerIndex(), byteBuf.readableBytes());
- } else {
- byte[] bytes = new byte[byteBuf.readableBytes()];
- byteBuf.getBytes(byteBuf.readerIndex(), bytes);
- return new String(bytes, 0, byteBuf.readableBytes());
- }
- }
-
private void send(final String topic, final ByteBuf payload, final MqttQoS
publishQoS) {
Map<Channel, MqttQoS> subscribers =
Singleton.INST.get(SubscribeRepository.class).get(topic);
//// todo thread pool
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Subscribe.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Subscribe.java
index 246c3ca4a2..947797513c 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Subscribe.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Subscribe.java
@@ -29,8 +29,6 @@ import io.netty.handler.codec.mqtt.MqttSubAckPayload;
import io.netty.handler.codec.mqtt.MqttSubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
-import io.netty.util.CharsetUtil;
-import org.apache.commons.lang3.StringUtils;
import org.apache.shenyu.common.utils.Singleton;
import org.apache.shenyu.protocol.mqtt.repositories.ChannelRepository;
import org.apache.shenyu.protocol.mqtt.repositories.MqttSession;
@@ -80,8 +78,8 @@ public class Subscribe extends MessageType {
}
for (String ackTopic : ackTopics) {
- String message =
Singleton.INST.get(TopicRepository.class).get(ackTopic);
- if (StringUtils.isNotEmpty(message)) {
+ byte[] message =
Singleton.INST.get(TopicRepository.class).get(ackTopic);
+ if (Objects.nonNull(message)) {
sendSubMessage(ackTopic, message, packetId, channel);
}
}
@@ -117,10 +115,10 @@ public class Subscribe extends MessageType {
* @param packetId packetId
* @param channel channel
*/
- private void sendSubMessage(final String topic, final String message,
final int packetId, final Channel channel) {
+ private void sendSubMessage(final String topic, final byte[] message,
final int packetId, final Channel channel) {
MqttFixedHeader fixedHeader = new
MqttFixedHeader(MqttMessageType.PUBLISH, false, AT_MOST_ONCE, true, 0);
MqttPublishVariableHeader varHeader = new
MqttPublishVariableHeader(topic, packetId);
- MqttPublishMessage mqttPublishMessage = new
MqttPublishMessage(fixedHeader, varHeader, Unpooled.copiedBuffer(message,
CharsetUtil.UTF_8));
+ MqttPublishMessage mqttPublishMessage = new
MqttPublishMessage(fixedHeader, varHeader, Unpooled.wrappedBuffer(message));
channel.writeAndFlush(mqttPublishMessage);
}
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/TopicRepository.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/TopicRepository.java
index eaf053d3ba..f8c196dbd6 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/TopicRepository.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/repositories/TopicRepository.java
@@ -26,12 +26,12 @@ import java.util.concurrent.ConcurrentHashMap;
* Save the posted message.
* {@link org.apache.shenyu.protocol.mqtt.agent.MessageAgent}
*/
-public class TopicRepository implements BaseRepository<String, String> {
+public class TopicRepository implements BaseRepository<String, byte[]> {
- private static final Map<String, String> TOPIC_FACTORY = new
ConcurrentHashMap<>();
+ private static final Map<String, byte[]> TOPIC_FACTORY = new
ConcurrentHashMap<>();
@Override
- public void add(final String topic, final String message) {
+ public void add(final String topic, final byte[] message) {
//// todo MessageAgent.java. Carry out message processing and
processing
CompletableFuture.runAsync(() -> TOPIC_FACTORY.put(topic, message));
}
@@ -42,7 +42,7 @@ public class TopicRepository implements
BaseRepository<String, String> {
}
@Override
- public String get(final String topic) {
+ public byte[] get(final String topic) {
return TOPIC_FACTORY.getOrDefault(topic, null);
}
diff --git
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
index 72fe5bc5eb..42f72bd51a 100644
---
a/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
+++
b/shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/PublishTest.java
@@ -18,6 +18,7 @@
package org.apache.shenyu.protocol.mqtt;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
@@ -34,6 +35,7 @@ import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import io.netty.handler.codec.mqtt.MqttVersion;
import io.netty.util.CharsetUtil;
@@ -48,6 +50,9 @@ import org.awaitility.core.ThrowingRunnable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.ArgumentCaptor;
import java.nio.charset.StandardCharsets;
@@ -55,8 +60,11 @@ import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Objects;
+import java.util.stream.Stream;
import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -145,10 +153,25 @@ public final class PublishTest {
new MqttContext().setPassword(null);
}
- @Test
- public void retainedPublishStoresMessage() {
- new Publish().publish(publisherContext(),
publishMessage(RETAINED_TOPIC, PAYLOAD, true));
- awaitAssert(() -> assertEquals(PAYLOAD,
TOPIC_REPOSITORY.get(RETAINED_TOPIC)));
+ @ParameterizedTest
+ @MethodSource("retainedPayloads")
+ public void retainedPublishReplaysOriginalBytes(final byte[] expected,
final ByteBuf payload) {
+ int readerIndex = payload.readerIndex();
+ MqttPublishMessage message = new MqttPublishMessage(
+ new MqttFixedHeader(MqttMessageType.PUBLISH, false,
MqttQoS.AT_LEAST_ONCE, true, 0),
+ new MqttPublishVariableHeader(RETAINED_TOPIC,
PUBLISHER_PACKET_ID), payload);
+ try {
+ new Publish().publish(publisherContext(), message);
+ assertEquals(readerIndex, payload.readerIndex());
+ awaitAssert(() -> {
+ assertArrayEquals(expected,
TOPIC_REPOSITORY.get(RETAINED_TOPIC));
+ assertEquals(1, payload.refCnt());
+ });
+ payload.setZero(readerIndex, payload.readableBytes());
+ } finally {
+ message.release();
+ }
+ assertRetainedReplay(RETAINED_TOPIC, expected);
}
@Test
@@ -157,17 +180,6 @@ public final class PublishTest {
assertNull(TOPIC_REPOSITORY.get(NON_RETAINED_TOPIC));
}
- @Test
- public void retainedPublishReadsMessageFromDirectPayload() {
- ByteBuf payload =
Unpooled.directBuffer().writeBytes(PAYLOAD.getBytes(StandardCharsets.UTF_8));
- try {
- new Publish().publish(publisherContext(),
publishMessage(RETAINED_TOPIC, payload, true));
- awaitAssert(() -> assertEquals(PAYLOAD,
TOPIC_REPOSITORY.get(RETAINED_TOPIC)));
- } finally {
- payload.release();
- }
- }
-
@Test
public void publishBeforeConnectClosesChannel() {
EmbeddedChannel channel = channel(false);
@@ -188,7 +200,7 @@ public final class PublishTest {
new Connect().connect(ctx, connectMessage());
new Publish().publish(ctx, publishMessage(END_TO_END_TOPIC, PAYLOAD,
true));
- awaitAssert(() -> assertEquals(PAYLOAD,
TOPIC_REPOSITORY.get(END_TO_END_TOPIC)));
+ awaitAssert(() ->
assertArrayEquals(PAYLOAD.getBytes(StandardCharsets.UTF_8),
TOPIC_REPOSITORY.get(END_TO_END_TOPIC)));
assertEquals(CLIENT_ID, CHANNEL_REPOSITORY.get(channel));
CHANNEL_REPOSITORY.remove(channel);
@@ -196,15 +208,28 @@ public final class PublishTest {
}
@Test
- public void zeroByteRetainedPublishClearsRetainedMessage() {
+ public void retainedPublishReplacesAndClearsMessage() {
Publish publish = new Publish();
ChannelHandlerContext ctx = publisherContext();
- publish.publish(ctx, publishMessage(CLEARED_TOPIC, PAYLOAD, true));
- awaitAssert(() -> assertEquals(PAYLOAD,
TOPIC_REPOSITORY.get(CLEARED_TOPIC)));
+ byte[][] messages = {{0, (byte) 0xff, (byte) 0x80, 0x41}, {(byte)
0xfe, 0, 0x42}};
+ for (byte[] bytes : messages) {
+ MqttPublishMessage message = publishMessage(CLEARED_TOPIC,
Unpooled.wrappedBuffer(bytes), true);
+ try {
+ publish.publish(ctx, message);
+ awaitAssert(() -> {
+ assertArrayEquals(bytes,
TOPIC_REPOSITORY.get(CLEARED_TOPIC));
+ assertEquals(1, message.refCnt());
+ });
+ } finally {
+ message.release();
+ }
+ assertRetainedReplay(CLEARED_TOPIC, bytes);
+ }
publish.publish(ctx, publishMessage(CLEARED_TOPIC, "", true));
assertNull(TOPIC_REPOSITORY.get(CLEARED_TOPIC));
+ assertRetainedReplay(CLEARED_TOPIC, null);
}
@Test
@@ -314,6 +339,44 @@ public final class PublishTest {
assertEquals(PUBLISHER_PACKET_ID, ((MqttMessageIdVariableHeader)
pubRec.variableHeader()).messageId());
}
+ private static Stream<Arguments> retainedPayloads() {
+ byte[] text = "hello 中文".getBytes(StandardCharsets.UTF_8);
+ byte[] binary = {0, (byte) 0xff, (byte) 0x80, 0x41};
+ ByteBuf sliced =
Unpooled.buffer().writeByte(0x11).writeByte(0x22).writeBytes(binary).writeByte(0x33)
+ .slice(1, binary.length + 2).setIndex(1, binary.length + 1);
+ ByteBuf direct =
Unpooled.directBuffer().writeByte(0x11).writeBytes(binary).writeByte(0x22)
+ .setIndex(1, binary.length + 1);
+ return Stream.of(
+ Arguments.of(text, Unpooled.copiedBuffer(text)),
+ Arguments.of(binary, sliced),
+ Arguments.of(binary, direct));
+ }
+
+ private void assertRetainedReplay(final String topic, final byte[]
expected) {
+ EmbeddedChannel channel = channel(true);
+ CHANNEL_REPOSITORY.add(channel, CLIENT_ID);
+ try {
+ new Subscribe().subscribe(channel.pipeline().lastContext(),
MqttMessageBuilders.subscribe()
+ .messageId(1).addSubscription(MqttQoS.AT_LEAST_ONCE,
topic).build());
+ boolean replayed = false;
+ for (MqttMessage message = channel.readOutbound();
Objects.nonNull(message); message = channel.readOutbound()) {
+ try {
+ if (message instanceof MqttPublishMessage &&
message.fixedHeader().isRetain()) {
+ replayed = true;
+ assertArrayEquals(expected,
ByteBufUtil.getBytes(((MqttPublishMessage) message).payload()));
+ }
+ } finally {
+ ReferenceCountUtil.release(message);
+ }
+ }
+ assertEquals(Objects.nonNull(expected), replayed);
+ } finally {
+ SUBSCRIBE_REPOSITORY.remove(channel);
+ CHANNEL_REPOSITORY.remove(channel);
+ channel.finishAndReleaseAll();
+ }
+ }
+
/**
* Creates a real channel, optionally already connected, so that {@link
Publish} can
* read the connection attribute and close the channel as it does in
production.
