Seperate proxy and broker a single module (#1510)

Author: Technoboy-

.ci/Dockerfile

@@ -6,4 +6,4 @@ ARG MOP_VERSION
 USER root
 RUN rm -rf /pulsar/protocols/pulsar-protocol-handler-mqtt-*.nar
 
-COPY ../mqtt-impl/target/pulsar-protocol-handler-mqtt-${MOP_VERSION}.nar /pulsar/protocols
+COPY ../mqtt-broker/target/pulsar-protocol-handler-mqtt-${MOP_VERSION}.nar /pulsar/protocols

.github/workflows/pr_test.yml

@@ -61,7 +61,7 @@ jobs:
         id: list-test
         run: |
           TESTS=`find tests/src/test/java/io/streamnative/pulsar/handlers/mqtt \
-          mqtt-impl/src/test/java/io/streamnative/pulsar/handlers/mqtt -name "*Test.java" | \
+          mqtt-broker/src/test/java/io/streamnative/pulsar/handlers/mqtt -name "*Test.java" | \
           awk -F "/" '{ print $NF }' | \
           awk -F "." '{ print $1 }' | \
           jq -R -s -c 'split("\n") | map(select(. != ""))'`
@@ -146,7 +146,7 @@ jobs:
       - name: Download jacoco artifact
         uses: actions/download-artifact@v3
         with:
-          path: mqtt-impl/target
+          path: mqtt-broker/target
 
       - name: Merge jacoco report
         run: mvn jacoco:merge

README.md

@@ -27,6 +27,13 @@
     <name>StreamNative :: Pulsar Protocol Handler :: MQTT</name>
     <description>MQTT on Pulsar implemented using Pulsar Protocol Handler</description>
 
+    <dependencies>
+        <dependency>
+            <groupId>io.streamnative.pulsar.handlers</groupId>
+            <artifactId>pulsar-protocol-handler-mqtt-proxy</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
     <build>
         <plugins>
             <plugin>

mqtt-impl/pom.xml mqtt-broker/pom.xmlmqtt-impl/pom.xml renamed to mqtt-broker/pom.xml

@@ -93,16 +93,17 @@ public MQTTService(BrokerService brokerService, MQTTServerConfiguration serverCo
         this.brokerService = brokerService;
         this.pulsarService = brokerService.pulsar();
         this.serverConfiguration = serverConfiguration;
-        this.pskConfiguration = new PSKConfiguration(serverConfiguration);
+        this.pskConfiguration = new PSKConfiguration(serverConfiguration.getMqttTlsPskIdentityHint(),
+                serverConfiguration.getMqttTlsPskIdentity(), serverConfiguration.getMqttTlsPskIdentityFile(),
+                serverConfiguration.getMqttTlsProtocols(), serverConfiguration.getMqttTlsCiphers());
         this.authorizationService = brokerService.getAuthorizationService();
         this.bundleOwnershipListener = new MQTTNamespaceBundleOwnershipListener(pulsarService.getNamespaceService());
         this.metricsCollector = new MQTTMetricsCollector(serverConfiguration);
         this.metricsProvider = new MQTTMetricsProvider(metricsCollector);
         this.pulsarService.addPrometheusRawMetricsProvider(metricsProvider);
         this.authenticationService = serverConfiguration.isMqttAuthenticationEnabled()
             ? new MQTTAuthenticationService(brokerService,
-                serverConfiguration.getMqttAuthenticationMethods(),
-                serverConfiguration.isMqttProxyMTlsAuthenticationEnabled()) : null;
+                serverConfiguration.getMqttAuthenticationMethods()) : null;
         this.connectionManager = new MQTTConnectionManager(pulsarService.getAdvertisedAddress());
         this.subscriptionManager = new MQTTSubscriptionManager();
         if (getServerConfiguration().isMqttProxyEnabled()) {

mqtt-impl/src/assemble/bin.xml mqtt-broker/src/assemble/bin.xmlmqtt-impl/src/assemble/bin.xml renamed to mqtt-broker/src/assemble/bin.xml

@@ -0,0 +1,60 @@
+/**
+ * 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.streamnative.pulsar.handlers.mqtt.broker.channel;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.channel.ChannelHandlerContext;
+import io.streamnative.pulsar.handlers.mqtt.broker.MQTTService;
+import io.streamnative.pulsar.handlers.mqtt.broker.processor.MQTTBrokerProtocolMethodProcessor;
+import io.streamnative.pulsar.handlers.mqtt.common.MQTTCommonInboundHandler;
+import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterMessage;
+import java.util.concurrent.CompletableFuture;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * MQTT in bound handler.
+ */
+@Sharable
+@Slf4j
+public class MQTTBrokerInboundHandler extends MQTTCommonInboundHandler {
+
+    public static final String NAME = "handler";
+
+    private final MQTTService mqttService;
+
+    public MQTTBrokerInboundHandler(MQTTService mqttService) {
+        this.mqttService = mqttService;
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        super.channelActive(ctx);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object message) {
+        checkArgument(message instanceof MqttAdapterMessage);
+        MqttAdapterMessage adapterMsg = (MqttAdapterMessage) message;
+        processors.computeIfAbsent(adapterMsg.getClientId(), key -> {
+            MQTTBrokerProtocolMethodProcessor p = new MQTTBrokerProtocolMethodProcessor(mqttService, ctx);
+            CompletableFuture<Void> inactiveFuture = p.getInactiveFuture();
+            inactiveFuture.whenComplete((id, ex) -> {
+                processors.remove(adapterMsg.getClientId());
+            });
+            return p;
+        });
+        super.channelRead(ctx, message);
+    }
+}

…ers/mqtt/broker/MQTTProtocolHandler.java …ers/mqtt/broker/MQTTProtocolHandler.javamqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/MQTTProtocolHandler.java renamed to mqtt-broker/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/MQTTProtocolHandler.java mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/MQTTProtocolHandler.java renamed to mqtt-broker/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/MQTTProtocolHandler.java

@@ -98,7 +98,7 @@ public void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addLast(MqttAdapterEncoder.NAME, MqttAdapterEncoder.INSTANCE);
         // Handler
         ch.pipeline().addLast(CombineAdapterHandler.NAME, new CombineAdapterHandler());
-        ch.pipeline().addLast(MQTTInboundHandler.NAME, new MQTTInboundHandler(mqttService));
+        ch.pipeline().addLast(MQTTBrokerInboundHandler.NAME, new MQTTBrokerInboundHandler(mqttService));
     }
 
     /**