opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-01 15:47:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

add mqtt support and fix NPE of shutdown method in SnodeController

Browse Source
This commit is contained in:
chengxiangwang
2019-01-29 10:34:18 +08:00
parent 5d60a63501
commit 359e443247
9 changed files with 68 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/src/main/java/org/apache/rocketmq/common/protocol/header/mqtt/MqttHeader.java
+6 -2
View File
@@ -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 {
}
}
*/
remoting/src/main/java/org/apache/rocketmq/remoting/RemotingServerFactory.java
+1 -1
View File
@@ -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() {
remoting/src/main/java/org/apache/rocketmq/remoting/ServerConfig.java
+9
View File
@@ -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;
}
remoting/src/main/java/org/apache/rocketmq/remoting/transport/mqtt/MqttRemotingServer.java
+1 -1
View File
@@ -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",
remoting/src/main/resources/META-INF/service/org.apache.rocketmq.remoting.RemotingClient
+1 -1
View File
@@ -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
mqtt=org.apache.rocketmq.remoting.transport.mqtt.MqttRemotingClient
remoting/src/main/resources/META-INF/service/org.apache.rocketmq.remoting.RemotingServer
+1 -1
View File
@@ -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
snode/src/main/java/org/apache/rocketmq/snode/SnodeController.java
+37 -12
View File
@@ -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() {
snode/src/main/java/org/apache/rocketmq/snode/SnodeStartup.java
+3 -3
View File
@@ -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))));
snode/src/main/java/org/apache/rocketmq/snode/processor/DefaultMqttMessageProcessor.java
+9 -1
View File
@@ -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;
}