Refactor MoP to prepare for split Proxy to seperate module (#1509)

Author: Technoboy-

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

@@ -11,24 +11,26 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.mqtt;
+package io.streamnative.pulsar.handlers.mqtt.broker;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static io.streamnative.pulsar.handlers.mqtt.utils.ConfigurationUtils.LISTENER_DEL;
-import static io.streamnative.pulsar.handlers.mqtt.utils.ConfigurationUtils.PLAINTEXT_PREFIX;
-import static io.streamnative.pulsar.handlers.mqtt.utils.ConfigurationUtils.PROTOCOL_NAME;
-import static io.streamnative.pulsar.handlers.mqtt.utils.ConfigurationUtils.SSL_PREFIX;
-import static io.streamnative.pulsar.handlers.mqtt.utils.ConfigurationUtils.SSL_PSK_PREFIX;
-import static io.streamnative.pulsar.handlers.mqtt.utils.ConfigurationUtils.WS_PLAINTEXT_PREFIX;
-import static io.streamnative.pulsar.handlers.mqtt.utils.ConfigurationUtils.WS_SSL_PREFIX;
-import static io.streamnative.pulsar.handlers.mqtt.utils.ConfigurationUtils.getListenerPort;
+import static io.streamnative.pulsar.handlers.mqtt.common.utils.ConfigurationUtils.LISTENER_DEL;
+import static io.streamnative.pulsar.handlers.mqtt.common.utils.ConfigurationUtils.PLAINTEXT_PREFIX;
+import static io.streamnative.pulsar.handlers.mqtt.common.utils.ConfigurationUtils.PROTOCOL_NAME;
+import static io.streamnative.pulsar.handlers.mqtt.common.utils.ConfigurationUtils.SSL_PREFIX;
+import static io.streamnative.pulsar.handlers.mqtt.common.utils.ConfigurationUtils.SSL_PSK_PREFIX;
+import static io.streamnative.pulsar.handlers.mqtt.common.utils.ConfigurationUtils.WS_PLAINTEXT_PREFIX;
+import static io.streamnative.pulsar.handlers.mqtt.common.utils.ConfigurationUtils.WS_SSL_PREFIX;
+import static io.streamnative.pulsar.handlers.mqtt.common.utils.ConfigurationUtils.getListenerPort;
 import com.google.common.collect.ImmutableMap;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.util.concurrent.DefaultThreadFactory;
+import io.streamnative.pulsar.handlers.mqtt.MopVersion;
+import io.streamnative.pulsar.handlers.mqtt.broker.channel.MQTTChannelInitializer;
+import io.streamnative.pulsar.handlers.mqtt.common.utils.ConfigurationUtils;
 import io.streamnative.pulsar.handlers.mqtt.proxy.MQTTProxyConfiguration;
 import io.streamnative.pulsar.handlers.mqtt.proxy.MQTTProxyService;
-import io.streamnative.pulsar.handlers.mqtt.utils.ConfigurationUtils;
 import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.concurrent.Executors;
@@ -97,7 +99,7 @@ public void start(BrokerService brokerService) {
             try {
                 MQTTProxyConfiguration proxyConfig =
                         ConfigurationUtils.create(mqttConfig.getProperties(), MQTTProxyConfiguration.class);
-                proxyService = new MQTTProxyService(mqttService, proxyConfig);
+                proxyService = new MQTTProxyService(brokerService, proxyConfig);
                 proxyService.start();
                 log.info("Start MQTT proxy service at port: {}", proxyConfig.getMqttProxyPort());
             } catch (Exception ex) {

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

@@ -11,8 +11,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.mqtt;
+package io.streamnative.pulsar.handlers.mqtt.broker;
 
+import io.streamnative.pulsar.handlers.mqtt.common.MQTTCommonConfiguration;
 import lombok.Getter;
 import lombok.Setter;
 import org.apache.pulsar.common.configuration.FieldContext;

…ve/pulsar/handlers/mqtt/MQTTService.java …ar/handlers/mqtt/broker/MQTTService.javamqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/MQTTService.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/MQTTService.java mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/MQTTService.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/MQTTService.java

@@ -11,20 +11,25 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.mqtt;
-
-import io.streamnative.pulsar.handlers.mqtt.support.MQTTMetricsCollector;
-import io.streamnative.pulsar.handlers.mqtt.support.MQTTMetricsProvider;
-import io.streamnative.pulsar.handlers.mqtt.support.QosPublishHandlersImpl;
-import io.streamnative.pulsar.handlers.mqtt.support.RetainedMessageHandler;
-import io.streamnative.pulsar.handlers.mqtt.support.WillMessageHandler;
-import io.streamnative.pulsar.handlers.mqtt.support.event.DisableEventCenter;
-import io.streamnative.pulsar.handlers.mqtt.support.event.PulsarEventCenter;
-import io.streamnative.pulsar.handlers.mqtt.support.event.PulsarEventCenterImpl;
-import io.streamnative.pulsar.handlers.mqtt.support.psk.PSKConfiguration;
-import io.streamnative.pulsar.handlers.mqtt.support.systemtopic.SystemEventService;
+package io.streamnative.pulsar.handlers.mqtt.broker;
+
+import io.streamnative.pulsar.handlers.mqtt.broker.impl.MQTTNamespaceBundleOwnershipListener;
+import io.streamnative.pulsar.handlers.mqtt.broker.impl.MQTTSubscriptionManager;
+import io.streamnative.pulsar.handlers.mqtt.broker.metric.MQTTMetricsCollector;
+import io.streamnative.pulsar.handlers.mqtt.broker.metric.MQTTMetricsProvider;
+import io.streamnative.pulsar.handlers.mqtt.broker.mqtt5.WillMessageHandler;
+import io.streamnative.pulsar.handlers.mqtt.broker.qos.QosPublishHandlers;
+import io.streamnative.pulsar.handlers.mqtt.broker.qos.QosPublishHandlersImpl;
+import io.streamnative.pulsar.handlers.mqtt.common.MQTTConnectionManager;
+import io.streamnative.pulsar.handlers.mqtt.common.authentication.MQTTAuthenticationService;
+import io.streamnative.pulsar.handlers.mqtt.common.event.DisableEventCenter;
+import io.streamnative.pulsar.handlers.mqtt.common.event.PulsarEventCenter;
+import io.streamnative.pulsar.handlers.mqtt.common.event.PulsarEventCenterImpl;
+import io.streamnative.pulsar.handlers.mqtt.common.mqtt5.RetainedMessageHandler;
+import io.streamnative.pulsar.handlers.mqtt.common.psk.PSKConfiguration;
+import io.streamnative.pulsar.handlers.mqtt.common.systemtopic.DisabledSystemEventService;
+import io.streamnative.pulsar.handlers.mqtt.common.systemtopic.SystemEventService;
 import lombok.Getter;
-import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.authorization.AuthorizationService;
@@ -82,8 +87,7 @@ public class MQTTService {
     private final QosPublishHandlers qosPublishHandlers;
 
     @Getter
-    @Setter
-    private SystemEventService eventService;
+    private final SystemEventService eventService;
 
     public MQTTService(BrokerService brokerService, MQTTServerConfiguration serverConfiguration) {
         this.brokerService = brokerService;
@@ -107,21 +111,15 @@ public MQTTService(BrokerService brokerService, MQTTServerConfiguration serverCo
             this.eventCenter = new PulsarEventCenterImpl(brokerService,
                     serverConfiguration.getEventCenterCallbackPoolThreadNum());
         }
-        this.retainedMessageHandler = new RetainedMessageHandler(this);
+        this.eventService = new DisabledSystemEventService();
+        this.retainedMessageHandler = new RetainedMessageHandler(eventService);
         this.qosPublishHandlers = new QosPublishHandlersImpl(this);
         this.willMessageHandler = new WillMessageHandler(this);
     }
 
-    public boolean isSystemTopicEnabled() {
-        return eventService != null;
-    }
-
     public void close() {
         this.connectionManager.close();
         this.eventCenter.shutdown();
-        if (eventService != null) {
-            eventService.close();
-        }
         this.willMessageHandler.close();
     }
 }

…andlers/mqtt/MQTTChannelInitializer.java …oker/channel/MQTTChannelInitializer.javamqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/MQTTChannelInitializer.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTChannelInitializer.java mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/MQTTChannelInitializer.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTChannelInitializer.java

@@ -11,7 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.mqtt;
+package io.streamnative.pulsar.handlers.mqtt.broker.channel;
 
 import static org.apache.pulsar.client.impl.PulsarChannelInitializer.TLS_HANDLER;
 import io.netty.channel.ChannelInitializer;
@@ -24,11 +24,14 @@
 import io.netty.handler.codec.mqtt.MqttDecoder;
 import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.timeout.IdleStateHandler;
-import io.streamnative.pulsar.handlers.mqtt.adapter.CombineAdapterHandler;
-import io.streamnative.pulsar.handlers.mqtt.adapter.MqttAdapterDecoder;
-import io.streamnative.pulsar.handlers.mqtt.adapter.MqttAdapterEncoder;
-import io.streamnative.pulsar.handlers.mqtt.codec.MqttWebSocketCodec;
-import io.streamnative.pulsar.handlers.mqtt.support.psk.PSKUtils;
+import io.streamnative.pulsar.handlers.mqtt.broker.MQTTServerConfiguration;
+import io.streamnative.pulsar.handlers.mqtt.broker.MQTTService;
+import io.streamnative.pulsar.handlers.mqtt.broker.codec.MqttWebSocketCodec;
+import io.streamnative.pulsar.handlers.mqtt.common.Constants;
+import io.streamnative.pulsar.handlers.mqtt.common.adapter.CombineAdapterHandler;
+import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterDecoder;
+import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterEncoder;
+import io.streamnative.pulsar.handlers.mqtt.common.psk.PSKUtils;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;

…ar/handlers/mqtt/MQTTInboundHandler.java …t/broker/channel/MQTTInboundHandler.javamqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/MQTTInboundHandler.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTInboundHandler.java mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/MQTTInboundHandler.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTInboundHandler.java

@@ -11,10 +11,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.mqtt;
+package io.streamnative.pulsar.handlers.mqtt.broker.channel;
 
 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.common.MQTTCommonInboundHandler;
 import lombok.extern.slf4j.Slf4j;
 /**
  * MQTT in bound handler.

…handlers/mqtt/support/MQTTServerCnx.java …s/mqtt/broker/channel/MQTTServerCnx.javamqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/support/MQTTServerCnx.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTServerCnx.java mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/support/MQTTServerCnx.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTServerCnx.java

@@ -11,9 +11,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.mqtt.support;
+package io.streamnative.pulsar.handlers.mqtt.broker.channel;
 
 import io.netty.channel.ChannelHandlerContext;
+import io.streamnative.pulsar.handlers.mqtt.broker.impl.consumer.MQTTConsumer;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import lombok.extern.slf4j.Slf4j;

mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/package-info.java

@@ -0,0 +1,18 @@
+/**
+ * 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 info.
+ */
+package io.streamnative.pulsar.handlers.mqtt.broker.channel;

…dlers/mqtt/codec/MqttWebSocketCodec.java …qtt/broker/codec/MqttWebSocketCodec.javamqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/codec/MqttWebSocketCodec.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/codec/MqttWebSocketCodec.java mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/codec/MqttWebSocketCodec.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/codec/MqttWebSocketCodec.java

@@ -11,7 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.mqtt.codec;
+package io.streamnative.pulsar.handlers.mqtt.broker.codec;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;

…/handlers/mqtt/adapter/package-info.java …lers/mqtt/broker/codec/package-info.javamqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/adapter/package-info.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/codec/package-info.java mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/adapter/package-info.java renamed to mqtt-impl/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/codec/package-info.java

@@ -15,4 +15,4 @@
 /**
  * Package info.
  */
-package io.streamnative.pulsar.handlers.mqtt.adapter;
+package io.streamnative.pulsar.handlers.mqtt.broker.codec;

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

@@ -11,7 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.mqtt;
+package io.streamnative.pulsar.handlers.mqtt.broker.impl;
 
 import java.util.ArrayList;
 import java.util.List;