anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加限频配置项

Browse Source
This commit is contained in:
heavyrain2012
2021-10-26 10:11:05 +08:00
parent 5e9bbcb1e9
commit ffd1492ea2
12 changed files with 101 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/BrokerConstants.java
+4
View File
@@ -178,6 +178,10 @@ public final class BrokerConstants {
public static final String MESSAGES_RECALL_TIME_LIMIT = "message.recall_time_limit";
public static final String MESSAGES_DISABLE_GROUP_MANAGER_RECALL = "message.disable_group_manager_recall";
public static final String HTTP_ADMIN_RATE_LIMIT = "http.admin.rate_limit";
public static final String HTTP_ROBOT_RATE_LIMIT = "http.robot.rate_limit";
public static final String HTTP_CHANNEL_RATE_LIMIT = "http.channel.rate_limit";
public static final String CLIENT_REQUEST_RATE_LIMIT = "client.request_rate_limit";
private BrokerConstants() {
}
broker/src/main/java/io/moquette/imhandler/IMHandler.java
+18 -5
View File
@@ -38,6 +38,7 @@ import java.util.*;
import static cn.wildfirechat.common.ErrorCode.ERROR_CODE_OVER_FREQUENCY;
import static cn.wildfirechat.common.ErrorCode.ERROR_CODE_SUCCESS;
import static io.moquette.BrokerConstants.CLIENT_REQUEST_RATE_LIMIT;
/**
* 请求处理接口<br>
@@ -52,7 +53,7 @@ abstract public class IMHandler<T> {
protected static Server mServer = null;
protected static MessagesPublisher publisher;
private static ThreadPoolExecutorWrapper m_imBusinessExecutor;
private static final RateLimiter mLimitCounter = new RateLimiter(5, 100);
private static RateLimiter mLimitCounter;
private Method parseDataMethod;
private Class dataCls;
@@ -147,12 +148,24 @@ abstract public class IMHandler<T> {
publisher = p;
m_imBusinessExecutor = businessExecutor;
mServer = server;
int clientRateLimit = 100;
try {
clientRateLimit = Integer.parseInt(server.getConfig().getProperty(CLIENT_REQUEST_RATE_LIMIT, "100"));
} catch (Exception e) {
}
if(clientRateLimit == 0) {
clientRateLimit = 100;
}
mLimitCounter = new RateLimiter(5, clientRateLimit);
}
public ErrorCode preAction(String clientID, String fromUser, String topic, Qos1PublishHandler.IMCallback callback) {
public ErrorCode preAction(String clientID, String fromUser, String topic, Qos1PublishHandler.IMCallback callback, boolean isAdmin, boolean isRobotOrChannel) {
LOG.info("imHandler fromUser={}, clientId={}, topic={}", fromUser, clientID, topic);
if(!mLimitCounter.isGranted(clientID + fromUser + topic)) {
if(!isAdmin && !isRobotOrChannel && !mLimitCounter.isGranted(clientID + fromUser + topic)) {
ByteBuf ackPayload = Unpooled.buffer();
ackPayload.ensureWritable(1).writeByte(ERROR_CODE_OVER_FREQUENCY.getCode());
try {
@@ -166,7 +179,7 @@ abstract public class IMHandler<T> {
return ErrorCode.ERROR_CODE_SUCCESS;
}
public void doHandler(String clientID, String fromUser, String topic, byte[] payloadContent, Qos1PublishHandler.IMCallback callback, boolean isAdmin) {
public void doHandler(String clientID, String fromUser, String topic, byte[] payloadContent, Qos1PublishHandler.IMCallback callback, boolean isAdmin, boolean isRobotOrChannel) {
m_imBusinessExecutor.execute(() -> {
Qos1PublishHandler.IMCallback callbackWrapper = new Qos1PublishHandler.IMCallback() {
@Override
@@ -177,7 +190,7 @@ abstract public class IMHandler<T> {
}
};
ErrorCode preActionCode = preAction(clientID, fromUser, topic, callbackWrapper);
ErrorCode preActionCode = preAction(clientID, fromUser, topic, callbackWrapper, isAdmin, isRobotOrChannel);
if (preActionCode == ErrorCode.ERROR_CODE_SUCCESS) {
ByteBuf ackPayload = Unpooled.buffer(1);
broker/src/main/java/io/moquette/server/Server.java
+2
View File
@@ -24,6 +24,7 @@ import com.hazelcast.config.FileSystemXmlConfig;
import com.hazelcast.core.*;
import com.xiaoleilu.loServer.LoServer;
import com.xiaoleilu.loServer.ServerSetting;
import com.xiaoleilu.loServer.action.Action;
import com.xiaoleilu.loServer.action.admin.AdminAction;
import io.moquette.BrokerConstants;
import io.moquette.persistence.*;
@@ -257,6 +258,7 @@ public class Server {
initMediaServerConfig(config);
Action.init(config);
String monitorEventAddress = config.getProperty(MONITOR_Exception_Event_Address);
Utility.setMonitorEventAddress(monitorEventAddress);
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+1 -15
View File
@@ -69,7 +69,6 @@ public class Qos1PublishHandler extends QosPublishHandler {
private final MessagesPublisher publisher;
private final ISessionsStore m_sessionStore;
private final ThreadPoolExecutorWrapper m_imBusinessExecutor;
private final RateLimiter mLimitCounter = new RateLimiter(5, 100);
private HashMap<String, IMHandler> m_imHandlers = new HashMap<>();
@@ -148,19 +147,6 @@ public class Qos1PublishHandler extends QosPublishHandler {
void imHandler(String clientID, String fromUser, String topic, byte[] payloadContent, IMCallback callback, boolean isAdmin, boolean isRobotOrChannel) {
LOG.info("imHandler fromUser={}, topic={}", fromUser, topic);
if(!isAdmin && !isRobotOrChannel && !mLimitCounter.isGranted(clientID + fromUser + topic)) {
ByteBuf ackPayload = Unpooled.buffer();
ackPayload.ensureWritable(1).writeByte(ERROR_CODE_OVER_FREQUENCY.getCode());
try {
callback.onIMHandled(ERROR_CODE_OVER_FREQUENCY, ackPayload);
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
}
LOG.warn("user {} request over frequency", fromUser);
return;
}
IMCallback wrapper = (errorcode, ackPayload) -> {
ackPayload.resetReaderIndex();
byte code = ackPayload.readByte();
@@ -215,7 +201,7 @@ public class Qos1PublishHandler extends QosPublishHandler {
IMHandler handler = m_imHandlers.get(topic);
if (handler != null) {
handler.doHandler(clientID, fromUser, topic, payloadContent, wrapper, isAdmin);
handler.doHandler(clientID, fromUser, topic, payloadContent, wrapper, isAdmin, isRobotOrChannel);
} else {
LOG.error("imHandler unknown topic={}", topic);
ByteBuf ackPayload = Unpooled.buffer();