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

1，robot和channel的api请求要与客户端请求分开限频

Browse Source
This commit is contained in:
imhao183
2021-04-23 08:41:14 +08:00
parent 7335fa4c2a
commit b4fe064807
50 changed files with 271 additions and 548 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/imhandler/AddFriendHandler.java
+1 -1
View File
@@ -30,7 +30,7 @@ public class AddFriendHandler extends GroupHandler<WFCMessage.AddFriendRequest>
publisher.publishNotification(IMTopic.NotifyFriendRequestTopic, request.getTargetUid(), head[0], fromUser, request.getReason());
} else if(user != null && user.getType() == ProtoConstants.UserType.UserType_Robot) {
WFCMessage.HandleFriendRequest handleFriendRequest = WFCMessage.HandleFriendRequest.newBuilder().setTargetUid(fromUser).setStatus(ProtoConstants.FriendRequestStatus.RequestStatus_Accepted).build();
mServer.internalRpcMsg(request.getTargetUid(), null, handleFriendRequest.toByteArray(), 0, fromUser, HandleFriendRequestTopic, false);
mServer.onApiMessage(request.getTargetUid(), null, handleFriendRequest.toByteArray(), 0, fromUser, HandleFriendRequestTopic, false, false);
}
}
return errorCode;
broker/src/main/java/io/moquette/imhandler/ChannelListenMember.java
+1 -1
View File
@@ -25,7 +25,7 @@ public class ChannelListenMember extends IMHandler<WFCMessage.ListenChannel> {
ErrorCode errorCode = m_messagesStore.listenChannel(fromUser, request.getChannelId(), request.getListen()>0);
if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) {
WFCMessage.ModifyUserSettingReq modifyUserSettingReq = WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingListenedChannels).setKey(request.getChannelId()).setValue(request.getListen() > 0 ? "1" : "0").build();
mServer.internalRpcMsg(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false);
mServer.onApiMessage(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false, false);
publisher.notifyChannelListenStatusChanged(m_messagesStore.getChannelInfo(request.getChannelId()), fromUser, request.getListen() > 0);
}
broker/src/main/java/io/moquette/imhandler/CreateChannelHandler.java
+1 -1
View File
@@ -40,7 +40,7 @@ public class CreateChannelHandler extends GroupHandler<WFCMessage.ChannelInfo> {
if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) {
WFCMessage.ModifyUserSettingReq modifyUserSettingReq = WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingMyChannels).setKey(request.getTargetId()).setValue("1").build();
mServer.internalRpcMsg(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false);
mServer.onApiMessage(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false, false);
byte[] data = request.getTargetId().getBytes();
ackPayload.ensureWritable(data.length).writeBytes(data);
return ErrorCode.ERROR_CODE_SUCCESS;
broker/src/main/java/io/moquette/imhandler/DistoryChannelHandler.java
+1 -1
View File
@@ -24,7 +24,7 @@ public class DistoryChannelHandler extends GroupHandler<WFCMessage.IDBuf> {
ErrorCode errorCode = m_messagesStore.distoryChannel(fromUser, request.getId());
if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) {
WFCMessage.ModifyUserSettingReq modifyUserSettingReq = WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingMyChannels).setKey(request.getId()).setValue("0").build();
mServer.internalRpcMsg(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false);
mServer.onApiMessage(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false, false);
}
return errorCode;
}
broker/src/main/java/io/moquette/persistence/RPCCenter.java → broker/src/main/java/io/moquette/persistence/ServerAPIHelper.java
+13 -20
View File
@@ -16,18 +16,18 @@ import cn.wildfirechat.common.ErrorCode;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class RPCCenter {
private static final Logger LOG = LoggerFactory.getLogger(RPCCenter.class);
public class ServerAPIHelper {
private static final Logger LOG = LoggerFactory.getLogger(ServerAPIHelper.class);
public static final String CHECK_USER_ONLINE_REQUEST = "check_user_online";
public static final String KICKOFF_USER_REQUEST = "kickoff_user";
private Server server;
public ConcurrentHashMap<Integer, RequestInfo> requestMap = new ConcurrentHashMap<>();
private AtomicInteger aiRequestId = new AtomicInteger(1);
public ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
private static Server server;
public static ConcurrentHashMap<Integer, RequestInfo> requestMap = new ConcurrentHashMap<>();
private static AtomicInteger aiRequestId = new AtomicInteger(1);
public static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
public void init(Server server) {
this.server = server;
public static void init(Server s) {
server = s;
}
public interface Callback {
@@ -37,18 +37,11 @@ public class RPCCenter {
Executor getResponseExecutor();
}
private static RPCCenter instance;
public static RPCCenter getInstance() {
if (instance == null) {
instance = new RPCCenter();
}
return instance;
public static void sendRequest(String fromUser, String clientId, String request, byte[] message, String target, TargetEntry.Type type, Callback callback, boolean isAdmin) {
sendRequest(fromUser, clientId, request, message, target, type, callback, isAdmin, false);
}
protected RPCCenter() {}
public void sendRequest(String fromUser, String clientId, String request, byte[] message, String target, TargetEntry.Type type, Callback callback, boolean isAdmin) {
public static void sendRequest(String fromUser, String clientId, String request, byte[] message, String target, TargetEntry.Type type, Callback callback, boolean isAdmin, boolean isRobotOrChannel) {
int requestId = 0;
if (callback != null) {
@@ -61,10 +54,10 @@ public class RPCCenter {
requestMap.put(requestId, new RequestInfo(fromUser, clientId, callback, message, requestId, request));
}
server.internalRpcMsg(fromUser, clientId, message, requestId, "", request, isAdmin);
server.onApiMessage(fromUser, clientId, message, requestId, "", request, isAdmin, isRobotOrChannel);
}
public void sendResponse(int errorCode, byte[] message, String toUuid, int requestId) {
public static void sendResponse(int errorCode, byte[] message, String toUuid, int requestId) {
LOG.debug("send async reponse to {} with requestId {}", toUuid, requestId);
if (requestId > 0) {
RequestInfo info = requestMap.remove(requestId);
broker/src/main/java/io/moquette/persistence/remote/RequestInfo.java
+5 -5
View File
@@ -8,7 +8,7 @@
package io.moquette.persistence.remote;
import io.moquette.persistence.RPCCenter;
import io.moquette.persistence.ServerAPIHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -17,7 +17,7 @@ import java.util.concurrent.TimeUnit;
public class RequestInfo {
private static final Logger LOG = LoggerFactory.getLogger(RequestInfo.class);
public final RPCCenter.Callback callback;
public final ServerAPIHelper.Callback callback;
final byte[] message;
final int requestId;
final String fromUser;
@@ -25,15 +25,15 @@ public class RequestInfo {
final String clientId;
public final ScheduledFuture future;
public RequestInfo(String fromUser, String clientId, RPCCenter.Callback callback, byte[] message, int requestId, String request) {
public RequestInfo(String fromUser, String clientId, ServerAPIHelper.Callback callback, byte[] message, int requestId, String request) {
this.callback = callback;
this.message = message;
this.requestId = requestId;
this.fromUser = fromUser;
this.clientId = clientId;
this.request = request;
this.future = RPCCenter.getInstance().scheduledExecutorService.schedule(() -> {
RequestInfo info = RPCCenter.getInstance().requestMap.remove(requestId);
this.future = ServerAPIHelper.scheduledExecutorService.schedule(() -> {
RequestInfo info = ServerAPIHelper.requestMap.remove(requestId);
if (info != null) {
LOG.error("Request timeout. fromUser {}, cliendId {}, requestId {}, request {}", fromUser, clientId, requestId, request);
info.callback.getResponseExecutor().execute(() -> {
broker/src/main/java/io/moquette/server/Server.java
+4 -10
View File
@@ -35,7 +35,6 @@ import io.moquette.spi.IStore;
import io.moquette.spi.impl.ProtocolProcessor;
import io.moquette.spi.impl.ProtocolProcessorBootstrapper;
import io.moquette.spi.impl.security.AES;
import io.moquette.spi.impl.subscriptions.Token;
import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
import io.moquette.spi.security.ISslContextCreator;
@@ -410,7 +409,7 @@ public class Server {
e.printStackTrace();
}
}
RPCCenter.getInstance().init(this);
ServerAPIHelper.init(this);
return true;
}
@@ -418,14 +417,9 @@ public class Server {
return hazelcastInstance;
}
public void internalRpcMsg(String fromUser, String clientId, byte[] message, int messageId, String from, String request, boolean isAdmin) {
if (!m_initialized) {
LOG.error("Moquette is not started, internal message cannot be notify");
return;
}
LOG.debug("internalNotifyMsg");
m_processor.onRpcMsg(fromUser, clientId, message, messageId, from, request, isAdmin);
public void onApiMessage(String fromUser, String clientId, byte[] message, int messageId, String from, String request, boolean isAdmin, boolean isRobotOrChannel) {
LOG.debug("onApiMessage");
m_processor.onApiMessage(fromUser, clientId, message, messageId, from, request, isAdmin, isRobotOrChannel);
}
public boolean isShutdowning() {
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+4 -5
View File
@@ -21,10 +21,9 @@ import cn.wildfirechat.pojos.UserOnlineStatus;
import cn.wildfirechat.proto.ProtoConstants;
import cn.wildfirechat.proto.WFCMessage;
import com.google.gson.Gson;
import com.hazelcast.core.Member;
import com.hazelcast.util.StringUtil;
import io.moquette.BrokerConstants;
import io.moquette.persistence.RPCCenter;
import io.moquette.persistence.ServerAPIHelper;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
import io.moquette.persistence.MemorySessionStore;
@@ -702,13 +701,13 @@ public class ProtocolProcessor {
return m_sessionsStore;
}
public void onRpcMsg(String fromUser, String clientId, byte[] message, int messageId, String from, String request, boolean isAdmin) {
if(request.equals(RPCCenter.KICKOFF_USER_REQUEST)) {
public void onApiMessage(String fromUser, String clientId, byte[] message, int messageId, String from, String request, boolean isAdmin, boolean isRobotOrChannel) {
if(request.equals(ServerAPIHelper.KICKOFF_USER_REQUEST)) {
String userId = new String(message);
mServer.getImBusinessScheduler().execute(()->handleTargetRemovedFromCurrentNode(new TargetEntry(TargetEntry.Type.TARGET_TYPE_USER, userId)));
return;
}
qos1PublishHandler.onRpcMsg(fromUser, clientId, message, messageId, from, request, isAdmin);
qos1PublishHandler.onApiMessage(fromUser, clientId, message, messageId, from, request, isAdmin, isRobotOrChannel);
}
public void shutdown() {
messagesPublisher.stopChatroomScheduler();
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+8 -10
View File
@@ -28,16 +28,14 @@ import java.util.zip.GZIPOutputStream;
import cn.wildfirechat.server.ThreadPoolExecutorWrapper;
import com.google.gson.Gson;
import com.xiaoleilu.loServer.RestResult;
import com.xiaoleilu.loServer.action.ClassUtil;
import cn.wildfirechat.pojos.OutputCheckUserOnline;
import io.moquette.persistence.RPCCenter;
import io.moquette.persistence.ServerAPIHelper;
import io.moquette.imhandler.Handler;
import io.moquette.imhandler.IMHandler;
import io.moquette.persistence.MemorySessionStore;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.Server;
import io.moquette.spi.ClientSession;
import io.moquette.spi.impl.security.AES;
import io.netty.handler.codec.mqtt.MqttVersion;
import io.netty.util.ReferenceCountUtil;
@@ -106,18 +104,18 @@ public class Qos1PublishHandler extends QosPublishHandler {
}
}
public void onRpcMsg(String fromUser, String clientId, byte[] message, int requestId, String from, String request, boolean isAdmin) {
if (request.equals(RPCCenter.CHECK_USER_ONLINE_REQUEST)) {
checkUserOnlineHandler(message, ackPayload -> RPCCenter.getInstance().sendResponse(ERROR_CODE_SUCCESS.getCode(), ackPayload, from, requestId));
public void onApiMessage(String fromUser, String clientId, byte[] message, int requestId, String from, String request, boolean isAdmin, boolean isRobotOrChannel) {
if (request.equals(ServerAPIHelper.CHECK_USER_ONLINE_REQUEST)) {
checkUserOnlineHandler(message, ackPayload -> ServerAPIHelper.sendResponse(ERROR_CODE_SUCCESS.getCode(), ackPayload, from, requestId));
} else {
imHandler(clientId, fromUser, request, message, (errorCode, ackPayload) -> {
if (requestId > 0) {
byte[] response = new byte[ackPayload.readableBytes()];
ackPayload.readBytes(response);
ReferenceCountUtil.release(ackPayload);
RPCCenter.getInstance().sendResponse(errorCode.getCode(), response, from, requestId);
ServerAPIHelper.sendResponse(errorCode.getCode(), response, from, requestId);
}
}, isAdmin);
}, isAdmin, isRobotOrChannel);
}
}
@@ -148,7 +146,7 @@ public class Qos1PublishHandler extends QosPublishHandler {
});
}
void imHandler(String clientID, String fromUser, String topic, byte[] payloadContent, IMCallback callback, boolean isAdmin) {
void imHandler(String clientID, String fromUser, String topic, byte[] payloadContent, IMCallback callback, boolean isAdmin, boolean isRobotOrChannel) {
LOG.info("imHandler fromUser={}, topic={}", fromUser, topic);
if(!mLimitCounter.isGranted(clientID + fromUser + topic)) {
ByteBuf ackPayload = Unpooled.buffer();
@@ -263,7 +261,7 @@ public class Qos1PublishHandler extends QosPublishHandler {
MemorySessionStore.Session session = m_sessionStore.getSession(clientID);
payloadContent = AES.AESDecrypt(payloadContent, session.getSecret(), true);
imHandler(clientID, username, imtopic, payloadContent, (errorCode, ackPayload) -> sendPubAck(clientID, messageID, ackPayload, errorCode), false);
imHandler(clientID, username, imtopic, payloadContent, (errorCode, ackPayload) -> sendPubAck(clientID, messageID, ackPayload, errorCode), false, false);
}
private void sendPubAck(String clientId, int messageID, ByteBuf payload, ErrorCode errorCode) {