Skip to content

ARTEMIS-6190 MQTT Will message properties propagation #6616

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
borisalekseev wants to merge 1 commit into
base: main
Choose a base branch
from
Open

ARTEMIS-6190 MQTT Will message properties propagation #6616

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 10 additions & 9 deletions ...l/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnectionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,6 @@
*/
package org.apache.activemq.artemis.core.protocol.mqtt;

import java.util.List;

import io.netty.buffer.ByteBufAllocator;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttProperties;
Expand Down Expand Up @@ -93,18 +91,21 @@ synchronized void connect(MqttConnectMessage connect, String validatedUser, Stri
session.getState().setWillRetain(connect.variableHeader().isWillRetain());
session.getState().setWillTopic(connect.payload().willTopic());
session.getState().setWillStatus(MQTTSessionState.WillStatus.NOT_SENT);
session.getState().setWillDelayInterval(0);
session.getState().setWillPublishProperties(MqttProperties.NO_PROPERTIES);

if (session.getVersion() == MQTTVersion.MQTT_5) {
MqttProperties willProperties = connect.payload().willProperties();
if (willProperties != null) {
MqttProperties.MqttProperty willDelayInterval = willProperties.getProperty(WILL_DELAY_INTERVAL.value());
if (willDelayInterval != null) {
session.getState().setWillDelayInterval((int) willDelayInterval.value());
}
List<? extends MqttProperties.MqttProperty> userProperties = willProperties.getProperties(MqttProperties.MqttPropertyType.USER_PROPERTY.value());
if (userProperties != null) {
session.getState().setWillUserProperties(userProperties);
MqttProperties publishProperties = new MqttProperties();
for (MqttProperties.MqttProperty property : willProperties.listAll()) {
if (property.propertyId() == WILL_DELAY_INTERVAL.value()) {
session.getState().setWillDelayInterval((int) property.value());
} else {
publishProperties.add(property);
}
}
session.getState().setWillPublishProperties(publishProperties);
}
}
}
Expand Down
19 changes: 8 additions & 11 deletions ...tt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSession.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import io.netty.buffer.EmptyByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.handler.codec.mqtt.MqttMessageBuilders;
import io.netty.handler.codec.mqtt.MqttProperties;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.apache.activemq.artemis.api.core.ActiveMQSecurityException;
Expand Down Expand Up @@ -271,16 +270,14 @@ public void sendWillMessage() {
if (state.getWillStatus() == MQTTSessionState.WillStatus.NOT_SENT) {
try {
state.setWillStatus(MQTTSessionState.WillStatus.SENDING);
MqttProperties properties;
if (state.getWillUserProperties() == null) {
properties = MqttProperties.NO_PROPERTIES;
} else {
properties = new MqttProperties();
for (MqttProperties.MqttProperty userProperty : state.getWillUserProperties()) {
properties.add(userProperty);
}
}
MqttPublishMessage publishMessage = MqttMessageBuilders.publish().messageId(0).qos(MqttQoS.valueOf(state.getWillQoSLevel())).retained(state.isWillRetain()).topicName(state.getWillTopic()).payload(state.getWillMessage() == null ? new EmptyByteBuf(PooledByteBufAllocator.DEFAULT) : state.getWillMessage()).properties(properties).build();
MqttPublishMessage publishMessage = MqttMessageBuilders
.publish()
.messageId(0)
.qos(MqttQoS.valueOf(state.getWillQoSLevel()))
.retained(state.isWillRetain())
.topicName(state.getWillTopic())
.payload(state.getWillMessage() == null ? new EmptyByteBuf(PooledByteBufAllocator.DEFAULT) : state.getWillMessage())
.properties(state.getWillPublishProperties()).build();
logger.debug("{} sending will message: {}", this, publishMessage);
getMqttPublishManager().sendToQueue(publishMessage, true);
state.setWillStatus(MQTTSessionState.WillStatus.SENT);
Expand Down
11 changes: 6 additions & 5 deletions ...otocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSessionState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ public class MQTTSessionState {

private long willDelayInterval = 0;

private List<? extends MqttProperties.MqttProperty> willUserProperties;
private MqttProperties willPublishProperties = MqttProperties.NO_PROPERTIES;

private WillStatus willStatus = WillStatus.NOT_SENT;

Expand Down Expand Up @@ -163,6 +163,7 @@ public synchronized void clear() throws Exception {
willStatus = WillStatus.NOT_SENT;
failed = false;
willDelayInterval = 0;
willPublishProperties = MqttProperties.NO_PROPERTIES;
willRetain = false;
willTopic = null;
clientMaxPacketSize = 0;
Expand Down Expand Up @@ -318,12 +319,12 @@ public void setWillDelayInterval(long willDelayInterval) {
this.willDelayInterval = willDelayInterval;
}

public void setWillUserProperties(List<? extends MqttProperties.MqttProperty> userProperties) {
this.willUserProperties = userProperties;
public void setWillPublishProperties(MqttProperties willPublishProperties) {
this.willPublishProperties = willPublishProperties;
}

public List<? extends MqttProperties.MqttProperty> getWillUserProperties() {
return willUserProperties;
public MqttProperties getWillPublishProperties() {
return willPublishProperties;
}

public WillStatus getWillStatus() {
Expand Down
46 changes: 38 additions & 8 deletions ...on-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/MQTT5Test.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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.assertNotNull;
Expand Down Expand Up @@ -272,13 +273,16 @@ public void testAddressAutoCreationNegative() throws Exception {
}

/**
* There is no normative statement in the spec about supporting user properties on will messages, but it is implied
* in various places.
* Verifies that the Application Message properties in Will Properties are included in the published Will Message.
*/
@Test
@Timeout(DEFAULT_TIMEOUT_SEC)
public void testWillMessageProperties() throws Exception {
final byte[] WILL = RandomUtil.randomBytes();
final long MESSAGE_EXPIRY_INTERVAL = 60;
final String CONTENT_TYPE = "application/octet-stream";
final String RESPONSE_TOPIC = "/topic/reply";
final byte[] CORRELATION_DATA = RandomUtil.randomBytes();
final String[][] properties = new String[10][2];
for (String[] property : properties) {
property[0] = RandomUtil.randomUUIDString();
Expand All @@ -287,16 +291,18 @@ public void testWillMessageProperties() throws Exception {

// consumer of the will message
MqttClient client1 = createPahoClient("willConsumer");
runAfter(() -> {
if (client1.isConnected()) {
client1.disconnect();
}
client1.close();
});
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<MqttProperties> receivedProperties = new AtomicReference<>();
client1.setCallback(new DefaultMqttCallback() {
@Override
public void messageArrived(String topic, MqttMessage message) {
int i = 0;
for (UserProperty property : message.getProperties().getUserProperties()) {
assertEquals(properties[i][0], property.getKey());
assertEquals(properties[i][1], property.getValue());
i++;
}
receivedProperties.set(message.getProperties());
latch.countDown();
}
});
Expand All @@ -305,7 +311,18 @@ public void messageArrived(String topic, MqttMessage message) {

// consumer to generate the will
MqttClient client2 = createPahoClient("willGenerator");
runAfter(() -> {
if (client2.isConnected()) {
client2.disconnectForcibly(0, 0, false);
}
client2.close();
});
MqttProperties willMessageProperties = new MqttProperties();
willMessageProperties.setPayloadFormat(true);
willMessageProperties.setMessageExpiryInterval(MESSAGE_EXPIRY_INTERVAL);
willMessageProperties.setContentType(CONTENT_TYPE);
willMessageProperties.setResponseTopic(RESPONSE_TOPIC);
willMessageProperties.setCorrelationData(CORRELATION_DATA);
List<UserProperty> userProperties = new ArrayList<>();
for (String[] property : properties) {
userProperties.add(new UserProperty(property[0], property[1]));
Expand All @@ -318,6 +335,19 @@ public void messageArrived(String topic, MqttMessage message) {
client2.connect(options);
client2.disconnectForcibly(0, 0, false);
assertTrue(latch.await(2, TimeUnit.SECONDS));

MqttProperties actualProperties = receivedProperties.get();
assertTrue(actualProperties.getPayloadFormat());
assertTrue(actualProperties.getMessageExpiryInterval() > 0);
assertTrue(actualProperties.getMessageExpiryInterval() <= MESSAGE_EXPIRY_INTERVAL);
assertEquals(CONTENT_TYPE, actualProperties.getContentType());
assertEquals(RESPONSE_TOPIC, actualProperties.getResponseTopic());
assertArrayEquals(CORRELATION_DATA, actualProperties.getCorrelationData());
assertEquals(properties.length, actualProperties.getUserProperties().size());
for (int i = 0; i < properties.length; i++) {
assertEquals(properties[i][0], actualProperties.getUserProperties().get(i).getKey());
assertEquals(properties[i][1], actualProperties.getUserProperties().get(i).getValue());
}
}

/**
Expand Down
Loading