From 9fdc6c2e1b0ee0a014ce525ad55bf466b16a8598 Mon Sep 17 00:00:00 2001 From: heavyrain lee Date: Fri, 26 Apr 2019 08:01:48 +0800 Subject: [PATCH] Add admin flag --- .../main/java/com/xiaoleilu/loServer/action/IMAction.java | 2 +- .../java/com/xiaoleilu/loServer/action/RouteAction.java | 2 +- .../xiaoleilu/loServer/action/admin/BlockUserAction.java | 2 +- .../loServer/action/admin/CheckUserOnlineAction.java | 2 +- .../loServer/action/admin/CreateGroupAction.java | 2 +- .../xiaoleilu/loServer/action/admin/GetIMTokenAction.java | 2 +- .../loServer/action/admin/RecallMessageAction.java | 2 +- .../loServer/action/admin/SendMessageAction.java | 2 +- .../loServer/action/channel/SendMessageAction.java | 2 +- .../loServer/action/robot/SendMessageAction.java | 2 +- .../main/java/io/moquette/imhandler/AddFriendHandler.java | 2 +- .../java/io/moquette/imhandler/ChannelListenMember.java | 2 +- .../java/io/moquette/imhandler/CreateChannelHandler.java | 2 +- .../java/io/moquette/imhandler/DistoryChannelHandler.java | 2 +- .../src/main/java/io/moquette/persistence/RPCCenter.java | 4 ++-- broker/src/main/java/io/moquette/server/Server.java | 4 ++-- .../main/java/io/moquette/spi/impl/ProtocolProcessor.java | 4 ++-- .../java/io/moquette/spi/impl/Qos1PublishHandler.java | 8 ++++---- 18 files changed, 24 insertions(+), 24 deletions(-) diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/IMAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/IMAction.java index a227f786..8b834174 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/IMAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/IMAction.java @@ -102,7 +102,7 @@ public class IMAction extends Action { ctx.executor().execute(command); }; } - }); + }, false); } } catch (InvalidProtocolBufferException e) { sendResponse(response, ErrorCode.ERROR_CODE_INVALID_DATA, null); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java index b42b4eb7..a1aad627 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java @@ -132,7 +132,7 @@ public class RouteAction extends Action { ctx.executor().execute(command); }; } - }); + }, false); } } catch (InvalidProtocolBufferException e) { sendResponse(response, ErrorCode.ERROR_CODE_INVALID_DATA, null, 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 10b86900..b16eefa0 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 @@ -44,7 +44,7 @@ public class BlockUserAction extends AdminAction { result = RestResult.resultOf(errorCode); response.setContent(new Gson().toJson(result)); - RPCCenter.getInstance().sendRequest(null, null, RPCCenter.KICKOFF_USER_REQUEST, inputUserBlock.getUserId().getBytes(), inputUserBlock.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, null); + RPCCenter.getInstance().sendRequest(null, null, RPCCenter.KICKOFF_USER_REQUEST, inputUserBlock.getUserId().getBytes(), inputUserBlock.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, null, true); sendResponse(response, ErrorCode.ERROR_CODE_SUCCESS, null); } else { diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CheckUserOnlineAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CheckUserOnlineAction.java index d4a58cc1..b08ea810 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CheckUserOnlineAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CheckUserOnlineAction.java @@ -34,7 +34,7 @@ public class CheckUserOnlineAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { InputGetUserInfo inputUserId = getRequestBody(request.getNettyRequest(), InputGetUserInfo.class); if (inputUserId == null || !StringUtil.isNullOrEmpty(inputUserId.getUserId())) { - RPCCenter.getInstance().sendRequest(null, null, RPCCenter.CHECK_USER_ONLINE_REQUEST, inputUserId.getUserId().getBytes(), inputUserId.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, null); + RPCCenter.getInstance().sendRequest(null, null, RPCCenter.CHECK_USER_ONLINE_REQUEST, inputUserId.getUserId().getBytes(), inputUserId.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, null, true); return; } else { sendResponse(response, ErrorCode.INVALID_PARAMETER, null); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateGroupAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateGroupAction.java index 87e86e07..8bd713a2 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateGroupAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateGroupAction.java @@ -74,7 +74,7 @@ public class CreateGroupAction extends AdminAction { ctx.executor().execute(command); }; } - }); + }, true); } else { response.setStatus(HttpResponseStatus.OK); RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetIMTokenAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetIMTokenAction.java index 060cc63c..95df5b71 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetIMTokenAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetIMTokenAction.java @@ -75,7 +75,7 @@ public class GetIMTokenAction extends AdminAction { ctx.executor().execute(command); }; } - }); + }, true); } } } diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMessageAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMessageAction.java index 73713e76..a7a038d8 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMessageAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMessageAction.java @@ -69,7 +69,7 @@ public class RecallMessageAction extends AdminAction { ctx.executor().execute(command); }; } - }); + }, true); } else { response.setStatus(HttpResponseStatus.OK); RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SendMessageAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SendMessageAction.java index 327be125..afc32a84 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SendMessageAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SendMessageAction.java @@ -73,7 +73,7 @@ public class SendMessageAction extends AdminAction { ctx.executor().execute(command); }; } - }); + }, true); } else { response.setStatus(HttpResponseStatus.OK); RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/channel/SendMessageAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/channel/SendMessageAction.java index d9c04b10..97c1a51a 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/channel/SendMessageAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/channel/SendMessageAction.java @@ -81,7 +81,7 @@ public class SendMessageAction extends ChannelAction { ctx.executor().execute(command); }; } - }); + }, false); } else { response.setStatus(HttpResponseStatus.OK); RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/SendMessageAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/SendMessageAction.java index 6479db79..f1c2f65b 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/SendMessageAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/SendMessageAction.java @@ -76,7 +76,7 @@ public class SendMessageAction extends RobotAction { ctx.executor().execute(command); }; } - }); + }, false); } else { response.setStatus(HttpResponseStatus.OK); RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); diff --git a/broker/src/main/java/io/moquette/imhandler/AddFriendHandler.java b/broker/src/main/java/io/moquette/imhandler/AddFriendHandler.java index 250bbd03..aa5a418a 100644 --- a/broker/src/main/java/io/moquette/imhandler/AddFriendHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/AddFriendHandler.java @@ -30,7 +30,7 @@ public class AddFriendHandler extends GroupHandler publisher.publishNotification(IMTopic.NotifyFriendRequestTopic, request.getTargetUid(), head[0]); } 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); + mServer.internalRpcMsg(request.getTargetUid(), null, handleFriendRequest.toByteArray(), 0, fromUser, HandleFriendRequestTopic, false); } } return errorCode; diff --git a/broker/src/main/java/io/moquette/imhandler/ChannelListenMember.java b/broker/src/main/java/io/moquette/imhandler/ChannelListenMember.java index 332d96d1..d3303afa 100644 --- a/broker/src/main/java/io/moquette/imhandler/ChannelListenMember.java +++ b/broker/src/main/java/io/moquette/imhandler/ChannelListenMember.java @@ -27,7 +27,7 @@ public class ChannelListenMember extends IMHandler { 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); + mServer.internalRpcMsg(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false); publisher.notifyChannelListenStatusChanged(m_messagesStore.getChannelInfo(request.getChannelId()), fromUser, request.getListen() > 0); } diff --git a/broker/src/main/java/io/moquette/imhandler/CreateChannelHandler.java b/broker/src/main/java/io/moquette/imhandler/CreateChannelHandler.java index bd4da74e..ca502973 100644 --- a/broker/src/main/java/io/moquette/imhandler/CreateChannelHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/CreateChannelHandler.java @@ -44,7 +44,7 @@ public class CreateChannelHandler extends GroupHandler { 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); + mServer.internalRpcMsg(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false); byte[] data = request.getTargetId().getBytes(); ackPayload.ensureWritable(data.length).writeBytes(data); return ErrorCode.ERROR_CODE_SUCCESS; diff --git a/broker/src/main/java/io/moquette/imhandler/DistoryChannelHandler.java b/broker/src/main/java/io/moquette/imhandler/DistoryChannelHandler.java index 1dfb39fc..15b7c7cd 100644 --- a/broker/src/main/java/io/moquette/imhandler/DistoryChannelHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/DistoryChannelHandler.java @@ -24,7 +24,7 @@ public class DistoryChannelHandler extends GroupHandler { 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); + mServer.internalRpcMsg(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false); } return errorCode; } diff --git a/broker/src/main/java/io/moquette/persistence/RPCCenter.java b/broker/src/main/java/io/moquette/persistence/RPCCenter.java index 8117343c..620059db 100644 --- a/broker/src/main/java/io/moquette/persistence/RPCCenter.java +++ b/broker/src/main/java/io/moquette/persistence/RPCCenter.java @@ -48,7 +48,7 @@ public class RPCCenter { protected RPCCenter() {} - public void sendRequest(String fromUser, String clientId, String request, byte[] message, String target, TargetEntry.Type type, Callback callback) { + public void sendRequest(String fromUser, String clientId, String request, byte[] message, String target, TargetEntry.Type type, Callback callback, boolean isAdmin) { int requestId = 0; if (callback != null) { @@ -61,7 +61,7 @@ public class RPCCenter { requestMap.put(requestId, new RequestInfo(fromUser, clientId, callback, message, requestId, request)); } - server.internalRpcMsg(fromUser, clientId, message, requestId, "", request); + server.internalRpcMsg(fromUser, clientId, message, requestId, "", request, isAdmin); } public void sendResponse(int errorCode, byte[] message, String toUuid, int requestId) { diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index d1cd171f..bb06e870 100755 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -389,14 +389,14 @@ public class Server { return hazelcastInstance; } - public void internalRpcMsg(String fromUser, String clientId, byte[] message, int messageId, String from, String request) { + 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); + m_processor.onRpcMsg(fromUser, clientId, message, messageId, from, request, isAdmin); } public void stopServer() { diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index f6f6e7d2..9e0e6049 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -608,12 +608,12 @@ public class ProtocolProcessor { return m_sessionsStore; } - public void onRpcMsg(String fromUser, String clientId, byte[] message, int messageId, String from, String request) { + public void onRpcMsg(String fromUser, String clientId, byte[] message, int messageId, String from, String request, boolean isAdmin) { if(request.equals(RPCCenter.KICKOFF_USER_REQUEST)) { mServer.getImBusinessScheduler().execute(()->handleTargetRemovedFromCurrentNode(new TargetEntry(TargetEntry.Type.TARGET_TYPE_USER, from))); return; } - qos1PublishHandler.onRpcMsg(fromUser, clientId, message, messageId, from, request); + qos1PublishHandler.onRpcMsg(fromUser, clientId, message, messageId, from, request, isAdmin); } public void shutdown() { messagesPublisher.stopChatroomScheduler(); 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 e744602b..7f5e77aa 100755 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -105,7 +105,7 @@ public class Qos1PublishHandler extends QosPublishHandler { } } - public void onRpcMsg(String fromUser, String clientId, byte[] message, int requestId, String from, String request) { + 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)); } else { @@ -116,7 +116,7 @@ public class Qos1PublishHandler extends QosPublishHandler { ReferenceCountUtil.release(ackPayload); RPCCenter.getInstance().sendResponse(ERROR_CODE_SUCCESS.getCode(), response, from, requestId); } - }); + }, isAdmin); } } @@ -145,7 +145,7 @@ public class Qos1PublishHandler extends QosPublishHandler { }); } - void imHandler(String clientID, String fromUser, String topic, byte[] payloadContent, IMCallback callback) { + void imHandler(String clientID, String fromUser, String topic, byte[] payloadContent, IMCallback callback, boolean isAdmin) { LOG.info("imHandler fromUser={}, topic={}", fromUser, topic); if(!mLimitCounter.isGranted(clientID + fromUser + topic)) { ByteBuf ackPayload = Unpooled.buffer(); @@ -237,7 +237,7 @@ public class Qos1PublishHandler extends QosPublishHandler { byte[] payloadContent = readBytesAndRewind(payload); 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)); + imHandler(clientID, username, imtopic, payloadContent, (errorCode, ackPayload) -> sendPubAck(clientID, messageID, ackPayload, errorCode), false); } private void sendPubAck(String clientId, int messageID, ByteBuf payload, ErrorCode errorCode) {