Skip to content

Support MQTT-5 Conn-ACK with subscriptionIdentifiersAvailable and maximumPacketSize #685

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

Merged
Technoboy- merged 2 commits into from
Jun 25, 2022
Merged

Support MQTT-5 Conn-ACK with subscriptionIdentifiersAvailable and maximumPacketSize #685

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3c9df91
Support MQTT-5 CONNACK with `subscriptionIdentifiersAvailable` and `m…
Technoboy- Jun 24, 2022
ad31756
fix proxy not set value.
Technoboy- Jun 24, 2022
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
3 changes: 2 additions & 1 deletion mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/Connection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,8 @@ public void sendConnAck() {
MqttConnectAck.MqttConnectSuccessAckBuilder builder = MqttConnectAck.successBuilder(protocolVersion)
.receiveMaximum(getServerRestrictions().getReceiveMaximum())
.cleanSession(clientRestrictions.isCleanSession())
.maximumQos(MqttQoS.AT_LEAST_ONCE.value());
.maximumQos(MqttQoS.AT_LEAST_ONCE.value())
.maximumPacketSize(getServerRestrictions().getMaximumPacketSize());
MqttProperties.StringProperty resInformation = (MqttProperties.StringProperty) connectMessage.variableHeader()
.properties().getProperty(MqttProperties.MqttPropertyType.RESPONSE_INFORMATION.value());
if (resInformation != null) {
Expand Down
16 changes: 16 additions & 0 deletions ...-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/messages/ack/MqttConnectAck.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,8 @@ public final static class MqttConnectSuccessAckBuilder {

private String responseInformation;

private int maximumPacketSize;

public MqttConnectSuccessAckBuilder(int protocolVersion) {
this.protocolVersion = protocolVersion;
}
Expand All @@ -73,6 +75,11 @@ public MqttConnectSuccessAckBuilder responseInformation(String responseInformati
return this;
}

public MqttConnectSuccessAckBuilder maximumPacketSize(int maximumPacketSize) {
this.maximumPacketSize = maximumPacketSize;
return this;
}

public MqttAck build() {
MqttMessageBuilders.ConnAckBuilder commonBuilder = MqttMessageBuilders.connAck()
.sessionPresent(!cleanSession);
Expand All @@ -88,8 +95,17 @@ public MqttAck build() {
MqttProperties.IntegerProperty maximumQosProperty =
new MqttProperties.IntegerProperty(MqttProperties.MqttPropertyType.MAXIMUM_QOS.value(),
maximumQos);
// Set Subscription Identifiers Available = 0 now.
MqttProperties.IntegerProperty subscriptionIdentifiersAvailableProperty =
new MqttProperties.IntegerProperty(
MqttProperties.MqttPropertyType.SUBSCRIPTION_IDENTIFIER_AVAILABLE.value(), 0);
MqttProperties.IntegerProperty maximumPacketSizeProperty =
new MqttProperties.IntegerProperty(
MqttProperties.MqttPropertyType.MAXIMUM_PACKET_SIZE.value(), maximumPacketSize);
properties.add(receiveMaximumProperty);
properties.add(maximumQosProperty);
properties.add(subscriptionIdentifiersAvailableProperty);
properties.add(maximumPacketSizeProperty);
if (StringUtils.isNotEmpty(responseInformation)) {
MqttProperties.StringProperty responseInformationProperty =
new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.RESPONSE_INFORMATION.value(),
Expand Down
1 change: 1 addition & 0 deletions ...ain/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyProtocolMethodProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,7 @@ public void doProcessConnect(MqttAdapterMessage adapter, String userRole, Client
final MqttConnectMessage msg = (MqttConnectMessage) adapter.getMqttMessage();
final ServerRestrictions serverRestrictions = ServerRestrictions.builder()
.receiveMaximum(proxyConfig.getReceiveMaximum())
.maximumPacketSize(proxyConfig.getMqttMessageMaxLength())
.build();
connection = Connection.builder()
.protocolVersion(msg.variableHeader().version())
Expand Down
3 changes: 3 additions & 0 deletions ...l/src/main/java/io/streamnative/pulsar/handlers/mqtt/restrictions/ServerRestrictions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,4 +24,7 @@
public class ServerRestrictions {
@Getter
private int receiveMaximum;

@Getter
private int maximumPacketSize;
}
1 change: 1 addition & 0 deletions .../java/io/streamnative/pulsar/handlers/mqtt/support/MQTTBrokerProtocolMethodProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,7 @@ public void doProcessConnect(MqttAdapterMessage adapterMsg, String userRole,
final MqttConnectMessage msg = (MqttConnectMessage) adapterMsg.getMqttMessage();
ServerRestrictions serverRestrictions = ServerRestrictions.builder()
.receiveMaximum(configuration.getReceiveMaximum())
.maximumPacketSize(configuration.getMqttMessageMaxLength())
.build();
connection = Connection.builder()
.protocolVersion(msg.variableHeader().version())
Expand Down