Skip to content

R
Rocketmq

Apache RocketMQ / Rocketmq
上一次同步 大约 3 年

通知 268
Star 16139
Fork 68
R
Rocketmq
提交 3629a0a7 编写于 作者: C chengxiangwang
浏览文件
操作

add new module rocketmq-mqtt

上级 a1f4358e
隐藏空白更改
内联 并排
Showing with 580 addition and 285 deletion
common/src/main/java/org/apache/rocketmq/common/MqttConfig.java
浏览文件 @ 3629a0a7
......@@ -50,6 +50,8 @@ public class MqttConfig {
@ImportantField
private boolean aclEnable = false;
private long houseKeepingInterval = 10 * 1000;
public int getListenPort() {
return listenPort;
}
......@@ -130,4 +132,12 @@ public class MqttConfig {
public void setPushMqttMessageThreadPoolQueueCapacity(int pushMqttMessageThreadPoolQueueCapacity) {
this.pushMqttMessageThreadPoolQueueCapacity = pushMqttMessageThreadPoolQueueCapacity;
}
public long getHouseKeepingInterval() {
return houseKeepingInterval;
}
public void setHouseKeepingInterval(long houseKeepingInterval) {
this.houseKeepingInterval = houseKeepingInterval;
}
}
snode/src/main/java/org/apache/rocketmq/snode/client/Client.java → common/src/main/java/org/apache/rocketmq/common/client/Client.java
浏览文件 @ 3629a0a7
......@@ -14,13 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.snode.client;
package org.apache.rocketmq.common.client;
import java.util.Objects;
import java.util.Set;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.apache.rocketmq.snode.client.impl.ClientRole;
public class Client {
......
snode/src/main/java/org/apache/rocketmq/snode/client/ClientManager.java → common/src/main/java/org/apache/rocketmq/common/client/ClientManager.java
浏览文件 @ 3629a0a7
......@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.snode.client;
package org.apache.rocketmq.common.client;
import java.util.List;
import java.util.Set;
......
snode/src/main/java/org/apache/rocketmq/snode/client/impl/ClientManagerImpl.java → common/src/main/java/org/apache/rocketmq/common/client/ClientManagerImpl.java
浏览文件 @ 3629a0a7
......@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.snode.client.impl;
package org.apache.rocketmq.common.client;
import java.util.ArrayList;
import java.util.Iterator;
......@@ -33,8 +33,6 @@ import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.ClientManager;
public abstract class ClientManagerImpl implements ClientManager {
......@@ -45,7 +43,7 @@ public abstract class ClientManagerImpl implements ClientManager {
.newSingleThreadScheduledExecutor(
new ThreadFactoryImpl("ClientHousekeepingScheduledThread"));
private final ConcurrentHashMap<String/*Producer or Consumer Group*/, ConcurrentHashMap<RemotingChannel, Client>> groupClientTable = new ConcurrentHashMap<>(
private final ConcurrentHashMap<String/*Producer or Consumer Group*/, ConcurrentHashMap<RemotingChannel, Client>> groupClientTable = new ConcurrentHashMap<String, ConcurrentHashMap<RemotingChannel, Client>>(
1024);
public abstract void onClosed(String group, RemotingChannel remotingChannel);
......@@ -176,7 +174,7 @@ public abstract class ClientManagerImpl implements ClientManager {
public List<RemotingChannel> getChannels(String groupId) {
if (groupId != null) {
List<RemotingChannel> result = new ArrayList<>();
List<RemotingChannel> result = new ArrayList<RemotingChannel>();
ConcurrentHashMap channelsMap = this.groupClientTable.get(groupId);
if (channelsMap != null) {
result.addAll(this.groupClientTable.get(groupId).keySet());
......@@ -189,7 +187,7 @@ public abstract class ClientManagerImpl implements ClientManager {
@Override
public List<String> getAllClientId(String groupId) {
List<String> result = new ArrayList<>();
List<String> result = new ArrayList<String>();
Map<RemotingChannel, Client> channelClientMap = this.groupClientTable.get(groupId);
if (channelClientMap != null) {
Iterator<Map.Entry<RemotingChannel, Client>> it = channelClientMap.entrySet()
......
snode/src/main/java/org/apache/rocketmq/snode/client/impl/ClientRole.java → common/src/main/java/org/apache/rocketmq/common/client/ClientRole.java
浏览文件 @ 3629a0a7
package org.apache.rocketmq.snode.client.impl;/*
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.rocketmq.common.client;/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
......
snode/src/main/java/org/apache/rocketmq/snode/client/impl/Subscription.java → common/src/main/java/org/apache/rocketmq/common/client/Subscription.java
浏览文件 @ 3629a0a7
......@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.snode.client.impl;
package org.apache.rocketmq.common.client;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
......@@ -27,7 +27,7 @@ public class Subscription {
private volatile MessageModel messageModel;
private volatile ConsumeFromWhere consumeFromWhere;
private volatile boolean cleanSession;
ConcurrentHashMap<String/*Topic*/, SubscriptionData> subscriptionTable = new ConcurrentHashMap<>();
ConcurrentHashMap<String/*Topic*/, SubscriptionData> subscriptionTable = new ConcurrentHashMap<String, SubscriptionData>();
private volatile long lastUpdateTimestamp = System.currentTimeMillis();
public SubscriptionData getSubscriptionData(String topic) {
......
common/src/main/java/org/apache/rocketmq/common/constant/LoggerName.java
浏览文件 @ 3629a0a7
......@@ -38,4 +38,5 @@ public class LoggerName {
public static final String WATER_MARK_LOGGER_NAME = "RocketmqWaterMark";
public static final String FILTER_LOGGER_NAME = "RocketmqFilter";
public static final String SNODE_LOGGER_NAME = "RocketmqSnode";
public static final String MQTT_LOGGER_NAME = "RocketmqMQTT";
}
distribution/conf/snode.conf
浏览文件 @ 3629a0a7
......@@ -15,4 +15,4 @@
namesrvAddr=localhost:9876
clusterName = DefaultCluster
snodeName = snode-a
embeddedModeEnable = false
distribution/release.xml
浏览文件 @ 3629a0a7
......@@ -67,6 +67,7 @@
<include>org.apache.rocketmq:rocketmq-namesrv</include>
<include>org.apache.rocketmq:rocketmq-example</include>
<include>org.apache.rocketmq:rocketmq-openmessaging</include>
<include>org.apache.rocketmq:rocketmq-mqtt</include>
<include>org.apache.rocketmq:rocketmq-snode</include>
</includes>
<binaries>
......
example/src/main/java/org/apache/rocketmq/example/mqtt/MqttSampleConsumer.java
浏览文件 @ 3629a0a7
......@@ -50,17 +50,17 @@ public class MqttSampleConsumer {
log.info("Connected");
sampleClient.setCallback(new MqttCallback() {
@Override public void connectionLost(Throwable throwable) {
System.out.println("connection lost." + throwable.getLocalizedMessage());
log.info("connection lost." + throwable.getLocalizedMessage());
}
@Override public void messageArrived(String s, MqttMessage message) throws Exception {
System.out.println(message.toString());
log.info(message.toString());
// System.exit(0);
}
@Override public void deliveryComplete(IMqttDeliveryToken token) {
try {
System.out.println("delivery complete." + token.getMessage().toString());
log.info("delivery complete." + token.getMessage().toString());
} catch (MqttException e) {
e.printStackTrace();
}
......
mqtt/pom.xml 0 → 100644
浏览文件 @ 3629a0a7
<!--
~ Licensed to the Apache Software Foundation (ASF) under one or more
~ contributor license agreements. See the NOTICE file distributed with
~ this work for additional information regarding copyright ownership.
~ The ASF licenses this file to You 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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-all</artifactId>
<version>4.4.1-SNAPSHOT</version>
</parent>
<artifactId>rocketmq-mqtt</artifactId>
<name>rocketmq-mqtt ${project.version}</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-common</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-store</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-remoting</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-client</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-srvutil</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-filter</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-acl</artifactId>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-core</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
</dependency>
<dependency>
<groupId>org.javassist</groupId>
<artifactId>javassist</artifactId>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
<dependency>
<groupId>org.yaml</groupId>
<artifactId>snakeyaml</artifactId>
</dependency>
<dependency>
<groupId>io.prometheus</groupId>
<artifactId>simpleclient</artifactId>
</dependency>
<dependency>
<groupId>io.prometheus</groupId>
<artifactId>simpleclient_httpserver</artifactId>
</dependency>
<dependency>
<groupId>io.prometheus</groupId>
<artifactId>simpleclient_hotspot</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-broker</artifactId>
</dependency>
</dependencies>
</project>
snode/src/main/java/org/apache/rocketmq/snode/client/impl/IOTClientManagerImpl.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/client/IOTClientManagerImpl.java
浏览文件 @ 3629a0a7
......@@ -14,34 +14,33 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.snode.client.impl;
package org.apache.rocketmq.mqtt.client;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.client.ClientManagerImpl;
import org.apache.rocketmq.common.client.Subscription;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.client.Client;
public class IOTClientManagerImpl extends ClientManagerImpl {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
public static final String IOT_GROUP = "IOT_GROUP";
private final SnodeController snodeController;
private final ConcurrentHashMap<String/*root topic*/, ConcurrentHashMap<Client, Set<SubscriptionData>>> topic2SubscriptionTable = new ConcurrentHashMap<>(
1024);
private final ConcurrentHashMap<String/*clientId*/, Subscription> clientId2Subscription = new ConcurrentHashMap<>(1024);
public IOTClientManagerImpl(SnodeController snodeController) {
this.snodeController = snodeController;
public IOTClientManagerImpl() {
}
public void onUnsubscribe(Client client, List<String> topics) {
......@@ -89,6 +88,9 @@ public class IOTClientManagerImpl extends ClientManagerImpl {
}
iterator1.remove();
}
if (next.getValue() == null || next.getValue().size() == 0) {
iterator.remove();
}
}
//remove offline messages
}
......@@ -97,10 +99,6 @@ public class IOTClientManagerImpl extends ClientManagerImpl {
return clientId2Subscription.get(clientId);
}
public SnodeController getSnodeController() {
return snodeController;
}
public ConcurrentHashMap<String, ConcurrentHashMap<Client, Set<SubscriptionData>>> getTopic2SubscriptionTable() {
return topic2SubscriptionTable;
}
......@@ -112,5 +110,4 @@ public class IOTClientManagerImpl extends ClientManagerImpl {
public void initSubscription(String clientId, Subscription subscription) {
clientId2Subscription.put(clientId, subscription);
}
}
mqtt/src/main/java/org/apache/rocketmq/mqtt/client/MqttClientHousekeepingService.java 0 → 100644
浏览文件 @ 3629a0a7
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.rocketmq.mqtt.client;
import io.netty.channel.Channel;
import io.netty.util.Attribute;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.client.ClientManager;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.constant.MqttConstant;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
public class MqttClientHousekeepingService implements ChannelEventListener {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final ClientManager iotClientManager;
public MqttClientHousekeepingService(final ClientManager iotClientManager) {
this.iotClientManager = iotClientManager;
}
public void start(long interval) {
this.iotClientManager.startScan(interval);
}
public void shutdown() {
this.iotClientManager.shutdown();
}
private Client getClient(RemotingChannel remotingChannel) {
if (remotingChannel instanceof NettyChannelImpl) {
Channel channel = ((NettyChannelImpl) remotingChannel).getChannel();
Attribute<Client> clientAttribute = channel.attr(MqttConstant.MQTT_CLIENT_ATTRIBUTE_KEY);
if (clientAttribute != null) {
Client client = clientAttribute.get();
return client;
}
}
log.warn("RemotingChannel type error: {}", remotingChannel.getClass());
return null;
}
private void closeChannel(String remoteAddress, RemotingChannel remotingChannel) {
Client client = getClient(remotingChannel);
if (client != null) {
switch (client.getClientRole()) {
case IOTCLIENT:
this.iotClientManager.onClose(client.getGroups(), remotingChannel);
return;
default:
}
}
log.warn("Close channel without any role");
}
@Override
public void onChannelConnect(String remoteAddr, RemotingChannel channel) {
log.info("Remoting channel connected: {}", RemotingHelper.parseChannelRemoteAddr(channel.remoteAddress()));
}
@Override
public void onChannelClose(String remoteAddr, RemotingChannel remotingChannel) {
log.info("Remoting channel closed: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
closeChannel(remoteAddr, remotingChannel);
}
@Override
public void onChannelException(String remoteAddr, RemotingChannel remotingChannel) {
log.info("Remoting channel exception: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
closeChannel(remoteAddr, remotingChannel);
}
@Override
public void onChannelIdle(String remoteAddr, RemotingChannel remotingChannel) {
log.info("Remoting channel idle: {}", RemotingHelper.parseChannelRemoteAddr(remotingChannel.remoteAddress()));
closeChannel(remoteAddr, remotingChannel);
}
}
snode/src/main/java/org/apache/rocketmq/snode/constant/MqttConstant.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/constant/MqttConstant.java
浏览文件 @ 3629a0a7
......@@ -15,11 +15,16 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.constant;
package org.apache.rocketmq.mqtt.constant;
import io.netty.util.AttributeKey;
import org.apache.rocketmq.common.client.Client;
public class MqttConstant {
public static final int MAX_SUPPORTED_QOS = 0;
public static final String SUBSCRIPTION_FLAG_PLUS = "+";
public static final String SUBSCRIPTION_FLAG_SHARP = "#";
public static final String SUBSCRIPTION_SEPARATOR = "/";
public static final long DEFAULT_TIMEOUT_MILLS = 3000L;
public static final AttributeKey<Client> MQTT_CLIENT_ATTRIBUTE_KEY = AttributeKey.valueOf("mqtt.client");
}
snode/src/main/java/org/apache/rocketmq/snode/exception/MqttConnectException.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/exception/MqttConnectException.java
浏览文件 @ 3629a0a7
......@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.exception;
package org.apache.rocketmq.mqtt.exception;
public class MqttConnectException extends RuntimeException {
......
snode/src/main/java/org/apache/rocketmq/snode/exception/WrongMessageTypeException.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/exception/WrongMessageTypeException.java
浏览文件 @ 3629a0a7
......@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.exception;
package org.apache.rocketmq.mqtt.exception;
public class WrongMessageTypeException extends RuntimeException {
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MessageHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/MessageHandler.java
浏览文件 @ 3629a0a7
......@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.apache.rocketmq.remoting.RemotingChannel;
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttConnectMessageHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttConnectMessageHandler.java
浏览文件 @ 3629a0a7
......@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
......@@ -25,32 +25,32 @@ import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.util.HashSet;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.client.ClientManager;
import org.apache.rocketmq.common.client.ClientRole;
import org.apache.rocketmq.common.client.Subscription;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.mqtt.WillMessage;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.client.IOTClientManagerImpl;
import org.apache.rocketmq.mqtt.exception.MqttConnectException;
import org.apache.rocketmq.mqtt.exception.WrongMessageTypeException;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.transport.mqtt.MqttHeader;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.ClientManager;
import org.apache.rocketmq.snode.client.impl.ClientRole;
import org.apache.rocketmq.snode.client.impl.IOTClientManagerImpl;
import org.apache.rocketmq.snode.client.impl.Subscription;
import org.apache.rocketmq.snode.exception.MqttConnectException;
import org.apache.rocketmq.snode.exception.WrongMessageTypeException;
public class MqttConnectMessageHandler implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
private final SnodeController snodeController;
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private static final int MIN_AVAILABLE_VERSION = 3;
private static final int MAX_AVAILABLE_VERSION = 4;
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
public MqttConnectMessageHandler(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttConnectMessageHandler(DefaultMqttMessageProcessor defaultMqttMessageProcessor) {
this.defaultMqttMessageProcessor = defaultMqttMessageProcessor;
}
@Override
......@@ -97,7 +97,7 @@ public class MqttConnectMessageHandler implements MessageHandler {
remotingChannel.close();
return null;
}
IOTClientManagerImpl iotClientManager = (IOTClientManagerImpl) snodeController.getIotClientManager();
IOTClientManagerImpl iotClientManager = (IOTClientManagerImpl) defaultMqttMessageProcessor.getIotClientManager();
//set Session Present according to whether the server has already stored Session State for the clientId
if (mqttConnectMessage.variableHeader().isCleanSession()) {
mqttHeader.setSessionPresent(false);
......@@ -120,7 +120,11 @@ public class MqttConnectMessageHandler implements MessageHandler {
Client client = new Client();
client.setClientId(payload.clientIdentifier());
client.setClientRole(ClientRole.IOTCLIENT);
client.setGroups(new HashSet<String>(){{add("IOT_GROUP");}});
client.setGroups(new HashSet<String>() {
{
add("IOT_GROUP");
}
});
client.setConnected(true);
client.setRemotingChannel(remotingChannel);
client.setLastUpdateTimestamp(System.currentTimeMillis());
......@@ -138,7 +142,7 @@ public class MqttConnectMessageHandler implements MessageHandler {
willMessage.setWillTopic(payload.willTopic());
willMessage.setRetain(mqttConnectMessage.variableHeader().isWillRetain());
willMessage.setBody(payload.willMessageInBytes());
snodeController.getWillMessageService().saveWillMessage(client.getClientId(), willMessage);
defaultMqttMessageProcessor.getWillMessageService().saveWillMessage(client.getClientId(), willMessage);
}
mqttHeader.setConnectReturnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED.name());
......@@ -148,7 +152,7 @@ public class MqttConnectMessageHandler implements MessageHandler {
}
private boolean alreadyStoredSession(String clientId) {
IOTClientManagerImpl iotClientManager = (IOTClientManagerImpl) snodeController.getIotClientManager();
IOTClientManagerImpl iotClientManager = (IOTClientManagerImpl) defaultMqttMessageProcessor.getIotClientManager();
Subscription subscription = iotClientManager.getSubscriptionByClientId(clientId);
if (subscription == null) {
return false;
......@@ -164,7 +168,7 @@ public class MqttConnectMessageHandler implements MessageHandler {
}
private boolean isConnected(RemotingChannel remotingChannel, String clientId) {
ClientManager iotClientManager = snodeController.getIotClientManager();
ClientManager iotClientManager = defaultMqttMessageProcessor.getIotClientManager();
Client client = iotClientManager.getClient(IOTClientManagerImpl.IOT_GROUP, remotingChannel);
if (client != null && client.getClientId().equals(clientId) && client.isConnected()) {
return true;
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttDisconnectMessageHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttDisconnectMessageHandler.java
浏览文件 @ 3629a0a7
......@@ -15,27 +15,27 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.client.IOTClientManagerImpl;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.impl.IOTClientManagerImpl;
public class MqttDisconnectMessageHandler implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
private final SnodeController snodeController;
public MqttDisconnectMessageHandler(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttDisconnectMessageHandler(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
/**
......@@ -56,10 +56,10 @@ public class MqttDisconnectMessageHandler implements MessageHandler {
}
//discard will message associated with the current connection(client)
Client client = snodeController.getIotClientManager()
Client client = defaultMqttMessageProcessor.getIotClientManager()
.getClient(IOTClientManagerImpl.IOT_GROUP, remotingChannel);
if (client != null) {
snodeController.getWillMessageService().deleteWillMessage(client.getClientId());
defaultMqttMessageProcessor.getWillMessageService().deleteWillMessage(client.getClientId());
}
client.setConnected(false);
if (remotingChannel.isActive()) {
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttMessageForwarder.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttMessageForwarder.java
浏览文件 @ 3629a0a7
......@@ -15,24 +15,23 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
public class MqttMessageForwarder implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
private final SnodeController snodeController;
/* private SubscriptionStore subscriptionStore;
public MqttMessageForwarder(SubscriptionStore subscriptionStore) {
this.subscriptionStore = subscriptionStore;
}*/
public MqttMessageForwarder(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttMessageForwarder(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
/**
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttMessageSender.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttMessageSender.java
浏览文件 @ 3629a0a7
......@@ -15,20 +15,25 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
public class MqttMessageSender implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
private final SnodeController snodeController;
public MqttMessageSender(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttMessageSender(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
/**
* send the PUBLISH message to client
*
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttPingreqMessageHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttPingreqMessageHandler.java
浏览文件 @ 3629a0a7
......@@ -15,29 +15,28 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
public class MqttPingreqMessageHandler implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
private final SnodeController snodeController;
public MqttPingreqMessageHandler(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttPingreqMessageHandler(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
/**
* handle the PINGREQ message from client
* <ol>
* <li>check client exists</li>
* <li>check client is connected</li>
* <li>generate the PINGRESP message</li>
* <li>send the PINGRESP message to the client</li>
* </ol>
* handle the PINGREQ message from client <ol> <li>check client exists</li> <li>check client is connected</li>
* <li>generate the PINGRESP message</li> <li>send the PINGRESP message to the client</li> </ol>
*
* @param message
* @return
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttPubackMessageHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttPubackMessageHandler.java
浏览文件 @ 3629a0a7
......@@ -15,26 +15,30 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
public class MqttPubackMessageHandler implements MessageHandler {
private final SnodeController snodeController;
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
public MqttPubackMessageHandler(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttPubackMessageHandler(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
/**
* handle the PUBACK message from the client
* <ol>
* <li>remove the message from the published in-flight messages</li>
* <li>ack the message in the MessageStore</li>
* </ol>
* handle the PUBACK message from the client <ol> <li>remove the message from the published in-flight messages</li>
* <li>ack the message in the MessageStore</li> </ol>
*
* @param
* @return
*/
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttPubcompMessageHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttPubcompMessageHandler.java
浏览文件 @ 3629a0a7
......@@ -15,22 +15,28 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
public class MqttPubcompMessageHandler implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
private final SnodeController snodeController;
public MqttPubcompMessageHandler(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttPubcompMessageHandler(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
/**
* handle the PUBCOMP message from the client
*
* @param message
* @return
*/
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttPublishMessageHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttPublishMessageHandler.java
浏览文件 @ 3629a0a7
......@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
......@@ -27,21 +27,20 @@ import io.netty.handler.codec.mqtt.MqttQoS;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.transport.mqtt.MqttHeader;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.exception.WrongMessageTypeException;
import org.apache.rocketmq.snode.util.MqttUtil;
import org.apache.rocketmq.mqtt.exception.WrongMessageTypeException;
import org.apache.rocketmq.mqtt.util.MqttUtil;
public class MqttPublishMessageHandler implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
private final SnodeController snodeController;
public MqttPublishMessageHandler(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttPublishMessageHandler(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
@Override
......@@ -61,7 +60,7 @@ public class MqttPublishMessageHandler implements MessageHandler {
ByteBuf payload = mqttPublishMessage.payload();
if (fixedHeader.qosLevel().equals(MqttQoS.AT_MOST_ONCE)) {
snodeController.getMqttPushService().pushMessageQos0(variableHeader.topicName(), payload);
defaultMqttMessageProcessor.getMqttPushService().pushMessageQos0(variableHeader.topicName(), payload);
} else if (fixedHeader.qosLevel().equals(MqttQoS.AT_LEAST_ONCE)) {
//Push messages to subscribers and add it to IN-FLIGHT messages
}
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttPubrecMessageHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttPubrecMessageHandler.java
浏览文件 @ 3629a0a7
......@@ -15,22 +15,28 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
public class MqttPubrecMessageHandler implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
private final SnodeController snodeController;
public MqttPubrecMessageHandler(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttPubrecMessageHandler(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
/**
* handle the PUBREC message from the clinet
*
* @param message
* @return
*/
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttPubrelMessageHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttPubrelMessageHandler.java
浏览文件 @ 3629a0a7
......@@ -15,23 +15,28 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
public class MqttPubrelMessageHandler implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
private final SnodeController snodeController;
public MqttPubrelMessageHandler(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttPubrelMessageHandler(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
/**
* handle the PUBREL message from the client
*
* @param message
* @return
*/
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttSubscribeMessageHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttSubscribeMessageHandler.java
浏览文件 @ 3629a0a7
......@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
......@@ -29,31 +29,32 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.client.Subscription;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.heartbeat.MqttSubscriptionData;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.client.IOTClientManagerImpl;
import org.apache.rocketmq.mqtt.constant.MqttConstant;
import org.apache.rocketmq.mqtt.exception.WrongMessageTypeException;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.mqtt.util.MqttUtil;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.transport.mqtt.MqttHeader;
import org.apache.rocketmq.remoting.util.MqttEncodeDecodeUtil;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.impl.IOTClientManagerImpl;
import org.apache.rocketmq.snode.client.impl.Subscription;
import org.apache.rocketmq.snode.constant.MqttConstant;
import org.apache.rocketmq.snode.exception.WrongMessageTypeException;
import org.apache.rocketmq.snode.util.MqttUtil;
public class MqttSubscribeMessageHandler implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
private final SnodeController snodeController;
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
public MqttSubscribeMessageHandler(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttSubscribeMessageHandler(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
/**
......@@ -72,7 +73,7 @@ public class MqttSubscribeMessageHandler implements MessageHandler {
}
MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) message;
MqttSubscribePayload payload = mqttSubscribeMessage.payload();
IOTClientManagerImpl iotClientManager = (IOTClientManagerImpl) snodeController.getIotClientManager();
IOTClientManagerImpl iotClientManager = (IOTClientManagerImpl) defaultMqttMessageProcessor.getIotClientManager();
Client client = iotClientManager.getClient(IOTClientManagerImpl.IOT_GROUP, remotingChannel);
if (client == null) {
log.error("Can't find associated client, the connection will be closed. remotingChannel={}, MqttMessage={}", remotingChannel.toString(), message.toString());
......@@ -135,7 +136,7 @@ public class MqttSubscribeMessageHandler implements MessageHandler {
//2.update topic2SubscriptionTable
String rootTopic = MqttUtil.getRootTopic(mqttTopicSubscription.topicName());
ConcurrentHashMap<Client, Set<SubscriptionData>> client2SubscriptionData = topic2SubscriptionTable.get(rootTopic);
if (client2SubscriptionData == null) {
if (client2SubscriptionData == null || client2SubscriptionData.size() == 0) {
client2SubscriptionData = new ConcurrentHashMap<>();
ConcurrentHashMap<Client, Set<SubscriptionData>> prev = topic2SubscriptionTable.putIfAbsent(rootTopic, client2SubscriptionData);
if (prev != null) {
......
snode/src/main/java/org/apache/rocketmq/snode/processor/mqtthandler/MqttUnsubscribeMessagHandler.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/mqtthandler/impl/MqttUnsubscribeMessagHandler.java
浏览文件 @ 3629a0a7
......@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor.mqtthandler;
package org.apache.rocketmq.mqtt.mqtthandler.impl;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
......@@ -27,31 +27,31 @@ import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.client.Subscription;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.client.IOTClientManagerImpl;
import org.apache.rocketmq.mqtt.exception.WrongMessageTypeException;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.mqtt.util.MqttUtil;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.transport.mqtt.MqttHeader;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.impl.IOTClientManagerImpl;
import org.apache.rocketmq.snode.client.impl.Subscription;
import org.apache.rocketmq.snode.exception.WrongMessageTypeException;
import org.apache.rocketmq.snode.util.MqttUtil;
/**
* handle the UNSUBSCRIBE message from the client
*/
public class MqttUnsubscribeMessagHandler implements MessageHandler {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private final DefaultMqttMessageProcessor defaultMqttMessageProcessor;
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
private final SnodeController snodeController;
public MqttUnsubscribeMessagHandler(SnodeController snodeController) {
this.snodeController = snodeController;
public MqttUnsubscribeMessagHandler(DefaultMqttMessageProcessor processor) {
this.defaultMqttMessageProcessor = processor;
}
@Override
......@@ -73,7 +73,7 @@ public class MqttUnsubscribeMessagHandler implements MessageHandler {
remotingChannel.close();
return null;
}
IOTClientManagerImpl iotClientManager = (IOTClientManagerImpl) snodeController.getIotClientManager();
IOTClientManagerImpl iotClientManager = (IOTClientManagerImpl) defaultMqttMessageProcessor.getIotClientManager();
Client client = iotClientManager.getClient(IOTClientManagerImpl.IOT_GROUP, remotingChannel);
if (client == null) {
log.error("Can't find associated client, the connection will be closed. remotingChannel={}, MqttMessage={}", remotingChannel.toString(), message.toString());
......@@ -98,7 +98,7 @@ public class MqttUnsubscribeMessagHandler implements MessageHandler {
}
private void doUnsubscribe(Client client, List<String> topics, IOTClientManagerImpl iotClientManager) {
ConcurrentHashMap<String, Subscription> clientId2Subscription = iotClientManager.getClientId2Subscription();
ConcurrentHashMap<String, Subscription> clientId2Subscription = iotClientManager.getClientId2Subscription();
ConcurrentHashMap<String, ConcurrentHashMap<Client, Set<SubscriptionData>>> topic2SubscriptionTable = iotClientManager.getTopic2SubscriptionTable();
for (String topicFilter : topics) {
......
snode/src/main/java/org/apache/rocketmq/snode/processor/DefaultMqttMessageProcessor.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/processor/DefaultMqttMessageProcessor.java
浏览文件 @ 3629a0a7
......@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor;
package org.apache.rocketmq.mqtt.processor;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
......@@ -33,55 +33,73 @@ import io.netty.handler.codec.mqtt.MqttSubscribePayload;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;
import org.apache.rocketmq.common.MqttConfig;
import org.apache.rocketmq.common.client.ClientManager;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.client.IOTClientManagerImpl;
import org.apache.rocketmq.mqtt.client.MqttClientHousekeepingService;
import org.apache.rocketmq.mqtt.mqtthandler.MessageHandler;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttConnectMessageHandler;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttDisconnectMessageHandler;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttPingreqMessageHandler;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttPubackMessageHandler;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttPubcompMessageHandler;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttPublishMessageHandler;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttPubrecMessageHandler;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttPubrelMessageHandler;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttSubscribeMessageHandler;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttUnsubscribeMessagHandler;
import org.apache.rocketmq.mqtt.service.WillMessageService;
import org.apache.rocketmq.mqtt.service.impl.MqttPushServiceImpl;
import org.apache.rocketmq.mqtt.service.impl.WillMessageServiceImpl;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.RemotingServer;
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;
import org.apache.rocketmq.remoting.util.MqttEncodeDecodeUtil;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.processor.mqtthandler.MessageHandler;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttConnectMessageHandler;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttDisconnectMessageHandler;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttPingreqMessageHandler;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttPubackMessageHandler;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttPubcompMessageHandler;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttPublishMessageHandler;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttPubrecMessageHandler;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttPubrelMessageHandler;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttSubscribeMessageHandler;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttUnsubscribeMessagHandler;
public class DefaultMqttMessageProcessor implements RequestProcessor {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private Map<MqttMessageType, MessageHandler> type2handler = new HashMap<>();
private final SnodeController snodeController;
private static final int MIN_AVAILABLE_VERSION = 3;
private static final int MAX_AVAILABLE_VERSION = 4;
private WillMessageService willMessageService;
private MqttPushServiceImpl mqttPushService;
private ClientManager iotClientManager;
private RemotingServer mqttRemotingServer;
private MqttClientHousekeepingService mqttClientHousekeepingService;
private MqttConfig mqttConfig;
public DefaultMqttMessageProcessor(MqttConfig mqttConfig, RemotingServer mqttRemotingServer) {
this.willMessageService = new WillMessageServiceImpl();
this.mqttPushService = new MqttPushServiceImpl(this, mqttConfig);
this.iotClientManager = new IOTClientManagerImpl();
this.mqttRemotingServer = mqttRemotingServer;
this.mqttClientHousekeepingService = new MqttClientHousekeepingService(iotClientManager);
this.mqttClientHousekeepingService.start(mqttConfig.getHouseKeepingInterval());
public DefaultMqttMessageProcessor(SnodeController snodeController) {
this.snodeController = snodeController;
registerMessageHandler(MqttMessageType.CONNECT,
new MqttConnectMessageHandler(this.snodeController));
new MqttConnectMessageHandler(this));
registerMessageHandler(MqttMessageType.DISCONNECT,
new MqttDisconnectMessageHandler(this.snodeController));
new MqttDisconnectMessageHandler(this));
registerMessageHandler(MqttMessageType.PINGREQ,
new MqttPingreqMessageHandler(this.snodeController));
new MqttPingreqMessageHandler(this));
registerMessageHandler(MqttMessageType.PUBLISH,
new MqttPublishMessageHandler(this.snodeController));
registerMessageHandler(MqttMessageType.PUBACK, new MqttPubackMessageHandler(this.snodeController));
new MqttPublishMessageHandler(this));
registerMessageHandler(MqttMessageType.PUBACK, new MqttPubackMessageHandler(this));
registerMessageHandler(MqttMessageType.PUBCOMP,
new MqttPubcompMessageHandler(this.snodeController));
registerMessageHandler(MqttMessageType.PUBREC, new MqttPubrecMessageHandler(this.snodeController));
registerMessageHandler(MqttMessageType.PUBREL, new MqttPubrelMessageHandler(this.snodeController));
new MqttPubcompMessageHandler(this));
registerMessageHandler(MqttMessageType.PUBREC, new MqttPubrecMessageHandler(this));
registerMessageHandler(MqttMessageType.PUBREL, new MqttPubrelMessageHandler(this));
registerMessageHandler(MqttMessageType.SUBSCRIBE,
new MqttSubscribeMessageHandler(this.snodeController));
new MqttSubscribeMessageHandler(this));
registerMessageHandler(MqttMessageType.UNSUBSCRIBE,
new MqttUnsubscribeMessagHandler(this.snodeController));
new MqttUnsubscribeMessagHandler(this));
}
@Override
......@@ -127,4 +145,24 @@ public class DefaultMqttMessageProcessor implements RequestProcessor {
private void registerMessageHandler(MqttMessageType type, MessageHandler handler) {
type2handler.put(type, handler);
}
public WillMessageService getWillMessageService() {
return willMessageService;
}
public MqttPushServiceImpl getMqttPushService() {
return mqttPushService;
}
public ClientManager getIotClientManager() {
return iotClientManager;
}
public MqttConfig getMqttConfig() {
return mqttConfig;
}
public RemotingServer getMqttRemotingServer() {
return mqttRemotingServer;
}
}
snode/src/main/java/org/apache/rocketmq/snode/service/WillMessageService.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/service/WillMessageService.java
浏览文件 @ 3629a0a7
......@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.service;
package org.apache.rocketmq.mqtt.service;
import org.apache.rocketmq.common.message.mqtt.WillMessage;
......
snode/src/main/java/org/apache/rocketmq/snode/service/impl/MqttPushServiceImpl.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/service/impl/MqttPushServiceImpl.java
浏览文件 @ 3629a0a7
......@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.snode.service.impl;
package org.apache.rocketmq.mqtt.service.impl;
import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.mqtt.MqttMessageType;
......@@ -27,37 +27,38 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.rocketmq.common.MqttConfig;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.client.IOTClientManagerImpl;
import org.apache.rocketmq.mqtt.constant.MqttConstant;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.mqtt.util.MqttUtil;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.transport.mqtt.MqttHeader;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.impl.IOTClientManagerImpl;
import org.apache.rocketmq.snode.constant.SnodeConstant;
import org.apache.rocketmq.snode.util.MqttUtil;
public class MqttPushServiceImpl {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.MQTT_LOGGER_NAME);
private SnodeController snodeController;
private ExecutorService pushMqttMessageExecutorService;
private DefaultMqttMessageProcessor defaultMqttMessageProcessor;
public MqttPushServiceImpl(final SnodeController snodeController) {
this.snodeController = snodeController;
public MqttPushServiceImpl(DefaultMqttMessageProcessor defaultMqttMessageProcessor, MqttConfig mqttConfig) {
this.defaultMqttMessageProcessor = defaultMqttMessageProcessor;
pushMqttMessageExecutorService = ThreadUtils.newThreadPoolExecutor(
this.snodeController.getMqttConfig().getPushMqttMessageMinPoolSize(),
this.snodeController.getMqttConfig().getPushMqttMessageMaxPoolSize(),
mqttConfig.getPushMqttMessageMinPoolSize(),
mqttConfig.getPushMqttMessageMaxPoolSize(),
3000,
TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(this.snodeController.getMqttConfig().getPushMqttMessageThreadPoolQueueCapacity()),
new ArrayBlockingQueue<>(mqttConfig.getPushMqttMessageThreadPoolQueueCapacity()),
"pushMqttMessageThread",
false);
}
......@@ -86,7 +87,7 @@ public class MqttPushServiceImpl {
RemotingCommand requestCommand = buildRequestCommand(topic, qos, retain, packetId);
//find those clients publishing the message to
IOTClientManagerImpl iotClientManager = (IOTClientManagerImpl) snodeController.getIotClientManager();
IOTClientManagerImpl iotClientManager = (IOTClientManagerImpl) defaultMqttMessageProcessor.getIotClientManager();
ConcurrentHashMap<String, ConcurrentHashMap<Client, Set<SubscriptionData>>> topic2SubscriptionTable = iotClientManager.getTopic2SubscriptionTable();
Set<Client> clients = new HashSet<>();
if (topic2SubscriptionTable.containsKey(MqttUtil.getRootTopic(topic))) {
......@@ -109,12 +110,11 @@ public class MqttPushServiceImpl {
byte[] body = new byte[message.readableBytes()];
message.readBytes(body);
requestCommand.setBody(body);
snodeController.getMqttRemotingServer().push(remotingChannel, requestCommand, SnodeConstant.DEFAULT_TIMEOUT_MILLS);
defaultMqttMessageProcessor.getMqttRemotingServer().push(remotingChannel, requestCommand, MqttConstant.DEFAULT_TIMEOUT_MILLS);
}
} catch (Exception ex) {
log.warn("Exception was thrown when pushing MQTT message to topic: {}, exception={}", topic, ex.getMessage());
} finally {
System.out.println("Release Bytebuf");
ReferenceCountUtil.release(message);
}
} else {
......
snode/src/main/java/org/apache/rocketmq/snode/service/impl/WillMessageServiceImpl.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/service/impl/WillMessageServiceImpl.java
浏览文件 @ 3629a0a7
......@@ -15,20 +15,17 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.service.impl;
package org.apache.rocketmq.mqtt.service.impl;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.message.mqtt.WillMessage;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.service.WillMessageService;
import org.apache.rocketmq.mqtt.service.WillMessageService;
public class WillMessageServiceImpl implements WillMessageService {
private static ConcurrentHashMap<String/*clientId*/, WillMessage> willMessageTable = new ConcurrentHashMap<>();
private final SnodeController snodeController;
public WillMessageServiceImpl(SnodeController snodeController) {
this.snodeController = snodeController;
public WillMessageServiceImpl() {
}
@Override
......
snode/src/main/java/org/apache/rocketmq/snode/util/MqttUtil.java → mqtt/src/main/java/org/apache/rocketmq/mqtt/util/MqttUtil.java
浏览文件 @ 3629a0a7
......@@ -15,11 +15,11 @@
* limitations under the License.
*/
package org.apache.rocketmq.snode.util;
package org.apache.rocketmq.mqtt.util;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.util.UUID;
import org.apache.rocketmq.snode.constant.MqttConstant;
import org.apache.rocketmq.mqtt.constant.MqttConstant;
public class MqttUtil {
......
snode/src/test/java/org/apache/rocketmq/snode/processor/DefaultMqttMessageProcessorTest.java → mqtt/src/test/java/org/apache/rocketmq/mqtt/DefaultMqttMessageProcessorTest.java
浏览文件 @ 3629a0a7
......@@ -14,38 +14,37 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor;
package org.apache.rocketmq.mqtt;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;
import java.io.UnsupportedEncodingException;
import org.apache.rocketmq.common.MqttConfig;
import org.apache.rocketmq.common.SnodeConfig;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.transport.mqtt.MqttHeader;
import org.apache.rocketmq.remoting.transport.mqtt.MqttRemotingServer;
import org.apache.rocketmq.remoting.util.MqttEncodeDecodeUtil;
import org.apache.rocketmq.snode.SnodeController;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.junit.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class DefaultMqttMessageProcessorTest {
private DefaultMqttMessageProcessor defaultMqttMessageProcessor;
@Spy
private SnodeController snodeController = new SnodeController(new SnodeConfig(), new MqttConfig());
@Mock
private RemotingChannel remotingChannel;
@Mock
private MqttRemotingServer mqttRemotingServer;
private String topic = "SnodeTopic";
private String group = "SnodeGroup";
......@@ -54,7 +53,7 @@ public class DefaultMqttMessageProcessorTest {
@Before
public void init() {
defaultMqttMessageProcessor = new DefaultMqttMessageProcessor(snodeController);
defaultMqttMessageProcessor = new DefaultMqttMessageProcessor(new MqttConfig(), mqttRemotingServer);
}
@Test
......
snode/src/test/java/org/apache/rocketmq/snode/processor/MqttConnectMessageHandlerTest.java → mqtt/src/test/java/org/apache/rocketmq/mqtt/MqttConnectMessageHandlerTest.java
浏览文件 @ 3629a0a7
......@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor;
package org.apache.rocketmq.mqtt;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
......@@ -22,11 +22,9 @@ import io.netty.handler.codec.mqtt.MqttConnectVariableHeader;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.apache.rocketmq.common.MqttConfig;
import org.apache.rocketmq.common.SnodeConfig;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttConnectMessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttConnectMessageHandler;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
......@@ -38,10 +36,13 @@ public class MqttConnectMessageHandlerTest {
@Mock
private RemotingChannel remotingChannel;
@Mock
private DefaultMqttMessageProcessor defaultMqttMessageProcessor;
@Test
public void testHandlerMessage() throws Exception {
MqttConnectMessageHandler mqttConnectMessageHandler = new MqttConnectMessageHandler(new SnodeController(new SnodeConfig(), new MqttConfig()));
MqttConnectMessageHandler mqttConnectMessageHandler = new MqttConnectMessageHandler(defaultMqttMessageProcessor);
MqttConnectPayload payload = new MqttConnectPayload("1234567", "testTopic", "willMessage".getBytes(), null, "1234567".getBytes());
MqttConnectMessage mqttConnectMessage = new MqttConnectMessage(new MqttFixedHeader(
MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 200), new MqttConnectVariableHeader(null, 4, false, false, false, 0, false, false, 50), new MqttConnectPayload("abcd", "ttest", "message".getBytes(), "user", "password".getBytes()));
......
snode/src/test/java/org/apache/rocketmq/snode/processor/MqttDisconnectMessageHandlerTest.java → mqtt/src/test/java/org/apache/rocketmq/mqtt/MqttDisconnectMessageHandlerTest.java
浏览文件 @ 3629a0a7
......@@ -14,20 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.snode.processor;
package org.apache.rocketmq.mqtt;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.apache.rocketmq.common.MqttConfig;
import org.apache.rocketmq.common.SnodeConfig;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.message.mqtt.WillMessage;
import org.apache.rocketmq.mqtt.client.IOTClientManagerImpl;
import org.apache.rocketmq.mqtt.mqtthandler.impl.MqttDisconnectMessageHandler;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.impl.IOTClientManagerImpl;
import org.apache.rocketmq.snode.processor.mqtthandler.MqttDisconnectMessageHandler;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
......@@ -39,17 +37,19 @@ public class MqttDisconnectMessageHandlerTest {
@Mock
private RemotingChannel remotingChannel;
@Mock
private DefaultMqttMessageProcessor defaultMqttMessageProcessor;
@Test
public void testHandlerMessage() throws Exception {
SnodeController snodeController = new SnodeController(new SnodeConfig(), new MqttConfig());
MqttDisconnectMessageHandler mqttDisconnectMessageHandler = new MqttDisconnectMessageHandler(
snodeController);
defaultMqttMessageProcessor);
Client client = new Client();
client.setRemotingChannel(remotingChannel);
client.setClientId("123456");
snodeController.getIotClientManager().register(IOTClientManagerImpl.IOT_GROUP, client);
snodeController.getWillMessageService().saveWillMessage("123456", new WillMessage());
defaultMqttMessageProcessor.getIotClientManager().register(IOTClientManagerImpl.IOT_GROUP, client);
defaultMqttMessageProcessor.getWillMessageService().saveWillMessage("123456", new WillMessage());
MqttMessage mqttDisconnectMessage = new MqttMessage(new MqttFixedHeader(
MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 200));
......
snode/src/test/java/org/apache/rocketmq/snode/service/WillMessageServiceImplTest.java → mqtt/src/test/java/org/apache/rocketmq/mqtt/WillMessageServiceImplTest.java
浏览文件 @ 3629a0a7
......@@ -14,31 +14,24 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.snode.service;
package org.apache.rocketmq.mqtt;
import org.apache.rocketmq.common.MqttConfig;
import org.apache.rocketmq.common.SnodeConfig;
import org.apache.rocketmq.common.message.mqtt.WillMessage;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.SnodeTestBase;
import org.apache.rocketmq.snode.service.impl.WillMessageServiceImpl;
import org.apache.rocketmq.mqtt.service.WillMessageService;
import org.apache.rocketmq.mqtt.service.impl.WillMessageServiceImpl;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Spy;
import org.mockito.junit.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class WillMessageServiceImplTest extends SnodeTestBase {
@Spy
private SnodeController snodeController = new SnodeController(new SnodeConfig(), new MqttConfig());
public class WillMessageServiceImplTest {
private WillMessageService willMessageService;
@Before
public void init() {
willMessageService = new WillMessageServiceImpl(snodeController);
willMessageService = new WillMessageServiceImpl();
}
@Test
......
pom.xml
浏览文件 @ 3629a0a7
......@@ -126,6 +126,7 @@
<module>distribution</module>
<module>openmessaging</module>
<module>logging</module>
<module>mqtt</module>
<module>snode</module>
<module>acl</module>
</modules>
......@@ -533,6 +534,11 @@
<artifactId>rocketmq-snode</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-mqtt</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
......
snode/pom.xml
浏览文件 @ 3629a0a7
......@@ -97,6 +97,10 @@
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-broker</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>rocketmq-mqtt</artifactId>
</dependency>
</dependencies>
<build>
......
snode/src/main/java/org/apache/rocketmq/snode/SnodeController.java
浏览文件 @ 3629a0a7
......@@ -27,11 +27,13 @@ import org.apache.rocketmq.broker.BrokerStartup;
import org.apache.rocketmq.common.MqttConfig;
import org.apache.rocketmq.common.SnodeConfig;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.client.ClientManager;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.protocol.RequestCode;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.mqtt.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.remoting.ClientConfig;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.RemotingClientFactory;
......@@ -48,18 +50,15 @@ import org.apache.rocketmq.remoting.interceptor.ResponseContext;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.util.ServiceProvider;
import org.apache.rocketmq.snode.client.ClientHousekeepingService;
import org.apache.rocketmq.snode.client.ClientManager;
import org.apache.rocketmq.snode.client.SlowConsumerService;
import org.apache.rocketmq.snode.client.SubscriptionGroupManager;
import org.apache.rocketmq.snode.client.SubscriptionManager;
import org.apache.rocketmq.snode.client.impl.ConsumerManagerImpl;
import org.apache.rocketmq.snode.client.impl.IOTClientManagerImpl;
import org.apache.rocketmq.snode.client.impl.ProducerManagerImpl;
import org.apache.rocketmq.snode.client.impl.SlowConsumerServiceImpl;
import org.apache.rocketmq.snode.client.impl.SubscriptionManagerImpl;
import org.apache.rocketmq.snode.offset.ConsumerOffsetManager;
import org.apache.rocketmq.snode.processor.ConsumerManageProcessor;
import org.apache.rocketmq.snode.processor.DefaultMqttMessageProcessor;
import org.apache.rocketmq.snode.processor.HeartbeatProcessor;
import org.apache.rocketmq.snode.processor.PullMessageProcessor;
import org.apache.rocketmq.snode.processor.SendMessageProcessor;
......@@ -69,16 +68,13 @@ import org.apache.rocketmq.snode.service.MetricsService;
import org.apache.rocketmq.snode.service.NnodeService;
import org.apache.rocketmq.snode.service.PushService;
import org.apache.rocketmq.snode.service.ScheduledService;
import org.apache.rocketmq.snode.service.WillMessageService;
import org.apache.rocketmq.snode.service.impl.ClientServiceImpl;
import org.apache.rocketmq.snode.service.impl.LocalEnodeServiceImpl;
import org.apache.rocketmq.snode.service.impl.MetricsServiceImpl;
import org.apache.rocketmq.snode.service.impl.MqttPushServiceImpl;
import org.apache.rocketmq.snode.service.impl.NnodeServiceImpl;
import org.apache.rocketmq.snode.service.impl.PushServiceImpl;
import org.apache.rocketmq.snode.service.impl.RemoteEnodeServiceImpl;
import org.apache.rocketmq.snode.service.impl.ScheduledServiceImpl;
import org.apache.rocketmq.snode.service.impl.WillMessageServiceImpl;
public class SnodeController {
......@@ -105,7 +101,7 @@ public class SnodeController {
private ScheduledService scheduledService;
private ClientManager producerManager;
private ClientManager consumerManager;
private ClientManager iotClientManager;
// private ClientManager iotClientManager;
private SubscriptionManager subscriptionManager;
private ClientHousekeepingService clientHousekeepingService;
private SubscriptionGroupManager subscriptionGroupManager;
......@@ -122,8 +118,8 @@ public class SnodeController {
private ClientService clientService;
private SlowConsumerService slowConsumerService;
private MetricsService metricsService;
private WillMessageService willMessageService;
private MqttPushServiceImpl mqttPushService;
// private WillMessageService willMessageService;
// private MqttPushServiceImpl mqttPushService;
private final ScheduledExecutorService scheduledExecutorService = Executors
.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
......@@ -153,7 +149,12 @@ public class SnodeController {
if (this.mqttRemotingClient != null) {
this.mqttRemotingClient.init(this.mqttClientConfig, null);
}
this.mqttRemotingServer = RemotingServerFactory.getInstance().createRemotingServer(
RemotingUtil.MQTT_PROTOCOL);
if (this.mqttRemotingServer != null) {
this.mqttRemotingServer.init(this.mqttServerConfig, this.clientHousekeepingService);
this.mqttRemotingServer.registerInterceptorGroup(this.remotingServerInterceptorGroup);
}
this.sendMessageExecutor = ThreadUtils.newThreadPoolExecutor(
snodeConfig.getSnodeSendMessageMinPoolSize(),
snodeConfig.getSnodeSendMessageMaxPoolSize(),
......@@ -211,19 +212,19 @@ public class SnodeController {
this.sendMessageProcessor = new SendMessageProcessor(this);
this.heartbeatProcessor = new HeartbeatProcessor(this);
this.pullMessageProcessor = new PullMessageProcessor(this);
this.defaultMqttMessageProcessor = new DefaultMqttMessageProcessor(this);
this.defaultMqttMessageProcessor = new DefaultMqttMessageProcessor(this.mqttConfig, mqttRemotingServer);
this.pushService = new PushServiceImpl(this);
this.clientService = new ClientServiceImpl(this);
this.subscriptionManager = new SubscriptionManagerImpl();
this.producerManager = new ProducerManagerImpl();
this.consumerManager = new ConsumerManagerImpl(this);
this.iotClientManager = new IOTClientManagerImpl(this);
// this.iotClientManager = new IOTClientManagerImpl(this);
this.clientHousekeepingService = new ClientHousekeepingService(this.producerManager,
this.consumerManager, this.iotClientManager);
this.consumerManager, null);
this.slowConsumerService = new SlowConsumerServiceImpl(this);
this.metricsService = new MetricsServiceImpl();
this.willMessageService = new WillMessageServiceImpl(this);
this.mqttPushService = new MqttPushServiceImpl(this);
// this.willMessageService = new WillMessageServiceImpl(this);
// this.mqttPushService = new MqttPushServiceImpl(this);
}
public SnodeConfig getSnodeConfig() {
......@@ -258,12 +259,6 @@ public class SnodeController {
this.snodeServer.init(this.nettyServerConfig, this.clientHousekeepingService);
this.snodeServer.registerInterceptorGroup(this.remotingServerInterceptorGroup);
}
this.mqttRemotingServer = RemotingServerFactory.getInstance().createRemotingServer(
RemotingUtil.MQTT_PROTOCOL);
if (this.mqttRemotingServer != null) {
this.mqttRemotingServer.init(this.mqttServerConfig, this.clientHousekeepingService);
this.mqttRemotingServer.registerInterceptorGroup(this.remotingServerInterceptorGroup);
}
registerProcessor();
return true;
}
......@@ -507,13 +502,13 @@ public class SnodeController {
this.consumerManager = consumerManager;
}
public ClientManager getIotClientManager() {
/* public ClientManager getIotClientManager() {
return iotClientManager;
}
public void setIotClientManager(ClientManager iotClientManager) {
this.iotClientManager = iotClientManager;
}
}*/
public SubscriptionManager getSubscriptionManager() {
return subscriptionManager;
......@@ -551,7 +546,7 @@ public class SnodeController {
this.metricsService = metricsService;
}
public WillMessageService getWillMessageService() {
/* public WillMessageService getWillMessageService() {
return willMessageService;
}
......@@ -566,5 +561,5 @@ public class SnodeController {
public void setMqttPushService(MqttPushServiceImpl mqttPushService) {
this.mqttPushService = mqttPushService;
}
}*/
}
snode/src/main/java/org/apache/rocketmq/snode/client/ClientHousekeepingService.java
浏览文件 @ 3629a0a7
......@@ -18,6 +18,8 @@ package org.apache.rocketmq.snode.client;
import io.netty.channel.Channel;
import io.netty.util.Attribute;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.client.ClientManager;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
......@@ -49,7 +51,7 @@ public class ClientHousekeepingService implements ChannelEventListener {
public void shutdown() {
this.producerManager.shutdown();
this.consumerManager.shutdown();
this.iotClientManager.shutdown();
// this.iotClientManager.shutdown();
}
private Client getClient(RemotingChannel remotingChannel) {
......@@ -75,9 +77,9 @@ public class ClientHousekeepingService implements ChannelEventListener {
case Producer:
this.producerManager.onClose(client.getGroups(), remotingChannel);
return;
case IOTCLIENT:
this.iotClientManager.onClose(client.getGroups(), remotingChannel);
return;
// case IOTCLIENT:
// this.iotClientManager.onClose(client.getGroups(), remotingChannel);
// return;
default:
}
}
......
snode/src/main/java/org/apache/rocketmq/snode/client/SubscriptionManager.java
浏览文件 @ 3629a0a7
......@@ -17,13 +17,13 @@
package org.apache.rocketmq.snode.client;
import java.util.Set;
import org.apache.rocketmq.common.client.Subscription;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.RemotingChannel;
import org.apache.rocketmq.snode.client.impl.Subscription;
public interface SubscriptionManager {
boolean subscribe(String groupId, Set<SubscriptionData> subscriptionDataSet, ConsumeType consumeType,
......
snode/src/main/java/org/apache/rocketmq/snode/client/impl/ConsumerManagerImpl.java
浏览文件 @ 3629a0a7
......@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.snode.client.impl;
import org.apache.rocketmq.common.client.ClientManagerImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
......
snode/src/main/java/org/apache/rocketmq/snode/client/impl/ProducerManagerImpl.java
浏览文件 @ 3629a0a7
......@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.snode.client.impl;
import org.apache.rocketmq.common.client.ClientManagerImpl;
import org.apache.rocketmq.remoting.RemotingChannel;
public class ProducerManagerImpl extends ClientManagerImpl {
......
snode/src/main/java/org/apache/rocketmq/snode/client/impl/SubscriptionManagerImpl.java
浏览文件 @ 3629a0a7
......@@ -21,6 +21,7 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.common.client.Subscription;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageQueue;
......
snode/src/main/java/org/apache/rocketmq/snode/constant/SnodeConstant.java
浏览文件 @ 3629a0a7
......@@ -17,8 +17,8 @@
package org.apache.rocketmq.snode.constant;
import io.netty.util.AttributeKey;
import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.impl.ClientRole;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.client.ClientRole;
public class SnodeConstant {
public static final long HEARTBEAT_TIME_OUT = 3000;
......
snode/src/main/java/org/apache/rocketmq/snode/processor/HeartbeatProcessor.java
浏览文件 @ 3629a0a7
......@@ -20,6 +20,8 @@ import io.netty.channel.Channel;
import io.netty.util.Attribute;
import java.util.HashSet;
import java.util.Set;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.client.ClientRole;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.filter.ExpressionType;
import org.apache.rocketmq.common.protocol.RequestCode;
......@@ -41,8 +43,6 @@ import org.apache.rocketmq.remoting.netty.NettyChannelHandlerContextImpl;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.impl.ClientRole;
import org.apache.rocketmq.snode.constant.SnodeConstant;
public class HeartbeatProcessor implements RequestProcessor {
......
snode/src/main/java/org/apache/rocketmq/snode/processor/PullMessageProcessor.java
浏览文件 @ 3629a0a7
......@@ -17,6 +17,7 @@
package org.apache.rocketmq.snode.processor;
import java.util.concurrent.CompletableFuture;
import org.apache.rocketmq.common.client.Subscription;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.help.FAQUrl;
import org.apache.rocketmq.common.protocol.ResponseCode;
......@@ -36,7 +37,6 @@ import org.apache.rocketmq.remoting.interceptor.RequestContext;
import org.apache.rocketmq.remoting.interceptor.ResponseContext;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.client.impl.Subscription;
public class PullMessageProcessor implements RequestProcessor {
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.SNODE_LOGGER_NAME);
......
snode/src/main/java/org/apache/rocketmq/snode/service/impl/PushServiceImpl.java
浏览文件 @ 3629a0a7
......@@ -24,6 +24,8 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.client.Client;
import org.apache.rocketmq.common.client.Subscription;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
......@@ -40,8 +42,6 @@ import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.apache.rocketmq.remoting.netty.NettyChannelImpl;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.snode.SnodeController;
import org.apache.rocketmq.snode.client.Client;
import org.apache.rocketmq.snode.client.impl.Subscription;
import org.apache.rocketmq.snode.constant.SnodeConstant;
import org.apache.rocketmq.snode.service.PushService;
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或