diff --git a/broker/config/wildfirechat.conf b/broker/config/wildfirechat.conf
index 2d47d456..7287ba96 100755
--- a/broker/config/wildfirechat.conf
+++ b/broker/config/wildfirechat.conf
@@ -154,6 +154,20 @@ group.visible_quit_or_kickoff_notification 0
# 因此需要分段多次同步，此开关定义了每次同步的数据的条数。此需要客户端支持，客户端SDK在2021.5.25日后支持此功能。如果确认所有客户端都在这个日志之后才可以打开此设置。
#sync.data_part_size 1000
+#*********************************************************************
+# 限频配置
+#*********************************************************************
+##API接口限频设置，时间是10秒内允许的请求次数。
+##管理api频率限制，默认是10秒10000次。
+http.admin.rate_limit 10000
+##机器人api频率限制，默认是10秒1000次，按照机器人ID区分。
+http.robot.rate_limit 1000
+##频道api频率限制，默认是10秒1000次，按照频道ID区分。
+http.channel.rate_limit 1000
+
+##客户端请求频率限制，默认是5秒100次。通常情况下不要修改这个值。
+client.request_rate_limit 100
+
#*********************************************************************
# hazelcase configuration
#*********************************************************************
diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/Action.java b/broker/src/main/java/com/xiaoleilu/loServer/action/Action.java
index bb36da53..44a376b4 100755
--- a/broker/src/main/java/com/xiaoleilu/loServer/action/Action.java
+++ b/broker/src/main/java/com/xiaoleilu/loServer/action/Action.java
@@ -13,6 +13,7 @@ import com.xiaoleilu.loServer.RestResult;
import com.xiaoleilu.loServer.annotation.RequireAuthentication;
import com.xiaoleilu.loServer.handler.Request;
import com.xiaoleilu.loServer.handler.Response;
+import io.moquette.server.config.IConfig;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.Utils;
@@ -21,10 +22,12 @@ import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import cn.wildfirechat.common.ErrorCode;
+import win.liyufan.im.RateLimiter;
import java.nio.charset.StandardCharsets;
import static cn.wildfirechat.common.ErrorCode.ERROR_CODE_SUCCESS;
+import static io.moquette.BrokerConstants.*;
/**
* 请求处理接口
@@ -38,6 +41,36 @@ abstract public class Action {
public static ISessionsStore sessionsStore = null;
public ChannelHandlerContext ctx;
+
+ protected static RateLimiter adminLimiter = null;
+ protected static RateLimiter robotLimiter = null;
+ protected static RateLimiter channelLimiter = null;
+
+ public static void init(IConfig config) {
+ int adminRate = 10000;
+ int robotRate = 1000;
+ int channelRate = 1000;
+
+ try {
+ adminRate = Integer.parseInt(config.getProperty(HTTP_ADMIN_RATE_LIMIT, "10000"));
+ } catch (NumberFormatException e) {
+
+ }
+ try {
+ robotRate = Integer.parseInt(config.getProperty(HTTP_ROBOT_RATE_LIMIT, "1000"));
+ } catch (NumberFormatException e) {
+
+ }
+ try {
+ channelRate = Integer.parseInt(config.getProperty(HTTP_CHANNEL_RATE_LIMIT, "1000"));
+ } catch (NumberFormatException e) {
+
+ }
+ adminLimiter = new RateLimiter(10, adminRate);
+ robotLimiter = new RateLimiter(10, robotRate);
+ channelLimiter = new RateLimiter(10, channelRate);
+ }
+
protected class Result {
Object data;
ErrorCode errorCode;
diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AdminAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AdminAction.java
index 7d677de3..624beba1 100644
--- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AdminAction.java
+++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AdminAction.java
@@ -29,7 +29,6 @@ abstract public class AdminAction extends Action {
private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(AdminAction.class);
private static String SECRET_KEY = "123456";
private static boolean NO_CHECK_TIME = false;
- private final RateLimiter mLimitCounter = new RateLimiter(10, 500);
public static void setSecretKey(String secretKey) {
SECRET_KEY = secretKey;
}
@@ -44,7 +43,7 @@ abstract public class AdminAction extends Action {
@Override
public ErrorCode preAction(Request request, Response response) {
- if (!mLimitCounter.isGranted("admin")) {
+ if (!adminLimiter.isGranted("admin")) {
return ErrorCode.ERROR_CODE_OVER_FREQUENCY;
}
@@ -88,14 +87,16 @@ abstract public class AdminAction extends Action {
}
protected void sendResponse(Response response, ErrorCode errorCode, Object data) {
- response.setStatus(HttpResponseStatus.OK);
- if (errorCode == null) {
- errorCode = ErrorCode.ERROR_CODE_SUCCESS;
- }
+ if(response != null) {
+ response.setStatus(HttpResponseStatus.OK);
+ if (errorCode == null) {
+ errorCode = ErrorCode.ERROR_CODE_SUCCESS;
+ }
- RestResult result = RestResult.resultOf(errorCode, errorCode.getMsg(), data);
- response.setContent(new Gson().toJson(result));
- response.send();
+ RestResult result = RestResult.resultOf(errorCode, errorCode.getMsg(), data);
+ response.setContent(new Gson().toJson(result));
+ response.send();
+ }
}
protected void sendApiMessage(Response response, String fromUser, String topic, byte[] message, ApiCallback callback) {
sendApiMessage(response, fromUser, null, topic, message, callback, false);
diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlockUserAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlockUserAction.java
index c8ff5ee0..0d4d74cf 100755
--- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlockUserAction.java
+++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlockUserAction.java
@@ -45,7 +45,7 @@ public class BlockUserAction extends AdminAction {
response.setContent(new Gson().toJson(result));
if (inputUserBlock.getStatus() == 2) {
- sendApiMessage(response, null, ServerAPIHelper.KICKOFF_USER_REQUEST, inputUserBlock.getUserId().getBytes(), null);
+ sendApiMessage(null, null, ServerAPIHelper.KICKOFF_USER_REQUEST, inputUserBlock.getUserId().getBytes(), null);
}
sendResponse(response, ErrorCode.ERROR_CODE_SUCCESS, null);
diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ChannelAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ChannelAction.java
index 4a68a83e..ccf9055f 100644
--- a/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ChannelAction.java
+++ b/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ChannelAction.java
@@ -31,7 +31,6 @@ import java.util.concurrent.Executor;
abstract public class ChannelAction extends Action {
private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(ChannelAction.class);
- private final RateLimiter mLimitCounter = new RateLimiter(10, 200);
protected WFCMessage.ChannelInfo channelInfo;
protected interface Callback {
@@ -61,7 +60,7 @@ abstract public class ChannelAction extends Action {
return ErrorCode.ERROR_CODE_API_NOT_SIGNED;
}
- if (!mLimitCounter.isGranted(cid)) {
+ if (!channelLimiter.isGranted(cid)) {
return ErrorCode.ERROR_CODE_OVER_FREQUENCY;
}
diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/RobotAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/RobotAction.java
index 22b1968b..9eeff045 100644
--- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/RobotAction.java
+++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/RobotAction.java
@@ -33,7 +33,6 @@ import java.util.concurrent.Executor;
abstract public class RobotAction extends Action {
protected static final Logger LOG = LoggerFactory.getLogger(RobotAction.class);
- private final RateLimiter mLimitCounter = new RateLimiter(10, 1000);
protected WFCMessage.Robot robot;
@@ -64,7 +63,7 @@ abstract public class RobotAction extends Action {
return ErrorCode.ERROR_CODE_API_NOT_SIGNED;
}
- if (!mLimitCounter.isGranted(rid)) {
+ if (!robotLimiter.isGranted(rid)) {
return ErrorCode.ERROR_CODE_OVER_FREQUENCY;
}
diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java
index 4255160c..6e1a962d 100755
--- a/broker/src/main/java/io/moquette/BrokerConstants.java
+++ b/broker/src/main/java/io/moquette/BrokerConstants.java
@@ -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() {
}
diff --git a/broker/src/main/java/io/moquette/imhandler/IMHandler.java b/broker/src/main/java/io/moquette/imhandler/IMHandler.java
index 8a0734d0..4e4385b6 100755
--- a/broker/src/main/java/io/moquette/imhandler/IMHandler.java
+++ b/broker/src/main/java/io/moquette/imhandler/IMHandler.java
@@ -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;
/**
* 请求处理接口
@@ -52,7 +53,7 @@ abstract public class IMHandler {
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 {
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 {
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 {
}
};
- 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);
diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java
index e0a76deb..f2462612 100755
--- a/broker/src/main/java/io/moquette/server/Server.java
+++ b/broker/src/main/java/io/moquette/server/Server.java
@@ -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);
diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
index 2ee2def6..c0fae6ba 100755
--- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
@@ -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 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();
diff --git a/distribution/src/main/resources/wildfirechat.conf b/distribution/src/main/resources/wildfirechat.conf
index 2079ed15..4275356a 100755
--- a/distribution/src/main/resources/wildfirechat.conf
+++ b/distribution/src/main/resources/wildfirechat.conf
@@ -150,6 +150,20 @@ group.visible_quit_or_kickoff_notification 0
# 因此需要分段多次同步，此开关定义了每次同步的数据的条数。此需要客户端支持，客户端SDK在2021.5.25日后支持此功能。如果确认所有客户端都在这个日志之后才可以打开此设置。
#sync.data_part_size 1000
+#*********************************************************************
+# 限频配置
+#*********************************************************************
+##API接口限频设置，时间是10秒内允许的请求次数。
+##管理api频率限制，默认是10秒10000次。
+http.admin.rate_limit 10000
+##机器人api频率限制，默认是10秒1000次，按照机器人ID区分。
+http.robot.rate_limit 1000
+##频道api频率限制，默认是10秒1000次，按照频道ID区分。
+http.channel.rate_limit 1000
+
+##客户端请求频率限制，默认是5秒100次。通常情况下不要修改这个值。
+client.request_rate_limit 100
+
#*********************************************************************
# hazelcase configuration
#*********************************************************************
diff --git a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java
index a69dd74a..de93efee 100644
--- a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java
+++ b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java
@@ -292,7 +292,7 @@ public class Main {
}
IMResult result = RelationAdmin.sendFriendRequest("ff1", "ff2", "hello", true);
- if (result != null && result.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) {
+ if (result != null && (result.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS || result.getErrorCode() == ErrorCode.ERROR_CODE_ALREADY_FRIENDS)) {
System.out.println("send friend request success");
} else {
System.out.println("failure");
@@ -308,7 +308,7 @@ public class Main {
}
result = RelationAdmin.sendFriendRequest("ff1", "ff2", "hello3", true);
- if (result != null && result.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) {
+ if (result != null && (result.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS || result.getErrorCode() == ErrorCode.ERROR_CODE_ALREADY_FRIENDS)) {
System.out.println("success");
} else {
System.out.println("send friend request success");