You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@activemq.apache.org by GitBox <gi...@apache.org> on 2022/01/11 02:51:03 UTC
[GitHub] [activemq-artemis] jbertram opened a new pull request #3907: ARTEMIS-8454 Support MQTT 5
jbertram opened a new pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907
MQTT 5 is an OASIS standard which debuted in March 2019. It boasts
numerous improvments over its predecessor (i.e. MQTT 3.1.1) which will
benefit users. These improvements are summarized in the specification
at:
https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901293
The specification describes all the behavior necessary for a client or
server to conform. The spec is highlighted with special "normative"
conformance statements which distill the descriptions into concise terms.
The specification provides a helpful summary of all these statements. See:
https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901292
This commit implements all of the mandatory elements from the
specification and provides tests which are identified using the
corresponding normative conformance statement. *All* normative
conformance statements either have an explicit test or are noted in
comments with an explanation of why an explicit test doesn't exist. See
`org.apache.activemq.artemis.tests.integration.mqtt5` for all those
details.
This commit also includes documentation about how to configure
everything related to the new MQTT 5 features.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#issuecomment-1029572070
I fixed the failing tests in the test-suite. This is ready to be merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#issuecomment-1010355438
I just realized that I used the wrong Jira ID in the commit message and the branch name. I corrected the commit message and the PR title, but I left the branch name as-is to avoid having to create a whole new PR (and lose the comment history).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on a change in pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on a change in pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#discussion_r792921224
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java
##########
@@ -159,89 +176,180 @@ public void act(MqttMessage message) {
break;
case UNSUBACK:
case SUBACK:
+ case PINGREQ: // These are actually handled by the Netty thread directly so this packet should never make it here
case PINGRESP:
case CONNACK: // The server does not instantiate connections therefore any CONNACK received over a connection is an invalid control message.
default:
disconnect(true);
}
} catch (Exception e) {
- log.warn("Error processing Control Packet, Disconnecting Client", e);
+ MQTTLogger.LOGGER.errorProcessingControlPacket(message.toString(), e);
+ if (session.is5()) {
+ sendDisconnect(MQTTReasonCodes.IMPLEMENTATION_SPECIFIC_ERROR);
+ }
disconnect(true);
} finally {
ReferenceCountUtil.release(message);
}
}
- /**
- * Called during connection.
+ /*
+ * Scaffolding for "enhanced authentication" implementation.
+ *
+ * See:
+ * https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901217
+ * https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901256
+ *
+ * Tests for this are in:
+ * org.apache.activemq.artemis.tests.integration.mqtt5.spec.controlpackets.AuthTests
+ * org.apache.activemq.artemis.tests.integration.mqtt5.spec.EnhancedAuthenticationTests
*
- * @param connect
+ * This should integrate somehow with our existing SASL implementation for challenge/response conversations.
*/
+ void handleAuth(MqttMessage auth) throws Exception {
+ byte[] authenticationData = MQTTUtil.getProperty(byte[].class, ((MqttReasonCodeAndPropertiesVariableHeader)auth.variableHeader()).properties(), AUTHENTICATION_DATA);
+ String authenticationMethod = MQTTUtil.getProperty(String.class, ((MqttReasonCodeAndPropertiesVariableHeader)auth.variableHeader()).properties(), AUTHENTICATION_METHOD);
+
+ MqttReasonCodeAndPropertiesVariableHeader header = (MqttReasonCodeAndPropertiesVariableHeader) auth.variableHeader();
+ if (header.reasonCode() == MQTTReasonCodes.RE_AUTHENTICATE) {
+
+ } else if (header.reasonCode() == MQTTReasonCodes.CONTINUE_AUTHENTICATION) {
+
+ } else if (header.reasonCode() == MQTTReasonCodes.SUCCESS) {
+
+ }
+ }
+
void handleConnect(MqttConnectMessage connect) throws Exception {
- final String username = connect.payload().userName();
- final String password = connect.payload().passwordInBytes() == null ? null : new String( connect.payload().passwordInBytes(), CharsetUtil.UTF_8);
- final String validatedUser = server.validateUser(username, password, session.getConnection(), session.getProtocolManager().getSecurityDomain());
- if (connection.getTransportConnection().getRedirectTo() == null ||
- !protocolManager.getRedirectHandler().redirect(connection, session, connect)) {
- connectionEntry.ttl = connect.variableHeader().keepAliveTimeSeconds() * 1500L;
-
- String clientId = connect.payload().clientIdentifier();
- session.getConnectionManager().connect(clientId, username, password, connect.variableHeader().isWillFlag(), connect.payload().willMessageInBytes(), connect.payload().willTopic(), connect.variableHeader().isWillRetain(), connect.variableHeader().willQos(), connect.variableHeader().isCleanSession(), validatedUser);
+ if (connection.getTransportConnection().getRedirectTo() == null || !protocolManager.getRedirectHandler().redirect(connection, session, connect)) {
+ /* [MQTT-3.1.2-2] Reject unsupported clients. */
+ int packetVersion = connect.variableHeader().version();
+ if (packetVersion != MqttVersion.MQTT_3_1.protocolLevel() &&
+ packetVersion != MqttVersion.MQTT_3_1_1.protocolLevel() &&
+ packetVersion != MqttVersion.MQTT_5.protocolLevel()) {
+
+ {
+ /*
+ * The return code is different here for 3.1.1 vs 5, but since the broker supports *both* it's not clear
+ * which one should be chosen. Perhaps an acceptor property could be set to decide. Leaving the existing
Review comment:
I think that's fair. I'll push that change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on pull request #3907: ARTEMIS-8454 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#issuecomment-1009987982
@gtully, as noted in the documentation (which you probably saw) there currently are no supported authentication methods for the "Enhanced Authentication" which was introduced in MQTT 5. However, the scaffolding is there so it should be relatively easy to integrate (assuming you're familiar with SASL). See `org.apache.activemq.artemis.core.protocol.mqtt.MQTTProtocolHandler#handleAuth`.
Given your familiarity with SASL I was hoping to get your help here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on a change in pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on a change in pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#discussion_r792954441
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManager.java
##########
@@ -185,45 +243,83 @@ public void handleBuffer(RemotingConnection connection, ActiveMQBuffer buffer) {
@Override
public void addChannelHandlers(ChannelPipeline pipeline) {
pipeline.addLast(MqttEncoder.INSTANCE);
- pipeline.addLast(new MqttDecoder(MQTTUtil.MAX_MESSAGE_SIZE));
+ /*
+ * If we use the value from getMaximumPacketSize() here anytime a client sends a packet that's too large it
+ * will receive a DISCONNECT with a reason code of 0x81 instead of 0x95 like it should according to the spec.
+ * See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901086:
+ *
+ * If a Server receives a packet whose size exceeds this limit, this is a Protocol Error, the Server uses
+ * DISCONNECT with Reason Code 0x95 (Packet too large)...
+ *
+ * Therefore we check manually in org.apache.activemq.artemis.core.protocol.mqtt.MQTTProtocolHandler.handlePublish
+ */
+ pipeline.addLast(new MqttDecoder(MQTTUtil.MAX_PACKET_SIZE));
pipeline.addLast(new MQTTProtocolHandler(server, this));
}
/**
- * The protocol handler passes us an 8 byte long array from the transport. We sniff these first 8 bytes to see
- * if they match the first 8 bytes from MQTT Connect packet. In many other protocols the protocol name is the first
- * thing sent on the wire. However, in MQTT the protocol name doesn't come until later on in the CONNECT packet.
- *
- * In order to fully identify MQTT protocol via protocol name, we need up to 12 bytes. However, we can use other
- * information from the connect packet to infer that the MQTT protocol is being used. This is enough to identify MQTT
- * and add the Netty codec in the pipeline. The Netty codec takes care of things from here.
- *
- * MQTT CONNECT PACKET: See MQTT 3.1.1 Spec for more info.
- *
- * Byte 1: Fixed Header Packet Type. 0b0001000 (16) = MQTT Connect
- * Byte 2-[N]: Remaining length of the Connect Packet (encoded with 1-4 bytes).
- *
- * The next set of bytes represents the UTF8 encoded string MQTT (MQTT 3.1.1) or MQIsdp (MQTT 3.1)
- * Byte N: UTF8 MSB must be 0
- * Byte N+1: UTF8 LSB must be (4(MQTT) or 6(MQIsdp))
- * Byte N+1: M (first char from the protocol name).
- *
- * Max no bytes used in the sequence = 8.
+ * Relevant portions of the specs we support:
+ * MQTT 3.1 - https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#connect
+ * MQTT 3.1.1 - http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718028
+ * MQTT 5 - https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901033
*/
@Override
public boolean isProtocol(byte[] array) {
ByteBuf buf = Unpooled.wrappedBuffer(array);
- if (!(buf.readByte() == 16 && validateRemainingLength(buf) && buf.readByte() == (byte) 0)) return false;
+ // Parse "fixed header"
+ if (!(readByte(buf) == 16 && validateRemainingLength(buf) && readByte(buf) == (byte) 0)) {
+ return false;
+ }
+
+ // Start parsing the Protocol Name
+ byte b = readByte(buf); // LSB
+
+ // MQTT 3.1.1 & 5
+ if (b == 4 &&
+ (readByte(buf) != 77 || // M
+ readByte(buf) != 81 || // Q
+ readByte(buf) != 84 || // T
+ readByte(buf) != 84)) { // T
+ return false;
+ }
+
+ // MQTT 3.1
+ if (b == 6 &&
+ (readByte(buf) != 77 || // M
+ readByte(buf) != 81 || // Q
+ readByte(buf) != 73 || // I
+ readByte(buf) != 115 || // s
+ readByte(buf) != 100 || // d
+ readByte(buf) != 112)) { // p
+ return false;
+ }
+
+ // Protocol Version
+ b = readByte(buf);
Review comment:
You're right. I didn't catch this because I don't actually have a low-level client implementation where I can pass a bad version and validate the response from the broker. I'll omit the version check here so the spec-compliant response can be returned later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gtully commented on a change in pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
gtully commented on a change in pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#discussion_r789626358
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolHandler.java
##########
@@ -159,89 +176,180 @@ public void act(MqttMessage message) {
break;
case UNSUBACK:
case SUBACK:
+ case PINGREQ: // These are actually handled by the Netty thread directly so this packet should never make it here
case PINGRESP:
case CONNACK: // The server does not instantiate connections therefore any CONNACK received over a connection is an invalid control message.
default:
disconnect(true);
}
} catch (Exception e) {
- log.warn("Error processing Control Packet, Disconnecting Client", e);
+ MQTTLogger.LOGGER.errorProcessingControlPacket(message.toString(), e);
+ if (session.is5()) {
+ sendDisconnect(MQTTReasonCodes.IMPLEMENTATION_SPECIFIC_ERROR);
+ }
disconnect(true);
} finally {
ReferenceCountUtil.release(message);
}
}
- /**
- * Called during connection.
+ /*
+ * Scaffolding for "enhanced authentication" implementation.
+ *
+ * See:
+ * https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901217
+ * https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901256
+ *
+ * Tests for this are in:
+ * org.apache.activemq.artemis.tests.integration.mqtt5.spec.controlpackets.AuthTests
+ * org.apache.activemq.artemis.tests.integration.mqtt5.spec.EnhancedAuthenticationTests
*
- * @param connect
+ * This should integrate somehow with our existing SASL implementation for challenge/response conversations.
*/
+ void handleAuth(MqttMessage auth) throws Exception {
+ byte[] authenticationData = MQTTUtil.getProperty(byte[].class, ((MqttReasonCodeAndPropertiesVariableHeader)auth.variableHeader()).properties(), AUTHENTICATION_DATA);
+ String authenticationMethod = MQTTUtil.getProperty(String.class, ((MqttReasonCodeAndPropertiesVariableHeader)auth.variableHeader()).properties(), AUTHENTICATION_METHOD);
+
+ MqttReasonCodeAndPropertiesVariableHeader header = (MqttReasonCodeAndPropertiesVariableHeader) auth.variableHeader();
+ if (header.reasonCode() == MQTTReasonCodes.RE_AUTHENTICATE) {
+
+ } else if (header.reasonCode() == MQTTReasonCodes.CONTINUE_AUTHENTICATION) {
+
+ } else if (header.reasonCode() == MQTTReasonCodes.SUCCESS) {
+
+ }
+ }
+
void handleConnect(MqttConnectMessage connect) throws Exception {
- final String username = connect.payload().userName();
- final String password = connect.payload().passwordInBytes() == null ? null : new String( connect.payload().passwordInBytes(), CharsetUtil.UTF_8);
- final String validatedUser = server.validateUser(username, password, session.getConnection(), session.getProtocolManager().getSecurityDomain());
- if (connection.getTransportConnection().getRedirectTo() == null ||
- !protocolManager.getRedirectHandler().redirect(connection, session, connect)) {
- connectionEntry.ttl = connect.variableHeader().keepAliveTimeSeconds() * 1500L;
-
- String clientId = connect.payload().clientIdentifier();
- session.getConnectionManager().connect(clientId, username, password, connect.variableHeader().isWillFlag(), connect.payload().willMessageInBytes(), connect.payload().willTopic(), connect.variableHeader().isWillRetain(), connect.variableHeader().willQos(), connect.variableHeader().isCleanSession(), validatedUser);
+ if (connection.getTransportConnection().getRedirectTo() == null || !protocolManager.getRedirectHandler().redirect(connection, session, connect)) {
+ /* [MQTT-3.1.2-2] Reject unsupported clients. */
+ int packetVersion = connect.variableHeader().version();
+ if (packetVersion != MqttVersion.MQTT_3_1.protocolLevel() &&
+ packetVersion != MqttVersion.MQTT_3_1_1.protocolLevel() &&
+ packetVersion != MqttVersion.MQTT_5.protocolLevel()) {
+
+ {
+ /*
+ * The return code is different here for 3.1.1 vs 5, but since the broker supports *both* it's not clear
+ * which one should be chosen. Perhaps an acceptor property could be set to decide. Leaving the existing
Review comment:
this does not need to be configured, but if packedVersion > 3 I would use the v5 reason code
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
##########
@@ -284,11 +358,110 @@ private void sendServerMessage(int messageId, ICoreMessage message, int delivery
payload.writeBytes(bodyBuffer.byteBuf());
break;
}
- session.getProtocolHandler().send(messageId, address, qos, isRetain, payload, deliveryCount);
+
+ // [MQTT-3.3.1-2] The DUP flag MUST be set to 0 for all QoS 0 messages.
+ boolean redelivery = qos == 0 ? false : (deliveryCount > 1);
+
+ boolean isRetain = message.getBooleanProperty(MQTT_MESSAGE_RETAIN_KEY);
+ MqttProperties mqttProperties = getPublishProperties(message);
+
+ if (session.is5()) {
+ if (session.getState().getSubscription(message.getAddress()) != null && !session.getState().getSubscription(message.getAddress()).option().isRetainAsPublished()) {
+ isRetain = false;
+ }
+
+ // [MQTT-3.8.3-3] remove property used for no-local implementation
+ message.removeProperty(MessageUtil.CONNECTION_ID_PROPERTY_NAME);
+
+ if (session.getState().getClientTopicAliasMaximum() != null) {
+ Integer alias = session.getState().getServerTopicAlias(address);
+ if (alias == null) {
+ alias = session.getState().addServerTopicAlias(address);
+ if (alias != null) {
+ mqttProperties.add(new MqttProperties.IntegerProperty(TOPIC_ALIAS.value(), alias));
+ }
+ } else {
+ mqttProperties.add(new MqttProperties.IntegerProperty(TOPIC_ALIAS.value(), alias));
+ address = "";
+ }
+ }
+ }
+
+ int remainingLength = MQTTUtil.calculateRemainingLength(address, mqttProperties, payload);
+ MqttFixedHeader header = new MqttFixedHeader(MqttMessageType.PUBLISH, redelivery, MqttQoS.valueOf(qos), isRetain, remainingLength);
+ MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(address, messageId, mqttProperties);
+ MqttPublishMessage publish = new MqttPublishMessage(header, varHeader, payload);
+
+ if (session.is5()) {
+ int size = MQTTUtil.calculateMessageSize(publish);
Review comment:
maybe calculate size if maxSize != 0 to avoid the computation if unlimited
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManager.java
##########
@@ -185,45 +243,83 @@ public void handleBuffer(RemotingConnection connection, ActiveMQBuffer buffer) {
@Override
public void addChannelHandlers(ChannelPipeline pipeline) {
pipeline.addLast(MqttEncoder.INSTANCE);
- pipeline.addLast(new MqttDecoder(MQTTUtil.MAX_MESSAGE_SIZE));
+ /*
+ * If we use the value from getMaximumPacketSize() here anytime a client sends a packet that's too large it
+ * will receive a DISCONNECT with a reason code of 0x81 instead of 0x95 like it should according to the spec.
Review comment:
how bad is that? A 0x81 malformed is not accurate but it is the disconnect that is important? Doing a second size check or a size check higher up may be an unnecessary expense if it can be computed and rejected at the netty layer, even if the reason code is less accurate it may be a good tradeoff to make. I don't see an mqtt app making some decision based on the reason code, but maybe. Or maybe we can fix netty to use 0x95 for v5
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManager.java
##########
@@ -185,45 +243,83 @@ public void handleBuffer(RemotingConnection connection, ActiveMQBuffer buffer) {
@Override
public void addChannelHandlers(ChannelPipeline pipeline) {
pipeline.addLast(MqttEncoder.INSTANCE);
- pipeline.addLast(new MqttDecoder(MQTTUtil.MAX_MESSAGE_SIZE));
+ /*
+ * If we use the value from getMaximumPacketSize() here anytime a client sends a packet that's too large it
+ * will receive a DISCONNECT with a reason code of 0x81 instead of 0x95 like it should according to the spec.
+ * See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901086:
+ *
+ * If a Server receives a packet whose size exceeds this limit, this is a Protocol Error, the Server uses
+ * DISCONNECT with Reason Code 0x95 (Packet too large)...
+ *
+ * Therefore we check manually in org.apache.activemq.artemis.core.protocol.mqtt.MQTTProtocolHandler.handlePublish
+ */
+ pipeline.addLast(new MqttDecoder(MQTTUtil.MAX_PACKET_SIZE));
pipeline.addLast(new MQTTProtocolHandler(server, this));
}
/**
- * The protocol handler passes us an 8 byte long array from the transport. We sniff these first 8 bytes to see
- * if they match the first 8 bytes from MQTT Connect packet. In many other protocols the protocol name is the first
- * thing sent on the wire. However, in MQTT the protocol name doesn't come until later on in the CONNECT packet.
- *
- * In order to fully identify MQTT protocol via protocol name, we need up to 12 bytes. However, we can use other
- * information from the connect packet to infer that the MQTT protocol is being used. This is enough to identify MQTT
- * and add the Netty codec in the pipeline. The Netty codec takes care of things from here.
- *
- * MQTT CONNECT PACKET: See MQTT 3.1.1 Spec for more info.
- *
- * Byte 1: Fixed Header Packet Type. 0b0001000 (16) = MQTT Connect
- * Byte 2-[N]: Remaining length of the Connect Packet (encoded with 1-4 bytes).
- *
- * The next set of bytes represents the UTF8 encoded string MQTT (MQTT 3.1.1) or MQIsdp (MQTT 3.1)
- * Byte N: UTF8 MSB must be 0
- * Byte N+1: UTF8 LSB must be (4(MQTT) or 6(MQIsdp))
- * Byte N+1: M (first char from the protocol name).
- *
- * Max no bytes used in the sequence = 8.
+ * Relevant portions of the specs we support:
+ * MQTT 3.1 - https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#connect
+ * MQTT 3.1.1 - http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718028
+ * MQTT 5 - https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901033
*/
@Override
public boolean isProtocol(byte[] array) {
ByteBuf buf = Unpooled.wrappedBuffer(array);
- if (!(buf.readByte() == 16 && validateRemainingLength(buf) && buf.readByte() == (byte) 0)) return false;
+ // Parse "fixed header"
+ if (!(readByte(buf) == 16 && validateRemainingLength(buf) && readByte(buf) == (byte) 0)) {
+ return false;
+ }
+
+ // Start parsing the Protocol Name
+ byte b = readByte(buf); // LSB
+
+ // MQTT 3.1.1 & 5
+ if (b == 4 &&
+ (readByte(buf) != 77 || // M
+ readByte(buf) != 81 || // Q
+ readByte(buf) != 84 || // T
+ readByte(buf) != 84)) { // T
+ return false;
+ }
+
+ // MQTT 3.1
+ if (b == 6 &&
+ (readByte(buf) != 77 || // M
+ readByte(buf) != 81 || // Q
+ readByte(buf) != 73 || // I
+ readByte(buf) != 115 || // s
+ readByte(buf) != 100 || // d
+ readByte(buf) != 112)) { // p
+ return false;
+ }
+
+ // Protocol Version
+ b = readByte(buf);
Review comment:
How does this play with the protocol version checks? if we don't support the version we won't see it as mqtt, so we can never do version the checks in handleConnection. Or am i missing something.
https://github.com/apache/activemq-artemis/pull/3907/files#diff-49bf0ed78bad6e95bceeea1d35d0c2216c0c14a3fdbf23cca0c14adf8736d818L180
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSession.java
##########
@@ -212,4 +234,42 @@ public CoreMessageObjectPools getCoreMessageObjectPools() {
return coreMessageObjectPools;
}
+ public boolean is5() {
+ return five;
+ }
+
+ public void set5(boolean five) {
Review comment:
small point, but I am not sure on five and boolean, it may be better to keep it to int and int > 3, I imagine there will be mqtt 6 etc... and they will build on each other.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on pull request #3907: ARTEMIS-8454 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#issuecomment-1009558389
I apologize for the length of this PR. I realize it is a lot to review. However, I found that I couldn't break it up as changes in one functional area quickly bled over into other areas. I'd love any feedback.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on a change in pull request #3907: ARTEMIS-8454 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on a change in pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#discussion_r782351849
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java
##########
@@ -107,120 +155,339 @@ public static String convertMQTTAddressFilterToCoreRetain(String filter, Wildcar
return MQTT_RETAIN_ADDRESS_PREFIX + MQTT_WILDCARD.convert(filter, wildcardConfiguration);
}
- private static ICoreMessage createServerMessage(MQTTSession session,
- SimpleString address,
- boolean retain,
- int qos) {
+ private static ICoreMessage createServerMessage(MQTTSession session, SimpleString address, MqttPublishMessage mqttPublishMessage) {
long id = session.getServer().getStorageManager().generateID();
- CoreMessage message = new CoreMessage(id, DEFAULT_SERVER_MESSAGE_BUFFER_SIZE, session.getCoreMessageObjectPools());
+ CoreMessage message = new CoreMessage(id, mqttPublishMessage.fixedHeader().remainingLength(), session.getCoreMessageObjectPools());
message.setAddress(address);
- message.putBooleanProperty(MQTT_MESSAGE_RETAIN_KEY, retain);
- message.putIntProperty(MQTT_QOS_LEVEL_KEY, qos);
+ message.putBooleanProperty(MQTT_MESSAGE_RETAIN_KEY, mqttPublishMessage.fixedHeader().isRetain());
Review comment:
I didn't seriously consider implementing a native MQTTMessage at this point. My goal in this PR was to correctly implement all the new MQTT 5 features.
We've always used message properties when converting to/from core for MQTT-specific details. This PR isn't changing that. I'm not clear on how this would make the transition to a native MQTTMessage implementation more difficult later. Can you elaborate on that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#issuecomment-1016664063
@gtully, I'll take a look. No worries!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on a change in pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on a change in pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#discussion_r792960826
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTSession.java
##########
@@ -212,4 +234,42 @@ public CoreMessageObjectPools getCoreMessageObjectPools() {
return coreMessageObjectPools;
}
+ public boolean is5() {
+ return five;
+ }
+
+ public void set5(boolean five) {
Review comment:
That's a fair point. However, I'd prefer to leave it as a `boolean` for now and change it later if/when necessary. It will be just as easy to change it later as it would be to change it now. In my opinion the benefit of not changing it now is that using a `boolean` is more clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] clebertsuconic commented on pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
clebertsuconic commented on pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#issuecomment-1023283334
The only thing I don't like on this PR is the "leakage" of getMqttSessionScanInterval on ConfigurationImpl.
I would prefer it being a bean util property on the MQTT Protocol Manager. Make the scanInterval as part of the default configuration so it's easier for users to see it in there.
Do you really need it on the Configurationimpl / Configuration or would you be able to make it part of MQTTProtocolManager and a beanutil property?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] clebertsuconic merged pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
clebertsuconic merged pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on a change in pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on a change in pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#discussion_r792927791
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTPublishManager.java
##########
@@ -284,11 +358,110 @@ private void sendServerMessage(int messageId, ICoreMessage message, int delivery
payload.writeBytes(bodyBuffer.byteBuf());
break;
}
- session.getProtocolHandler().send(messageId, address, qos, isRetain, payload, deliveryCount);
+
+ // [MQTT-3.3.1-2] The DUP flag MUST be set to 0 for all QoS 0 messages.
+ boolean redelivery = qos == 0 ? false : (deliveryCount > 1);
+
+ boolean isRetain = message.getBooleanProperty(MQTT_MESSAGE_RETAIN_KEY);
+ MqttProperties mqttProperties = getPublishProperties(message);
+
+ if (session.is5()) {
+ if (session.getState().getSubscription(message.getAddress()) != null && !session.getState().getSubscription(message.getAddress()).option().isRetainAsPublished()) {
+ isRetain = false;
+ }
+
+ // [MQTT-3.8.3-3] remove property used for no-local implementation
+ message.removeProperty(MessageUtil.CONNECTION_ID_PROPERTY_NAME);
+
+ if (session.getState().getClientTopicAliasMaximum() != null) {
+ Integer alias = session.getState().getServerTopicAlias(address);
+ if (alias == null) {
+ alias = session.getState().addServerTopicAlias(address);
+ if (alias != null) {
+ mqttProperties.add(new MqttProperties.IntegerProperty(TOPIC_ALIAS.value(), alias));
+ }
+ } else {
+ mqttProperties.add(new MqttProperties.IntegerProperty(TOPIC_ALIAS.value(), alias));
+ address = "";
+ }
+ }
+ }
+
+ int remainingLength = MQTTUtil.calculateRemainingLength(address, mqttProperties, payload);
+ MqttFixedHeader header = new MqttFixedHeader(MqttMessageType.PUBLISH, redelivery, MqttQoS.valueOf(qos), isRetain, remainingLength);
+ MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(address, messageId, mqttProperties);
+ MqttPublishMessage publish = new MqttPublishMessage(header, varHeader, payload);
+
+ if (session.is5()) {
+ int size = MQTTUtil.calculateMessageSize(publish);
Review comment:
Good idea. I'll push that change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on a change in pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on a change in pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#discussion_r792925945
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTProtocolManager.java
##########
@@ -185,45 +243,83 @@ public void handleBuffer(RemotingConnection connection, ActiveMQBuffer buffer) {
@Override
public void addChannelHandlers(ChannelPipeline pipeline) {
pipeline.addLast(MqttEncoder.INSTANCE);
- pipeline.addLast(new MqttDecoder(MQTTUtil.MAX_MESSAGE_SIZE));
+ /*
+ * If we use the value from getMaximumPacketSize() here anytime a client sends a packet that's too large it
+ * will receive a DISCONNECT with a reason code of 0x81 instead of 0x95 like it should according to the spec.
Review comment:
My goal with this commit was to be as close to the spec as possible. I wanted to "get it right" first and then later "make it fast" if necessary. In any event, computing the size of the message is pretty cheap since it's just a small bit of simple arithmetic. See `org.apache.activemq.artemis.core.protocol.mqtt.MQTTUtil#calculateMessageSize`.
I'm certainly open to sending a PR to Netty for this, but that it outside the scope of this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] jbertram commented on pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
jbertram commented on pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#issuecomment-1024925406
There are a couple of failures on the test-suite that I need to resolve. I'll move this to draft for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gtully commented on pull request #3907: ARTEMIS-3638 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
gtully commented on pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#issuecomment-1016656322
@jbertram looking to resolve the merge conflict via the gh ui, seems not to have the desired effect. sorry! Maybe you do a rebase and resolve against main and force push from your local branch and whack my merge from main. I won't attempt that again any time soon :-)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gtully commented on pull request #3907: ARTEMIS-8454 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
gtully commented on pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#issuecomment-1009971195
looks good, lots of new tests that validate. great work.
I was peeking for some SASL bits, I guess that can layer nicely on this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] clebertsuconic commented on a change in pull request #3907: ARTEMIS-8454 Support MQTT 5
Posted by GitBox <gi...@apache.org>.
clebertsuconic commented on a change in pull request #3907:
URL: https://github.com/apache/activemq-artemis/pull/3907#discussion_r782319905
##########
File path: artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTUtil.java
##########
@@ -107,120 +155,339 @@ public static String convertMQTTAddressFilterToCoreRetain(String filter, Wildcar
return MQTT_RETAIN_ADDRESS_PREFIX + MQTT_WILDCARD.convert(filter, wildcardConfiguration);
}
- private static ICoreMessage createServerMessage(MQTTSession session,
- SimpleString address,
- boolean retain,
- int qos) {
+ private static ICoreMessage createServerMessage(MQTTSession session, SimpleString address, MqttPublishMessage mqttPublishMessage) {
long id = session.getServer().getStorageManager().generateID();
- CoreMessage message = new CoreMessage(id, DEFAULT_SERVER_MESSAGE_BUFFER_SIZE, session.getCoreMessageObjectPools());
+ CoreMessage message = new CoreMessage(id, mqttPublishMessage.fixedHeader().remainingLength(), session.getCoreMessageObjectPools());
message.setAddress(address);
- message.putBooleanProperty(MQTT_MESSAGE_RETAIN_KEY, retain);
- message.putIntProperty(MQTT_QOS_LEVEL_KEY, qos);
+ message.putBooleanProperty(MQTT_MESSAGE_RETAIN_KEY, mqttPublishMessage.fixedHeader().isRetain());
Review comment:
Did you consider the possibility to create a MQTTMessage?
If you're not doing that now, and we plan to eventually do it.. using application properties from CORE wouldn't make it a bit more difficult to make such transition?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org