update

streamnative · Sep 12, 2024 · 815bd7f · 815bd7f
1 parent 46bcbca
commit 815bd7f
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 6 deletions.
diff --git a/...ain/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyProtocolMethodProcessor.java b/...ain/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyProtocolMethodProcessor.java
@@ -13,6 +13,7 @@
  */
 package io.streamnative.pulsar.handlers.mqtt.proxy;
 
+import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.createMqtt5ConnectMessage;
 import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.createMqttPublishMessage;
 import static io.streamnative.pulsar.handlers.mqtt.utils.MqttMessageUtils.createMqttSubscribeMessage;
 import com.google.common.collect.Lists;
@@ -140,6 +141,12 @@ public void doProcessConnect(MqttAdapterMessage adapter, String userRole,
                 .build();
         connection.sendConnAck();
 
+        if (proxyConfig.isMqttAuthorizationEnabled()) {
+            MqttConnectMessage connectMessage = createMqtt5ConnectMessage(msg);
+            msg = connectMessage;
+            connection.setConnectMessage(msg);
+        }
+
         ConnectEvent connectEvent = ConnectEvent.builder()
                 .clientId(connection.getClientId())
                 .address(pulsarService.getAdvertisedAddress())

diff --git a/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/utils/MqttMessageUtils.java b/mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/utils/MqttMessageUtils.java
@@ -189,16 +189,12 @@ public static MqttPublishMessage createMqttWillMessage(WillMessage willMessage)
         return builder.build();
     }
 
-    public static MqttConnectMessage createMqttConnectMessage(MqttConnectMessage connectMessage,
-                                                              String authData) {
+    public static MqttConnectMessage createMqtt5ConnectMessage(MqttConnectMessage connectMessage) {
         final MqttConnectVariableHeader header = connectMessage.variableHeader();
-        MqttProperties properties = new MqttProperties();
-        properties.add(new MqttProperties.UserProperty(AUTHENTICATE_ROLE_KEY, authData));
         MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader(
                 MqttVersion.MQTT_5.protocolName(), MqttVersion.MQTT_5.protocolLevel(), header.hasUserName(),
                 header.hasPassword(), header.isWillRetain(), header.willQos(), header.isWillFlag(),
-                header.isCleanSession(), header.keepAliveTimeSeconds(), properties
-        );
+                header.isCleanSession(), header.keepAliveTimeSeconds(), connectMessage.variableHeader().properties());
         MqttConnectMessage newConnectMessage = new MqttConnectMessage(connectMessage.fixedHeader(), variableHeader,
                 connectMessage.payload());
         return newConnectMessage;