diff --git a/common/src/main/java/org/apache/rocketmq/common/protocol/header/mqtt/MqttHeader.java b/common/src/main/java/org/apache/rocketmq/common/protocol/header/mqtt/MqttHeader.java index cd8a922c34..319153980b 100644 --- a/common/src/main/java/org/apache/rocketmq/common/protocol/header/mqtt/MqttHeader.java +++ b/common/src/main/java/org/apache/rocketmq/common/protocol/header/mqtt/MqttHeader.java @@ -13,11 +13,14 @@ * 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. - */ + *//* + +*/ /** * $Id: EndTransactionResponseHeader.java 1835 2013-05-16 02:00:50Z vintagewang@apache.org $ - */ + *//* + package org.apache.rocketmq.common.protocol.header.mqtt; import io.netty.handler.codec.mqtt.MqttConnectReturnCode; @@ -215,3 +218,4 @@ public class MqttHeader implements CommandCustomHeader { } } +*/ diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/RemotingServerFactory.java b/remoting/src/main/java/org/apache/rocketmq/remoting/RemotingServerFactory.java index 7c75e8a58f..cb2c6734a6 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/RemotingServerFactory.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/RemotingServerFactory.java @@ -39,7 +39,7 @@ public class RemotingServerFactory { } public RemotingServer createRemotingServer(String protocol) { - return ServiceProvider.createInstance(protocolPathMap.get(protocol), RemotingClient.class); + return ServiceProvider.createInstance(protocolPathMap.get(protocol), RemotingServer.class); } public RemotingServer createRemotingServer() { diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/ServerConfig.java b/remoting/src/main/java/org/apache/rocketmq/remoting/ServerConfig.java index dda2528fab..97c3d3eb39 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/ServerConfig.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/ServerConfig.java @@ -20,6 +20,7 @@ import org.apache.rocketmq.remoting.netty.NettySystemConfig; public class ServerConfig implements Cloneable { private int listenPort = 8888; + private int mqttListenPort = 1883; private int serverWorkerThreads = 8; private int serverCallbackExecutorThreads = 8; private int serverSelectorThreads = 3; @@ -75,6 +76,14 @@ public class ServerConfig implements Cloneable { this.listenPort = listenPort; } + public int getMqttListenPort() { + return mqttListenPort; + } + + public void setMqttListenPort(int mqttListenPort) { + this.mqttListenPort = mqttListenPort; + } + public int getServerWorkerThreads() { return serverWorkerThreads; } diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/mqtt/MqttRemotingServer.java b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/mqtt/MqttRemotingServer.java index b7e65b36d7..f44c4e4ed8 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/mqtt/MqttRemotingServer.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/mqtt/MqttRemotingServer.java @@ -134,7 +134,7 @@ public class MqttRemotingServer extends NettyRemotingServerAbstract implements R serverConfig.getServerSelectorThreads())); this.socketChannelClass = NioServerSocketChannel.class; } - this.port = nettyServerConfig.getListenPort(); + this.port = nettyServerConfig.getMqttListenPort(); this.defaultEventExecutorGroup = new DefaultEventExecutorGroup( serverConfig.getServerWorkerThreads(), ThreadUtils.newGenericThreadFactory("NettyWorkerThreads", diff --git a/remoting/src/main/resources/META-INF/service/org.apache.rocketmq.remoting.RemotingClient b/remoting/src/main/resources/META-INF/service/org.apache.rocketmq.remoting.RemotingClient index 62a07e0894..2229e6bba9 100644 --- a/remoting/src/main/resources/META-INF/service/org.apache.rocketmq.remoting.RemotingClient +++ b/remoting/src/main/resources/META-INF/service/org.apache.rocketmq.remoting.RemotingClient @@ -1,3 +1,3 @@ rocketmq=org.apache.rocketmq.remoting.transport.rocketmq.NettyRemotingClient http2=org.apache.rocketmq.remoting.transport.http2.Http2ClientImpl -mqtt=org.apache.rocketmq.remoting.transport.mqtt.MqttClientImpl \ No newline at end of file +mqtt=org.apache.rocketmq.remoting.transport.mqtt.MqttRemotingClient \ No newline at end of file diff --git a/remoting/src/main/resources/META-INF/service/org.apache.rocketmq.remoting.RemotingServer b/remoting/src/main/resources/META-INF/service/org.apache.rocketmq.remoting.RemotingServer index 1526b66bd7..636e6d0097 100644 --- a/remoting/src/main/resources/META-INF/service/org.apache.rocketmq.remoting.RemotingServer +++ b/remoting/src/main/resources/META-INF/service/org.apache.rocketmq.remoting.RemotingServer @@ -1,3 +1,3 @@ rocketmq=org.apache.rocketmq.remoting.transport.rocketmq.NettyRemotingServer http2=org.apache.rocketmq.remoting.transport.http2.Http2ServerImpl -mqtt=org.apache.rocketmq.remoting.transport.mqtt.MqttServerImpl +mqtt=org.apache.rocketmq.remoting.transport.mqtt.MqttRemotingServer diff --git a/snode/src/main/java/org/apache/rocketmq/snode/SnodeController.java b/snode/src/main/java/org/apache/rocketmq/snode/SnodeController.java index 884c6b7691..47feab199d 100644 --- a/snode/src/main/java/org/apache/rocketmq/snode/SnodeController.java +++ b/snode/src/main/java/org/apache/rocketmq/snode/SnodeController.java @@ -121,6 +121,7 @@ public class SnodeController { this.nnodeService = new NnodeServiceImpl(this); this.scheduledService = new ScheduledServiceImpl(this); this.remotingClient = RemotingClientFactory.getInstance().createRemotingClient().init(this.getNettyClientConfig(), null); + this.mqttRemotingClient = RemotingClientFactory.getInstance().createRemotingClient(RemotingUtil.MQTT_PROTOCOL).init(this.getNettyClientConfig(), null); this.sendMessageExecutor = ThreadUtils.newThreadPoolExecutor( snodeConfig.getSnodeSendMessageMinPoolSize(), @@ -312,18 +313,42 @@ public class SnodeController { } public void shutdown() { - this.sendMessageExecutor.shutdown(); - this.pullMessageExecutor.shutdown(); - this.handleMqttMessageExecutor.shutdown(); - this.heartbeatExecutor.shutdown(); - this.consumerManagerExecutor.shutdown(); - this.scheduledExecutorService.shutdown(); - this.remotingClient.shutdown(); - this.mqttRemotingClient.shutdown(); - this.mqttRemotingServer.shutdown(); - this.scheduledService.shutdown(); - this.clientHousekeepingService.shutdown(); - this.pushService.shutdown(); + if (this.sendMessageExecutor != null) { + this.sendMessageExecutor.shutdown(); + } + if (this.pullMessageExecutor != null) { + this.pullMessageExecutor.shutdown(); + } + if (this.handleMqttMessageExecutor != null) { + this.handleMqttMessageExecutor.shutdown(); + } + if (this.heartbeatExecutor != null) { + this.heartbeatExecutor.shutdown(); + } + if (this.consumerManagerExecutor != null) { + this.consumerManagerExecutor.shutdown(); + } + if (this.scheduledExecutorService != null) { + this.scheduledExecutorService.shutdown(); + } + if (this.remotingClient != null) { + this.remotingClient.shutdown(); + } + if (this.mqttRemotingClient != null) { + this.mqttRemotingClient.shutdown(); + } + if(this.mqttRemotingServer != null) { + this.mqttRemotingServer.shutdown(); + } + if(this.scheduledService != null){ + this.scheduledService.shutdown(); + } + if(this.clientHousekeepingService != null) { + this.clientHousekeepingService.shutdown(); + } + if(this.pushService != null) { + this.pushService.shutdown(); + } } public RemotingServer getSnodeServer() { diff --git a/snode/src/main/java/org/apache/rocketmq/snode/SnodeStartup.java b/snode/src/main/java/org/apache/rocketmq/snode/SnodeStartup.java index 0e753e7ded..d6060a90c6 100644 --- a/snode/src/main/java/org/apache/rocketmq/snode/SnodeStartup.java +++ b/snode/src/main/java/org/apache/rocketmq/snode/SnodeStartup.java @@ -16,6 +16,8 @@ */ package org.apache.rocketmq.snode; +import static org.apache.rocketmq.remoting.netty.TlsSystemConfig.TLS_ENABLE; + import ch.qos.logback.classic.LoggerContext; import ch.qos.logback.classic.joran.JoranConfigurator; import ch.qos.logback.core.joran.spi.JoranException; @@ -41,8 +43,6 @@ import org.apache.rocketmq.snode.config.SnodeConfig; import org.apache.rocketmq.srvutil.ServerUtil; import org.slf4j.LoggerFactory; -import static org.apache.rocketmq.remoting.netty.TlsSystemConfig.TLS_ENABLE; - public class SnodeStartup { private static InternalLogger log; public static Properties properties = null; @@ -86,7 +86,7 @@ public class SnodeStartup { final ClientConfig nettyClientConfig = new ClientConfig(); nettyServerConfig.setListenPort(snodeConfig.getListenPort()); - + nettyServerConfig.setListenPort(11911); nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE, String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING)))); diff --git a/snode/src/main/java/org/apache/rocketmq/snode/processor/DefaultMqttMessageProcessor.java b/snode/src/main/java/org/apache/rocketmq/snode/processor/DefaultMqttMessageProcessor.java index 4606fcd717..79b3c46500 100644 --- a/snode/src/main/java/org/apache/rocketmq/snode/processor/DefaultMqttMessageProcessor.java +++ b/snode/src/main/java/org/apache/rocketmq/snode/processor/DefaultMqttMessageProcessor.java @@ -19,7 +19,9 @@ package org.apache.rocketmq.snode.processor; import org.apache.rocketmq.remoting.RemotingChannel; import org.apache.rocketmq.remoting.RequestProcessor; +import org.apache.rocketmq.remoting.exception.RemotingCommandException; import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import org.apache.rocketmq.remoting.transport.mqtt.MqttHeader; public class DefaultMqttMessageProcessor implements RequestProcessor { @@ -27,9 +29,15 @@ public class DefaultMqttMessageProcessor implements RequestProcessor { private static final int MAX_AVAILABLE_VERSION = 4; - @Override public RemotingCommand processRequest(RemotingChannel remotingChannel, RemotingCommand message) { + @Override public RemotingCommand processRequest(RemotingChannel remotingChannel, RemotingCommand message) + throws RemotingCommandException { //解析RemotingCommand，根据MqttMessageType做不同逻辑处理 //TODO + MqttHeader mqttHeader = (MqttHeader)message.decodeCommandCustomHeader(MqttHeader.class); + switch (mqttHeader.getMessageType()) { + case CONNECT: + case DISCONNECT: + } return null; }