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

Add admin flag

Browse Source
This commit is contained in:
heavyrain lee
2019-04-26 08:02:14 +08:00
parent 1d309816ba
commit 9fdc6c2e1b
18 changed files with 24 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/com/xiaoleilu/loServer/action/IMAction.java
+1 -1
View File
@@ -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);
broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java
+1 -1
View File
@@ -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);
broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlockUserAction.java
+1 -1
View File
@@ -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 {
broker/src/main/java/com/xiaoleilu/loServer/action/admin/CheckUserOnlineAction.java
+1 -1
View File
@@ -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);
broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateGroupAction.java
+1 -1
View File
@@ -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);
broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetIMTokenAction.java
+1 -1
View File
@@ -75,7 +75,7 @@ public class GetIMTokenAction extends AdminAction {
ctx.executor().execute(command);
};
}
});
}, true);
}
}
}
broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMessageAction.java
+1 -1
View File
@@ -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);
broker/src/main/java/com/xiaoleilu/loServer/action/admin/SendMessageAction.java
+1 -1
View File
@@ -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);
broker/src/main/java/com/xiaoleilu/loServer/action/channel/SendMessageAction.java
+1 -1
View File
@@ -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);
broker/src/main/java/com/xiaoleilu/loServer/action/robot/SendMessageAction.java
+1 -1
View File
@@ -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);
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]);
} 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;
broker/src/main/java/io/moquette/imhandler/ChannelListenMember.java
+1 -1
View File
@@ -27,7 +27,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);
mServer.internalRpcMsg(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, 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
@@ -44,7 +44,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);
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;
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);
mServer.internalRpcMsg(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false);
}
return errorCode;
}
broker/src/main/java/io/moquette/persistence/RPCCenter.java
+2 -2
View File
@@ -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) {
broker/src/main/java/io/moquette/server/Server.java
+2 -2
View File
@@ -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() {
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+2 -2
View File
@@ -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();
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+4 -4
View File
@@ -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) {