From 56cd9538549526c2bfd321b714e82635e42d9937 Mon Sep 17 00:00:00 2001 From: JIHUN KIM Date: Sat, 20 Jun 2026 23:03:30 +0900 Subject: [PATCH] Fix MQTT 5 QoS 2 pub reply encoding (#296) Motivation: MQTT 5 PUBREC, PUBREL, and PUBCOMP APIs accept reason codes and properties, but MqttEndpointImpl used a variable header type that Netty does not encode them from. As a result, supplied reason codes and properties were not written on the wire. Changes: Use MqttPubReplyMessageVariableHeader for PUBREC, PUBREL, and PUBCOMP replies. Add wire-level MQTT 5 regression tests covering PUBREC, PUBREL, and PUBCOMP reason codes and properties. Signed-off-by: Jihun Kim (cherry picked from commit d2a720e262f35f17fa7480024bfba95d472c8e98) --- .../io/vertx/mqtt/impl/MqttEndpointImpl.java | 12 +- .../server/Mqtt5ServerQos2PubReplyTest.java | 359 ++++++++++++++++++ 2 files changed, 365 insertions(+), 6 deletions(-) create mode 100644 src/test/java/io/vertx/mqtt/test/server/Mqtt5ServerQos2PubReplyTest.java diff --git a/src/main/java/io/vertx/mqtt/impl/MqttEndpointImpl.java b/src/main/java/io/vertx/mqtt/impl/MqttEndpointImpl.java index 480eb2d0..60d2342a 100644 --- a/src/main/java/io/vertx/mqtt/impl/MqttEndpointImpl.java +++ b/src/main/java/io/vertx/mqtt/impl/MqttEndpointImpl.java @@ -526,8 +526,8 @@ public MqttEndpointImpl publishReceived(int publishMessageId) { public MqttEndpointImpl publishReceived(int publishMessageId, MqttPubRecReasonCode reasonCode, MqttProperties properties) { MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0); - MqttMessageIdAndPropertiesVariableHeader variableHeader = - new MqttMessageIdAndPropertiesVariableHeader(publishMessageId, properties); + MqttPubReplyMessageVariableHeader variableHeader = + new MqttPubReplyMessageVariableHeader(publishMessageId, reasonCode.value(), properties); io.netty.handler.codec.mqtt.MqttMessage pubrec = MqttMessageFactory.newMessage(fixedHeader, variableHeader, null); @@ -544,8 +544,8 @@ public MqttEndpointImpl publishRelease(int publishMessageId) { public MqttEndpointImpl publishRelease(int publishMessageId, MqttPubRelReasonCode reasonCode, MqttProperties properties) { MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0); - MqttMessageIdAndPropertiesVariableHeader variableHeader = - new MqttMessageIdAndPropertiesVariableHeader(publishMessageId, properties); + MqttPubReplyMessageVariableHeader variableHeader = + new MqttPubReplyMessageVariableHeader(publishMessageId, reasonCode.value(), properties); io.netty.handler.codec.mqtt.MqttMessage pubrel = MqttMessageFactory.newMessage(fixedHeader, variableHeader, null); @@ -563,8 +563,8 @@ public MqttEndpointImpl publishComplete(int publishMessageId, MqttPubCompReasonC MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0); - MqttMessageIdAndPropertiesVariableHeader variableHeader = - new MqttMessageIdAndPropertiesVariableHeader(publishMessageId, properties); + MqttPubReplyMessageVariableHeader variableHeader = + new MqttPubReplyMessageVariableHeader(publishMessageId, reasonCode.value(), properties); io.netty.handler.codec.mqtt.MqttMessage pubcomp = MqttMessageFactory.newMessage(fixedHeader, variableHeader, null); diff --git a/src/test/java/io/vertx/mqtt/test/server/Mqtt5ServerQos2PubReplyTest.java b/src/test/java/io/vertx/mqtt/test/server/Mqtt5ServerQos2PubReplyTest.java new file mode 100644 index 00000000..2f6278ad --- /dev/null +++ b/src/test/java/io/vertx/mqtt/test/server/Mqtt5ServerQos2PubReplyTest.java @@ -0,0 +1,359 @@ +/* + * Copyright 2016 Red Hat Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.vertx.mqtt.test.server; + +import io.netty.bootstrap.Bootstrap; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelPipeline; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.SocketChannel; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.codec.mqtt.MqttConnectMessage; +import io.netty.handler.codec.mqtt.MqttConnectPayload; +import io.netty.handler.codec.mqtt.MqttConnectVariableHeader; +import io.netty.handler.codec.mqtt.MqttDecoder; +import io.netty.handler.codec.mqtt.MqttEncoder; +import io.netty.handler.codec.mqtt.MqttFixedHeader; +import io.netty.handler.codec.mqtt.MqttMessage; +import io.netty.handler.codec.mqtt.MqttMessageFactory; +import io.netty.handler.codec.mqtt.MqttMessageType; +import io.netty.handler.codec.mqtt.MqttProperties; +import io.netty.handler.codec.mqtt.MqttPubReplyMessageVariableHeader; +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.MqttVersion; +import io.netty.util.ReferenceCountUtil; +import io.vertx.core.buffer.Buffer; +import io.vertx.ext.unit.TestContext; +import io.vertx.mqtt.MqttEndpoint; +import io.vertx.mqtt.messages.codes.MqttPubCompReasonCode; +import io.vertx.mqtt.messages.codes.MqttPubRecReasonCode; +import io.vertx.mqtt.messages.codes.MqttPubRelReasonCode; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +/** + * MQTT 5 server testing about QoS 2 PUB reply reason codes and properties. + */ +public class Mqtt5ServerQos2PubReplyTest extends MqttServerBaseTest { + + private static final String MQTT_TOPIC = "/my_topic"; + private static final String MQTT_MESSAGE = "Hello Vert.x MQTT Server"; + private static final int CLIENT_PUBLISH_MESSAGE_ID = 7; + private static final int SERVER_PUBLISH_MESSAGE_ID = 9; + private static final String PUBREC_REASON = "pubrec-reason"; + private static final String PUBREL_REASON = "pubrel-reason"; + private static final String PUBCOMP_REASON = "pubcomp-reason"; + + private TestMode testMode; + + @Before + public void before(TestContext context) { + this.setUp(context); + } + + @After + public void after(TestContext context) { + this.tearDown(context); + } + + @Test + public void publishReceivedAndCompleteIncludeMqtt5ReasonCodeAndProperties(TestContext context) throws Exception { + this.testMode = TestMode.CLIENT_PUBLISH; + + try (RawMqttClient client = RawMqttClient.connect()) { + client.write(connectMessage("client-publish")); + client.await(MqttMessageType.CONNACK); + + client.write(publishMessage(CLIENT_PUBLISH_MESSAGE_ID)); + + MqttMessage pubrec = client.await(MqttMessageType.PUBREC); + assertPubReply(context, + pubrec, + CLIENT_PUBLISH_MESSAGE_ID, + MqttPubRecReasonCode.NO_MATCHING_SUBSCRIBERS.value(), + PUBREC_REASON); + + client.write(pubReply(MqttMessageType.PUBREL, + CLIENT_PUBLISH_MESSAGE_ID, + MqttPubRelReasonCode.SUCCESS.value(), + MqttProperties.NO_PROPERTIES)); + + MqttMessage pubcomp = client.await(MqttMessageType.PUBCOMP); + assertPubReply(context, + pubcomp, + CLIENT_PUBLISH_MESSAGE_ID, + MqttPubCompReasonCode.PACKET_IDENTIFIER_NOT_FOUND.value(), + PUBCOMP_REASON); + } + } + + @Test + public void publishReleaseIncludesMqtt5ReasonCodeAndProperties(TestContext context) throws Exception { + this.testMode = TestMode.SERVER_PUBLISH; + + try (RawMqttClient client = RawMqttClient.connect()) { + client.write(connectMessage("server-publish")); + client.await(MqttMessageType.CONNACK); + + MqttMessage publish = client.await(MqttMessageType.PUBLISH); + try { + context.assertEquals(SERVER_PUBLISH_MESSAGE_ID, ((MqttPublishMessage)publish).variableHeader().packetId()); + } finally { + ReferenceCountUtil.release(publish); + } + + client.write(pubReply(MqttMessageType.PUBREC, + SERVER_PUBLISH_MESSAGE_ID, + MqttPubRecReasonCode.SUCCESS.value(), + MqttProperties.NO_PROPERTIES)); + + MqttMessage pubrel = client.await(MqttMessageType.PUBREL); + assertPubReply(context, + pubrel, + SERVER_PUBLISH_MESSAGE_ID, + MqttPubRelReasonCode.PACKET_IDENTIFIER_NOT_FOUND.value(), + PUBREL_REASON); + + client.write(pubReply(MqttMessageType.PUBCOMP, + SERVER_PUBLISH_MESSAGE_ID, + MqttPubCompReasonCode.SUCCESS.value(), + MqttProperties.NO_PROPERTIES)); + } + } + + @Override + protected void endpointHandler(MqttEndpoint endpoint, TestContext context) { + if (this.testMode == TestMode.CLIENT_PUBLISH) { + endpoint.publishHandler(message -> { + endpoint.publishReceived(message.messageId(), + MqttPubRecReasonCode.NO_MATCHING_SUBSCRIBERS, + reasonProperties(PUBREC_REASON)); + }).publishReleaseMessageHandler(message -> { + endpoint.publishComplete(message.messageId(), + MqttPubCompReasonCode.PACKET_IDENTIFIER_NOT_FOUND, + reasonProperties(PUBCOMP_REASON)); + }); + } else { + endpoint.publishReceivedHandler(messageId -> { + endpoint.publishRelease(messageId, + MqttPubRelReasonCode.PACKET_IDENTIFIER_NOT_FOUND, + reasonProperties(PUBREL_REASON)); + }); + } + + endpoint.accept(false); + + if (this.testMode == TestMode.SERVER_PUBLISH) { + endpoint.publish(MQTT_TOPIC, + Buffer.buffer(MQTT_MESSAGE), + MqttQoS.EXACTLY_ONCE, + false, + false, + SERVER_PUBLISH_MESSAGE_ID, + MqttProperties.NO_PROPERTIES) + .onComplete(context.asyncAssertSuccess()); + } + } + + private static MqttConnectMessage connectMessage(String clientId) { + MqttFixedHeader fixedHeader = + new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0); + + MqttConnectVariableHeader variableHeader = + new MqttConnectVariableHeader(MqttVersion.MQTT_5.protocolName(), + MqttVersion.MQTT_5.protocolLevel(), + false, + false, + false, + 0, + false, + true, + 60, + MqttProperties.NO_PROPERTIES); + + byte[] willMessage = null; + byte[] password = null; + MqttConnectPayload payload = + new MqttConnectPayload(clientId, null, willMessage, null, password); + + return (MqttConnectMessage)MqttMessageFactory.newMessage(fixedHeader, variableHeader, payload); + } + + private static MqttPublishMessage publishMessage(int messageId) { + MqttFixedHeader fixedHeader = + new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.EXACTLY_ONCE, false, 0); + MqttPublishVariableHeader variableHeader = + new MqttPublishVariableHeader(MQTT_TOPIC, messageId, MqttProperties.NO_PROPERTIES); + ByteBuf payload = Unpooled.copiedBuffer(MQTT_MESSAGE.getBytes(StandardCharsets.UTF_8)); + return new MqttPublishMessage(fixedHeader, variableHeader, payload); + } + + private static MqttMessage pubReply(MqttMessageType messageType, int messageId, byte reasonCode, + MqttProperties properties) { + MqttFixedHeader fixedHeader = + new MqttFixedHeader(messageType, false, pubReplyQoS(messageType), false, 0); + MqttPubReplyMessageVariableHeader variableHeader = + new MqttPubReplyMessageVariableHeader(messageId, reasonCode, properties); + return new MqttMessage(fixedHeader, variableHeader); + } + + private static MqttQoS pubReplyQoS(MqttMessageType messageType) { + return messageType == MqttMessageType.PUBREL ? MqttQoS.AT_LEAST_ONCE : MqttQoS.AT_MOST_ONCE; + } + + private static MqttProperties reasonProperties(String reason) { + MqttProperties properties = new MqttProperties(); + properties.add(new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.REASON_STRING.value(), reason)); + return properties; + } + + private static void assertPubReply(TestContext context, MqttMessage message, int expectedMessageId, + byte expectedReasonCode, String expectedReason) { + try { + MqttPubReplyMessageVariableHeader variableHeader = + (MqttPubReplyMessageVariableHeader)message.variableHeader(); + + context.assertEquals(expectedMessageId, variableHeader.messageId()); + context.assertEquals(Byte.toUnsignedInt(expectedReasonCode), Byte.toUnsignedInt(variableHeader.reasonCode())); + + MqttProperties.MqttProperty reasonProperty = + variableHeader.properties().getProperty(MqttProperties.MqttPropertyType.REASON_STRING.value()); + context.assertNotNull(reasonProperty); + context.assertEquals(expectedReason, reasonProperty.value()); + } finally { + ReferenceCountUtil.release(message); + } + } + + private enum TestMode { + CLIENT_PUBLISH, + SERVER_PUBLISH + } + + private static class RawMqttClient implements AutoCloseable { + + private final EventLoopGroup group; + private final Channel channel; + private final BlockingQueue messages; + private final BlockingQueue failures; + + private RawMqttClient(EventLoopGroup group, Channel channel, BlockingQueue messages, + BlockingQueue failures) { + this.group = group; + this.channel = channel; + this.messages = messages; + this.failures = failures; + } + + static RawMqttClient connect() throws InterruptedException { + EventLoopGroup group = new NioEventLoopGroup(); + BlockingQueue messages = new LinkedBlockingQueue<>(); + BlockingQueue failures = new LinkedBlockingQueue<>(); + + Bootstrap bootstrap = new Bootstrap(); + bootstrap + .group(group) + .channel(NioSocketChannel.class) + .handler(new ChannelInitializer() { + @Override + protected void initChannel(SocketChannel ch) { + ChannelPipeline pipeline = ch.pipeline(); + pipeline.addLast("mqttDecoder", new MqttDecoder()); + pipeline.addLast("mqttEncoder", MqttEncoder.INSTANCE); + pipeline.addLast("messageHandler", new ChannelInboundHandlerAdapter() { + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + if (msg instanceof MqttMessage) { + messages.add((MqttMessage)msg); + } else { + ReferenceCountUtil.release(msg); + } + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + failures.add(cause); + ctx.close(); + } + }); + } + }); + + try { + ChannelFuture connectFuture = bootstrap.connect(MQTT_SERVER_HOST, MQTT_SERVER_PORT).sync(); + return new RawMqttClient(group, connectFuture.channel(), messages, failures); + } catch (InterruptedException | RuntimeException e) { + group.shutdownGracefully().awaitUninterruptibly(); + throw e; + } + } + + void write(MqttMessage message) throws InterruptedException { + this.channel.writeAndFlush(message).sync(); + } + + MqttMessage await(MqttMessageType messageType) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (System.nanoTime() < deadline) { + Throwable failure = this.failures.poll(); + if (failure != null) { + throw new AssertionError(failure); + } + MqttMessage message = this.messages.poll(100, TimeUnit.MILLISECONDS); + if (message != null) { + if (message.fixedHeader().messageType() == messageType) { + return message; + } + ReferenceCountUtil.release(message); + } + } + throw new AssertionError("Timed out waiting for " + messageType); + } + + @Override + public void close() throws InterruptedException { + try { + this.channel.close().sync(); + } finally { + releaseQueuedMessages(); + this.group.shutdownGracefully().sync(); + } + } + + private void releaseQueuedMessages() { + MqttMessage message; + while ((message = this.messages.poll()) != null) { + ReferenceCountUtil.release(message); + } + } + } +}