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 0bac7f3f..7bfc1e5e 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/Action.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/Action.java @@ -38,6 +38,7 @@ abstract public class Action { public static ISessionsStore sessionsStore = null; public ChannelHandlerContext ctx; + public Response response; public ErrorCode preAction(Request request, Response response) { if (getClass().getAnnotation(RequireAuthentication.class) != null) { 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 a3e5d6d9..7e9e399a 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/IMAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/IMAction.java @@ -15,7 +15,7 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.MemorySessionStore; import io.moquette.persistence.TargetEntry; import io.moquette.spi.impl.Utils; @@ -80,7 +80,7 @@ public class IMAction extends Action { if (userId == null) { sendResponse(response, ErrorCode.ERROR_CODE_TOKEN_ERROR, null); } else { - RPCCenter.getInstance().sendRequest(userId, wrapper.getClientId(), wrapper.getRequest(), wrapper.getData().toByteArray(), userId, TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(userId, wrapper.getClientId(), wrapper.getRequest(), wrapper.getData().toByteArray(), userId, TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { sendResponse(response, null, result); 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 82f4400a..1d490e3f 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java @@ -15,7 +15,7 @@ import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import io.moquette.persistence.MemorySessionStore; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.moquette.spi.impl.Utils; import io.moquette.spi.impl.security.AES; @@ -113,7 +113,7 @@ public class RouteAction extends Action { if (userId == null) { sendResponse(response, ErrorCode.ERROR_CODE_TOKEN_ERROR, null, base64Response); } else { - RPCCenter.getInstance().sendRequest(userId, wrapper.getClientId(), wrapper.getRequest(), wrapper.getData().toByteArray(), userId, TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(userId, wrapper.getClientId(), wrapper.getRequest(), wrapper.getData().toByteArray(), userId, TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { sendResponse(response, null, result, base64Response); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AddFriendRequestAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AddFriendRequestAction.java index 4488cf59..6c20a76f 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AddFriendRequestAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AddFriendRequestAction.java @@ -11,21 +11,17 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.InputAddFriendRequest; -import cn.wildfirechat.pojos.InputUpdateAlias; import cn.wildfirechat.proto.WFCMessage; -import com.google.gson.Gson; import com.hazelcast.util.StringUtil; -import com.xiaoleilu.loServer.RestResult; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; -import io.netty.handler.codec.http.HttpResponseStatus; import win.liyufan.im.IMTopic; import java.util.concurrent.Executor; @@ -50,7 +46,7 @@ public class AddFriendRequestAction extends AdminAction { } WFCMessage.AddFriendRequest addFriendRequest = WFCMessage.AddFriendRequest.newBuilder().setReason(input.getReason()).setTargetUid(input.getFriendUid()).build(); - RPCCenter.getInstance().sendRequest(input.getUserId(), null, IMTopic.AddFriendRequestTopic, addFriendRequest.toByteArray(), input.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(input.getUserId(), null, IMTopic.AddFriendRequestTopic, addFriendRequest.toByteArray(), input.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AddGroupMemberAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AddGroupMemberAction.java index 08db3180..65998447 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AddGroupMemberAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/AddGroupMemberAction.java @@ -17,7 +17,7 @@ import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.InputAddGroupMember; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -42,7 +42,7 @@ public class AddGroupMemberAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { InputAddGroupMember inputAddGroupMember = getRequestBody(request.getNettyRequest(), InputAddGroupMember.class); if (inputAddGroupMember.isValide()) { - RPCCenter.getInstance().sendRequest(inputAddGroupMember.getOperator(), null, IMTopic.AddGroupMemberTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), inputAddGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputAddGroupMember.getOperator(), null, IMTopic.AddGroupMemberTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), inputAddGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlacklistAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlacklistAction.java index 304a421d..0b9fbfed 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlacklistAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BlacklistAction.java @@ -18,7 +18,7 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -47,7 +47,7 @@ public class BlacklistAction extends AdminAction { && !StringUtil.isNullOrEmpty(inputData.getTargetUid()) ) { WFCMessage.BlackUserRequest friendRequest = WFCMessage.BlackUserRequest.newBuilder().setUid(inputData.getTargetUid()).setStatus(inputData.getStatus()).build(); - RPCCenter.getInstance().sendRequest(inputData.getUserId(), null, IMTopic.BlackListUserTopic, friendRequest.toByteArray(), inputData.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputData.getUserId(), null, IMTopic.BlackListUserTopic, friendRequest.toByteArray(), inputData.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); 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 792e74a6..ab3d825a 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 @@ -16,7 +16,7 @@ import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.InputOutputUserBlockStatus; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; @@ -46,7 +46,7 @@ public class BlockUserAction extends AdminAction { response.setContent(new Gson().toJson(result)); if (inputUserBlock.getStatus() == 2) { - RPCCenter.getInstance().sendRequest(null, null, RPCCenter.KICKOFF_USER_REQUEST, inputUserBlock.getUserId().getBytes(), inputUserBlock.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, null, true); + ServerAPIHelper.sendRequest(null, null, ServerAPIHelper.KICKOFF_USER_REQUEST, inputUserBlock.getUserId().getBytes(), inputUserBlock.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, null, true); } sendResponse(response, ErrorCode.ERROR_CODE_SUCCESS, null); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BroadcastMessageAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BroadcastMessageAction.java index 5bd0331c..eff2a5e9 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BroadcastMessageAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/BroadcastMessageAction.java @@ -13,8 +13,6 @@ import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.BroadMessageData; import cn.wildfirechat.pojos.BroadMessageResult; -import cn.wildfirechat.pojos.SendMessageData; -import cn.wildfirechat.pojos.SendMessageResult; import com.google.gson.Gson; import com.hazelcast.util.StringUtil; import com.xiaoleilu.loServer.RestResult; @@ -22,7 +20,7 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -46,7 +44,7 @@ public class BroadcastMessageAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { BroadMessageData sendMessageData = getRequestBody(request.getNettyRequest(), BroadMessageData.class); if (BroadMessageData.isValide(sendMessageData) && !StringUtil.isNullOrEmpty(sendMessageData.getSender())) { - RPCCenter.getInstance().sendRequest(sendMessageData.getSender(), null, IMTopic.BroadcastMessageTopic, sendMessageData.toProtoMessage().toByteArray(), sendMessageData.getSender(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(sendMessageData.getSender(), null, IMTopic.BroadcastMessageTopic, sendMessageData.toProtoMessage().toByteArray(), sendMessageData.getSender(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); 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 b0f6f8f7..27d07481 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 @@ -9,21 +9,16 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; -import cn.wildfirechat.pojos.BroadMessageResult; import cn.wildfirechat.pojos.OutputCheckUserOnline; import com.google.gson.Gson; -import com.xiaoleilu.loServer.RestResult; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.InputGetUserInfo; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; -import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.util.internal.StringUtil; import cn.wildfirechat.common.ErrorCode; @@ -43,7 +38,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, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(null, null, ServerAPIHelper.CHECK_USER_ONLINE_REQUEST, inputUserId.getUserId().getBytes(), inputUserId.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] res) { OutputCheckUserOnline out = new Gson().fromJson(new String(res), OutputCheckUserOnline.class); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateChannelAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateChannelAction.java index d902d40e..cc53b92f 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateChannelAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/CreateChannelAction.java @@ -11,14 +11,13 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.*; -import cn.wildfirechat.proto.WFCMessage; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -26,7 +25,6 @@ import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.util.internal.StringUtil; import win.liyufan.im.IMTopic; -import win.liyufan.im.UUIDGenerator; import java.util.concurrent.Executor; @@ -52,7 +50,7 @@ public class CreateChannelAction extends AdminAction { inputCreateChannel.setTargetId(messagesStore.getShortUUID()); } - RPCCenter.getInstance().sendRequest(inputCreateChannel.getOwner(), null, IMTopic.CreateChannelTopic, inputCreateChannel.toProtoChannelInfo().toByteArray(), inputCreateChannel.getOwner(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputCreateChannel.getOwner(), null, IMTopic.CreateChannelTopic, inputCreateChannel.toProtoChannelInfo().toByteArray(), inputCreateChannel.getOwner(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); 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 cdcf3233..153295e7 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 @@ -18,7 +18,7 @@ import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.InputCreateGroup; import cn.wildfirechat.pojos.OutputCreateGroupResult; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -43,7 +43,7 @@ public class CreateGroupAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { InputCreateGroup inputCreateGroup = getRequestBody(request.getNettyRequest(), InputCreateGroup.class); if (inputCreateGroup.isValide()) { - RPCCenter.getInstance().sendRequest(inputCreateGroup.getOperator(), null, IMTopic.CreateGroupTopic, inputCreateGroup.toProtoGroupRequest().toByteArray(), inputCreateGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputCreateGroup.getOperator(), null, IMTopic.CreateGroupTopic, inputCreateGroup.toProtoGroupRequest().toByteArray(), inputCreateGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/DestroyUserAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/DestroyUserAction.java index c7fd0a53..e6d3cd83 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/DestroyUserAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/DestroyUserAction.java @@ -11,9 +11,6 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.InputDestroyUser; -import cn.wildfirechat.pojos.InputOutputUserInfo; -import cn.wildfirechat.pojos.OutputCreateUser; -import cn.wildfirechat.pojos.OutputGetIMTokenData; import cn.wildfirechat.proto.WFCMessage; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; @@ -21,13 +18,12 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.util.internal.StringUtil; import win.liyufan.im.IMTopic; -import win.liyufan.im.UUIDGenerator; import java.util.Base64; import java.util.concurrent.Executor; @@ -49,7 +45,7 @@ public class DestroyUserAction extends AdminAction { && !StringUtil.isNullOrEmpty(inputDestroyUser.getUserId())) { WFCMessage.IDBuf idBuf = WFCMessage.IDBuf.newBuilder().setId(inputDestroyUser.getUserId()).build(); - RPCCenter.getInstance().sendRequest(inputDestroyUser.getUserId(), null, IMTopic.DestroyUserTopic, idBuf.toByteArray(), inputDestroyUser.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputDestroyUser.getUserId(), null, IMTopic.DestroyUserTopic, idBuf.toByteArray(), inputDestroyUser.getUserId(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ErrorCode errorCode1 = ErrorCode.fromCode(result[0]); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/DismissGroupAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/DismissGroupAction.java index 3f4cb48b..ab5ec629 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/DismissGroupAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/DismissGroupAction.java @@ -17,7 +17,7 @@ import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.InputDismissGroup; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -42,7 +42,7 @@ public class DismissGroupAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { InputDismissGroup inputDismissGroup = getRequestBody(request.getNettyRequest(), InputDismissGroup.class); if (inputDismissGroup.isValide()) { - RPCCenter.getInstance().sendRequest(inputDismissGroup.getOperator(), null, IMTopic.DismissGroupTopic, inputDismissGroup.toProtoGroupRequest().toByteArray(), inputDismissGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputDismissGroup.getOperator(), null, IMTopic.DismissGroupTopic, inputDismissGroup.toProtoGroupRequest().toByteArray(), inputDismissGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/FriendRelationAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/FriendRelationAction.java index 7ede42b6..cd2fe3f8 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/FriendRelationAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/FriendRelationAction.java @@ -17,7 +17,7 @@ import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.InputUpdateFriendStatusRequest; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -50,7 +50,7 @@ public class FriendRelationAction extends AdminAction { if (!StringUtil.isNullOrEmpty(friendAdd.getExtra())) { friendRequestBuilder = friendRequestBuilder.setExtra(friendAdd.getExtra()); } - RPCCenter.getInstance().sendRequest(friendAdd.getUserId(), null, IMTopic.HandleFriendRequestTopic, friendRequestBuilder.build().toByteArray(), friendAdd.getFriendUid(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(friendAdd.getUserId(), null, IMTopic.HandleFriendRequestTopic, friendRequestBuilder.build().toByteArray(), friendAdd.getFriendUid(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); 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 ac59a039..bf29649f 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 @@ -16,7 +16,7 @@ import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.InputGetToken; import cn.wildfirechat.pojos.OutputGetIMTokenData; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.handler.codec.http.FullHttpRequest; import cn.wildfirechat.common.ErrorCode; @@ -42,7 +42,7 @@ public class GetIMTokenAction extends AdminAction { WFCMessage.GetTokenRequest getTokenRequest = WFCMessage.GetTokenRequest.newBuilder().setUserId(userId).setClientId(input.getClientId()).setPlatform(input.getPlatform() == null ? 0 : input.getPlatform()).build(); - RPCCenter.getInstance().sendRequest(userId, input.getClientId(), IMTopic.GetTokenTopic, getTokenRequest.toByteArray(), userId, TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(userId, input.getClientId(), IMTopic.GetTokenTopic, getTokenRequest.toByteArray(), userId, TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ErrorCode errorCode1 = ErrorCode.fromCode(result[0]); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/KickoffGroupMemberAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/KickoffGroupMemberAction.java index 56f4eeeb..5390352f 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/KickoffGroupMemberAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/KickoffGroupMemberAction.java @@ -17,7 +17,7 @@ import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.InputKickoffGroupMember; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -42,7 +42,7 @@ public class KickoffGroupMemberAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { InputKickoffGroupMember inputKickoffGroupMember = getRequestBody(request.getNettyRequest(), InputKickoffGroupMember.class); if (inputKickoffGroupMember.isValide()) { - RPCCenter.getInstance().sendRequest(inputKickoffGroupMember.getOperator(), null, IMTopic.KickoffGroupMemberTopic, inputKickoffGroupMember.toProtoGroupRequest().toByteArray(), inputKickoffGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputKickoffGroupMember.getOperator(), null, IMTopic.KickoffGroupMemberTopic, inputKickoffGroupMember.toProtoGroupRequest().toByteArray(), inputKickoffGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/KickoffUserClientAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/KickoffUserClientAction.java index 46b17103..9d07eee1 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/KickoffUserClientAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/KickoffUserClientAction.java @@ -10,7 +10,6 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; -import cn.wildfirechat.pojos.InputOutputUserBlockStatus; import cn.wildfirechat.pojos.StringPairPojo; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; @@ -18,8 +17,6 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.util.internal.StringUtil; diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ModifyGroupInfoAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ModifyGroupInfoAction.java index 4cb48962..41179505 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ModifyGroupInfoAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ModifyGroupInfoAction.java @@ -11,7 +11,6 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; -import cn.wildfirechat.pojos.InputAddGroupMember; import cn.wildfirechat.pojos.InputModifyGroupInfo; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; @@ -19,7 +18,7 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -43,7 +42,7 @@ public class ModifyGroupInfoAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { InputModifyGroupInfo inputAddGroupMember = getRequestBody(request.getNettyRequest(), InputModifyGroupInfo.class); if (inputAddGroupMember.isValide()) { - RPCCenter.getInstance().sendRequest(inputAddGroupMember.getOperator(), null, IMTopic.ModifyGroupInfoTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), inputAddGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputAddGroupMember.getOperator(), null, IMTopic.ModifyGroupInfoTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), inputAddGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ModifyGroupMemberAliasAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ModifyGroupMemberAliasAction.java index 36280489..bd1919fc 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ModifyGroupMemberAliasAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ModifyGroupMemberAliasAction.java @@ -11,7 +11,6 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; -import cn.wildfirechat.pojos.InputKickoffGroupMember; import cn.wildfirechat.pojos.InputSetGroupMemberAlias; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; @@ -19,7 +18,7 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -43,7 +42,7 @@ public class ModifyGroupMemberAliasAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { InputSetGroupMemberAlias inputSetGroupMemberAlias = getRequestBody(request.getNettyRequest(), InputSetGroupMemberAlias.class); if (inputSetGroupMemberAlias.isValide()) { - RPCCenter.getInstance().sendRequest(inputSetGroupMemberAlias.getOperator(), null, IMTopic.ModifyGroupMemberAliasTopic, inputSetGroupMemberAlias.toProtoGroupRequest().toByteArray(), inputSetGroupMemberAlias.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputSetGroupMemberAlias.getOperator(), null, IMTopic.ModifyGroupMemberAliasTopic, inputSetGroupMemberAlias.toProtoGroupRequest().toByteArray(), inputSetGroupMemberAlias.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/MulticastMessageAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/MulticastMessageAction.java index 9f1b2dd0..3a0187e0 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/MulticastMessageAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/MulticastMessageAction.java @@ -11,18 +11,15 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; -import cn.wildfirechat.pojos.BroadMessageData; -import cn.wildfirechat.pojos.BroadMessageResult; import cn.wildfirechat.pojos.MultiMessageResult; import cn.wildfirechat.pojos.MulticastMessageData; import com.google.gson.Gson; -import com.hazelcast.util.StringUtil; import com.xiaoleilu.loServer.RestResult; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -46,7 +43,7 @@ public class MulticastMessageAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { MulticastMessageData sendMessageData = getRequestBody(request.getNettyRequest(), MulticastMessageData.class); if (MulticastMessageData.isValide(sendMessageData)) { - RPCCenter.getInstance().sendRequest(sendMessageData.getSender(), null, IMTopic.MultiCastMessageTopic, sendMessageData.toProtoMessage().toByteArray(), sendMessageData.getSender(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(sendMessageData.getSender(), null, IMTopic.MultiCastMessageTopic, sendMessageData.toProtoMessage().toByteArray(), sendMessageData.getSender(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/QuitGroupMemberAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/QuitGroupMemberAction.java index 873316c6..fae66ad1 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/QuitGroupMemberAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/QuitGroupMemberAction.java @@ -11,7 +11,6 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; -import cn.wildfirechat.pojos.InputKickoffGroupMember; import cn.wildfirechat.pojos.InputQuitGroup; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; @@ -19,7 +18,7 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -43,7 +42,7 @@ public class QuitGroupMemberAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { InputQuitGroup inputQuitGroup = getRequestBody(request.getNettyRequest(), InputQuitGroup.class); if (inputQuitGroup.isValide()) { - RPCCenter.getInstance().sendRequest(inputQuitGroup.getOperator(), null, IMTopic.QuitGroupTopic, inputQuitGroup.toProtoGroupRequest().toByteArray(), inputQuitGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputQuitGroup.getOperator(), null, IMTopic.QuitGroupTopic, inputQuitGroup.toProtoGroupRequest().toByteArray(), inputQuitGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); 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 4346bcb8..7455860a 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 @@ -17,7 +17,7 @@ import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.RecallMessageData; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -45,7 +45,7 @@ public class RecallMessageAction extends AdminAction { if (recallMessageData != null && !StringUtil.isNullOrEmpty(recallMessageData.getOperator())) { WFCMessage.INT64Buf idBuf = WFCMessage.INT64Buf.newBuilder().setId(recallMessageData.getMessageUid()).build(); - RPCCenter.getInstance().sendRequest(recallMessageData.getOperator(), null, IMTopic.RecallMessageTopic, idBuf.toByteArray(), recallMessageData.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(recallMessageData.getOperator(), null, IMTopic.RecallMessageTopic, idBuf.toByteArray(), recallMessageData.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMultiCastMessageAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMultiCastMessageAction.java index 620a8f7c..4763ec0c 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMultiCastMessageAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/RecallMultiCastMessageAction.java @@ -10,7 +10,6 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; -import cn.wildfirechat.pojos.RecallMessageData; import cn.wildfirechat.pojos.RecallMultiCastMessageData; import cn.wildfirechat.proto.WFCMessage; import com.google.gson.Gson; @@ -19,7 +18,7 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -46,7 +45,7 @@ public class RecallMultiCastMessageAction extends AdminAction { if (recallMessageData != null && !StringUtil.isNullOrEmpty(recallMessageData.operator)) { WFCMessage.RecallMultiCastMessageRequest recallRequest = WFCMessage.RecallMultiCastMessageRequest.newBuilder().setMessageId(recallMessageData.messageUid).addAllReceiver(recallMessageData.receivers).build(); - RPCCenter.getInstance().sendRequest(recallMessageData.operator, null, IMTopic.RecallMultiCastMessageTopic, recallRequest.toByteArray(), recallMessageData.operator, TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(recallMessageData.operator, null, IMTopic.RecallMultiCastMessageTopic, recallRequest.toByteArray(), recallMessageData.operator, TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); 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 94466da8..b5a9a0be 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 @@ -19,7 +19,7 @@ import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.SendMessageData; import cn.wildfirechat.pojos.SendMessageResult; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -44,7 +44,7 @@ public class SendMessageAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { SendMessageData sendMessageData = getRequestBody(request.getNettyRequest(), SendMessageData.class); if (SendMessageData.isValide(sendMessageData) && !StringUtil.isNullOrEmpty(sendMessageData.getSender())) { - RPCCenter.getInstance().sendRequest(sendMessageData.getSender(), null, IMTopic.SendMessageTopic, sendMessageData.toProtoMessage().toByteArray(), sendMessageData.getSender(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(sendMessageData.getSender(), null, IMTopic.SendMessageTopic, sendMessageData.toProtoMessage().toByteArray(), sendMessageData.getSender(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SetGroupManagerAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SetGroupManagerAction.java index 7a2d010b..fc3064ea 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SetGroupManagerAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SetGroupManagerAction.java @@ -11,7 +11,6 @@ package com.xiaoleilu.loServer.action.admin; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; -import cn.wildfirechat.pojos.InputAddGroupMember; import cn.wildfirechat.pojos.InputSetGroupManager; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; @@ -19,7 +18,7 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -43,7 +42,7 @@ public class SetGroupManagerAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { InputSetGroupManager inputAddGroupMember = getRequestBody(request.getNettyRequest(), InputSetGroupManager.class); if (inputAddGroupMember.isValide()) { - RPCCenter.getInstance().sendRequest(inputAddGroupMember.getOperator(), null, IMTopic.SetGroupManagerTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), inputAddGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputAddGroupMember.getOperator(), null, IMTopic.SetGroupManagerTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), inputAddGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/TransferGroupAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/TransferGroupAction.java index ad14da79..0f2534dc 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/TransferGroupAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/TransferGroupAction.java @@ -17,7 +17,7 @@ import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.InputTransferGroup; -import io.moquette.persistence.RPCCenter; +import io.moquette.persistence.ServerAPIHelper; import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -42,7 +42,7 @@ public class TransferGroupAction extends AdminAction { if (request.getNettyRequest() instanceof FullHttpRequest) { InputTransferGroup inputDismissGroup = getRequestBody(request.getNettyRequest(), InputTransferGroup.class); if (inputDismissGroup.isValide()) { - RPCCenter.getInstance().sendRequest(inputDismissGroup.getOperator(), null, IMTopic.TransferGroupTopic, inputDismissGroup.toProtoGroupRequest().toByteArray(), inputDismissGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { + ServerAPIHelper.sendRequest(inputDismissGroup.getOperator(), null, IMTopic.TransferGroupTopic, inputDismissGroup.toProtoGroupRequest().toByteArray(), inputDismissGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { @Override public void onSuccess(byte[] result) { ByteBuf byteBuf = Unpooled.buffer(); 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 978dc27d..d6b86527 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 @@ -15,6 +15,8 @@ import com.xiaoleilu.loServer.RestResult; import com.xiaoleilu.loServer.action.Action; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; +import io.moquette.persistence.ServerAPIHelper; +import io.moquette.persistence.TargetEntry; import io.moquette.spi.impl.Utils; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpRequest; @@ -26,11 +28,17 @@ import org.slf4j.LoggerFactory; import win.liyufan.im.RateLimiter; import win.liyufan.im.Utility; +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 { + void onSuccess(byte[] response); + } + @Override public ErrorCode preAction(Request request, Response response) { String nonce = request.getHeader("nonce"); @@ -85,7 +93,7 @@ abstract public class ChannelAction extends Action { return localSign.equals(sign) ? ErrorCode.ERROR_CODE_SUCCESS : ErrorCode.ERROR_CODE_AUTH_FAILURE; } - protected void sendResponse(Response response, ErrorCode errorCode, Object data) { + protected void sendResponse(ErrorCode errorCode, Object data) { response.setStatus(HttpResponseStatus.OK); if (errorCode == null) { errorCode = ErrorCode.ERROR_CODE_SUCCESS; @@ -107,4 +115,29 @@ abstract public class ChannelAction extends Action { } return null; } + + protected void sendApiMessage(String topic, byte[] message, Callback callback) { + ServerAPIHelper.sendRequest(channelInfo.getOwner(), null, topic, message, channelInfo.getOwner(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { + @Override + public void onSuccess(byte[] response) { + callback.onSuccess(response); + } + @Override + public void onError(ErrorCode errorCode) { + sendResponse(errorCode, null); + } + + @Override + public void onTimeout() { + sendResponse(ErrorCode.ERROR_CODE_TIMEOUT, null); + } + + @Override + public Executor getResponseExecutor() { + return command -> { + ctx.executor().execute(command); + }; + } + }, false, true); + } } diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ModifyChannelProfileAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ModifyChannelProfileAction.java index 6661b6f0..eebcb57a 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ModifyChannelProfileAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/channel/ModifyChannelProfileAction.java @@ -10,10 +10,7 @@ package com.xiaoleilu.loServer.action.channel; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; -import cn.wildfirechat.pojos.InputGetUserInfo; import cn.wildfirechat.pojos.InputModifyChannelInfo; -import cn.wildfirechat.pojos.InputOutputUserInfo; -import cn.wildfirechat.pojos.SendMessageResult; import cn.wildfirechat.proto.WFCMessage; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; @@ -21,17 +18,12 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; -import io.netty.util.internal.StringUtil; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Channel_Update_Profile) @HttpMethod("POST") public class ModifyChannelProfileAction extends ChannelAction { @@ -41,36 +33,16 @@ public class ModifyChannelProfileAction extends ChannelAction { InputModifyChannelInfo input = getRequestBody(request.getNettyRequest(), InputModifyChannelInfo.class); if (input != null) { WFCMessage.ModifyChannelInfo modifyChannelInfo = WFCMessage.ModifyChannelInfo.newBuilder().setChannelId(channelInfo.getTargetId()).setType(input.getType()).setValue(input.getValue()).build(); - RPCCenter.getInstance().sendRequest(channelInfo.getOwner(), null, IMTopic.ModifyChannelInfoTopic, modifyChannelInfo.toByteArray(), channelInfo.getOwner(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - sendResponse(response, null, null); - } else { - sendResponse(response, errorCode, null); - } + sendApiMessage(IMTopic.ModifyChannelInfoTopic, modifyChannelInfo.toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + sendResponse(null, null); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); 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 9940d609..86f230b5 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 @@ -20,8 +20,6 @@ import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.Conversation; import cn.wildfirechat.pojos.SendMessageData; import cn.wildfirechat.pojos.SendMessageResult; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; @@ -29,8 +27,6 @@ import io.netty.handler.codec.http.HttpResponseStatus; import cn.wildfirechat.common.ErrorCode; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - import static cn.wildfirechat.proto.ProtoConstants.ConversationType.ConversationType_Channel; @Route(APIPath.Channel_Message_Send) @@ -55,38 +51,18 @@ public class SendMessageAction extends ChannelAction { sendMessageData.setPayload(sendChannelMessageData.getPayload()); sendMessageData.setToUsers(sendChannelMessageData.getTargets()); if (SendMessageData.isValide(sendMessageData)) { - RPCCenter.getInstance().sendRequest(sendMessageData.getSender(), null, IMTopic.SendMessageTopic, sendMessageData.toProtoMessage().toByteArray(), sendMessageData.getSender(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - long messageId = byteBuf.readLong(); - long timestamp = byteBuf.readLong(); - sendResponse(response, null, new SendMessageResult(messageId, timestamp)); - } else { - sendResponse(response, errorCode, null); - } + sendApiMessage(IMTopic.SendMessageTopic, sendMessageData.toProtoMessage().toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + long messageId = byteBuf.readLong(); + long timestamp = byteBuf.readLong(); + sendResponse(null, new SendMessageResult(messageId, timestamp)); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/channel/SubscriberChannelAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/channel/SubscriberChannelAction.java index 83917351..f16c212e 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/channel/SubscriberChannelAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/channel/SubscriberChannelAction.java @@ -11,7 +11,6 @@ package com.xiaoleilu.loServer.action.channel; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.InputChannelSubscribe; -import cn.wildfirechat.pojos.InputModifyChannelInfo; import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; import com.google.gson.Gson; @@ -21,16 +20,12 @@ import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Channel_Subscribe) @HttpMethod("POST") public class SubscriberChannelAction extends ChannelAction { @@ -47,36 +42,16 @@ public class SubscriberChannelAction extends ChannelAction { return true; } WFCMessage.ListenChannel listenChannel = WFCMessage.ListenChannel.newBuilder().setChannelId(channelInfo.getTargetId()).setListen(input.getSubscribe()).build(); - RPCCenter.getInstance().sendRequest(input.getTarget(), null, IMTopic.ChannelListenTopic, listenChannel.toByteArray(), input.getTarget(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - sendResponse(response, null, null); - } else { - sendResponse(response, errorCode, null); - } + sendApiMessage(IMTopic.ChannelListenTopic, listenChannel.toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + sendResponse(null, null); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, true); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/AddGroupMemberAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/AddGroupMemberAction.java index f3137541..d8fc1d83 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/AddGroupMemberAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/AddGroupMemberAction.java @@ -14,21 +14,16 @@ import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.InputAddGroupMember; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; -import com.xiaoleilu.loServer.action.admin.AdminAction; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Robot_Group_Member_Add) @HttpMethod("POST") public class AddGroupMemberAction extends RobotAction { @@ -44,36 +39,16 @@ public class AddGroupMemberAction extends RobotAction { InputAddGroupMember inputAddGroupMember = getRequestBody(request.getNettyRequest(), InputAddGroupMember.class); inputAddGroupMember.setOperator(robot.getUid()); if (inputAddGroupMember.isValide()) { - RPCCenter.getInstance().sendRequest(inputAddGroupMember.getOperator(), null, IMTopic.AddGroupMemberTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), inputAddGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - sendResponse(response, null, null); - } else { - sendResponse(response, errorCode, null); - } + sendApiRequest(IMTopic.AddGroupMemberTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + sendResponse(null, null); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/CreateGroupAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/CreateGroupAction.java index cabd32ca..f302d671 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/CreateGroupAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/CreateGroupAction.java @@ -17,21 +17,16 @@ import cn.wildfirechat.pojos.PojoGroupInfo; import cn.wildfirechat.proto.WFCMessage; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; -import com.xiaoleilu.loServer.action.admin.AdminAction; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Robot_Create_Group) @HttpMethod("POST") public class CreateGroupAction extends RobotAction { @@ -50,39 +45,19 @@ public class CreateGroupAction extends RobotAction { if (inputCreateGroup.isValide()) { PojoGroupInfo group_info = inputCreateGroup.getGroup().getGroup_info(); WFCMessage.CreateGroupRequest createGroupRequest = inputCreateGroup.toProtoGroupRequest(); - RPCCenter.getInstance().sendRequest(inputCreateGroup.getOperator(), null, IMTopic.CreateGroupTopic, createGroupRequest.toByteArray(), inputCreateGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - byte[] data = new byte[byteBuf.readableBytes()]; - byteBuf.readBytes(data); - String groupId = new String(data); - sendResponse(response, null, new OutputCreateGroupResult(groupId)); - } else { - sendResponse(response, errorCode, null); - } + sendApiRequest(IMTopic.CreateGroupTopic, createGroupRequest.toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + byte[] data = new byte[byteBuf.readableBytes()]; + byteBuf.readBytes(data); + String groupId = new String(data); + sendResponse(null, new OutputCreateGroupResult(groupId)); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/DismissGroupAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/DismissGroupAction.java index 9e33a93f..7ff0516d 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/DismissGroupAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/DismissGroupAction.java @@ -14,21 +14,16 @@ import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.InputDismissGroup; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; -import com.xiaoleilu.loServer.action.admin.AdminAction; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Robot_Group_Dismiss) @HttpMethod("POST") public class DismissGroupAction extends RobotAction { @@ -44,36 +39,16 @@ public class DismissGroupAction extends RobotAction { InputDismissGroup inputDismissGroup = getRequestBody(request.getNettyRequest(), InputDismissGroup.class); inputDismissGroup.setOperator(robot.getUid()); if (inputDismissGroup.isValide()) { - RPCCenter.getInstance().sendRequest(inputDismissGroup.getOperator(), null, IMTopic.DismissGroupTopic, inputDismissGroup.toProtoGroupRequest().toByteArray(), inputDismissGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - sendResponse(response, null, null); - } else { - sendResponse(response, errorCode, null); - } + sendApiRequest(IMTopic.DismissGroupTopic, inputDismissGroup.toProtoGroupRequest().toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + sendResponse(null, null); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/KickoffGroupMemberAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/KickoffGroupMemberAction.java index e0061eeb..747997ce 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/KickoffGroupMemberAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/KickoffGroupMemberAction.java @@ -14,21 +14,16 @@ import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.InputKickoffGroupMember; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; -import com.xiaoleilu.loServer.action.admin.AdminAction; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Robot_Group_Member_Kickoff) @HttpMethod("POST") public class KickoffGroupMemberAction extends RobotAction { @@ -44,36 +39,16 @@ public class KickoffGroupMemberAction extends RobotAction { InputKickoffGroupMember inputKickoffGroupMember = getRequestBody(request.getNettyRequest(), InputKickoffGroupMember.class); inputKickoffGroupMember.setOperator(robot.getUid()); if (inputKickoffGroupMember.isValide()) { - RPCCenter.getInstance().sendRequest(inputKickoffGroupMember.getOperator(), null, IMTopic.KickoffGroupMemberTopic, inputKickoffGroupMember.toProtoGroupRequest().toByteArray(), inputKickoffGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - sendResponse(response, null, null); - } else { - sendResponse(response, errorCode, null); - } + sendApiRequest(IMTopic.KickoffGroupMemberTopic, inputKickoffGroupMember.toProtoGroupRequest().toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + sendResponse(null, null); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/ModifyGroupInfoAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/ModifyGroupInfoAction.java index 98fd27b6..1ca2b1ac 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/ModifyGroupInfoAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/ModifyGroupInfoAction.java @@ -14,21 +14,16 @@ import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.InputModifyGroupInfo; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; -import com.xiaoleilu.loServer.action.admin.AdminAction; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Robot_Group_Modify_Info) @HttpMethod("POST") public class ModifyGroupInfoAction extends RobotAction { @@ -44,36 +39,16 @@ public class ModifyGroupInfoAction extends RobotAction { InputModifyGroupInfo inputAddGroupMember = getRequestBody(request.getNettyRequest(), InputModifyGroupInfo.class); inputAddGroupMember.setOperator(robot.getUid()); if (inputAddGroupMember.isValide()) { - RPCCenter.getInstance().sendRequest(inputAddGroupMember.getOperator(), null, IMTopic.ModifyGroupInfoTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), inputAddGroupMember.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - sendResponse(response, null, null); - } else { - sendResponse(response, errorCode, null); - } + sendApiRequest(IMTopic.ModifyGroupInfoTopic, inputAddGroupMember.toProtoGroupRequest().toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + sendResponse(null, null); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/ModifyGroupMemberAliasAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/ModifyGroupMemberAliasAction.java index 4c71db89..83f311fc 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/ModifyGroupMemberAliasAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/ModifyGroupMemberAliasAction.java @@ -14,21 +14,16 @@ import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.InputSetGroupMemberAlias; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; -import com.xiaoleilu.loServer.action.admin.AdminAction; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Robot_Group_Set_Member_Alias) @HttpMethod("POST") public class ModifyGroupMemberAliasAction extends RobotAction { @@ -44,36 +39,16 @@ public class ModifyGroupMemberAliasAction extends RobotAction { InputSetGroupMemberAlias inputSetGroupMemberAlias = getRequestBody(request.getNettyRequest(), InputSetGroupMemberAlias.class); inputSetGroupMemberAlias.setOperator(robot.getUid()); if (inputSetGroupMemberAlias.isValide()) { - RPCCenter.getInstance().sendRequest(inputSetGroupMemberAlias.getOperator(), null, IMTopic.ModifyGroupMemberAliasTopic, inputSetGroupMemberAlias.toProtoGroupRequest().toByteArray(), inputSetGroupMemberAlias.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - sendResponse(response, null, null); - } else { - sendResponse(response, errorCode, null); - } + sendApiRequest(IMTopic.ModifyGroupMemberAliasTopic, inputSetGroupMemberAlias.toProtoGroupRequest().toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + sendResponse(null, null); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/QuitGroupMemberAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/QuitGroupMemberAction.java index f4cbd646..30b5163b 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/QuitGroupMemberAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/QuitGroupMemberAction.java @@ -14,21 +14,16 @@ import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.InputQuitGroup; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; -import com.xiaoleilu.loServer.action.admin.AdminAction; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Robot_Group_Member_Quit) @HttpMethod("POST") public class QuitGroupMemberAction extends RobotAction { @@ -44,36 +39,16 @@ public class QuitGroupMemberAction extends RobotAction { InputQuitGroup inputQuitGroup = getRequestBody(request.getNettyRequest(), InputQuitGroup.class); inputQuitGroup.setOperator(robot.getUid()); if (inputQuitGroup.isValide()) { - RPCCenter.getInstance().sendRequest(inputQuitGroup.getOperator(), null, IMTopic.QuitGroupTopic, inputQuitGroup.toProtoGroupRequest().toByteArray(), inputQuitGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - sendResponse(response, null, null); - } else { - sendResponse(response, errorCode, null); - } + sendApiRequest(IMTopic.QuitGroupTopic, inputQuitGroup.toProtoGroupRequest().toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + sendResponse(null, null); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); 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 d608e158..2da8adcc 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 @@ -15,6 +15,8 @@ import com.xiaoleilu.loServer.RestResult; import com.xiaoleilu.loServer.action.Action; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; +import io.moquette.persistence.ServerAPIHelper; +import io.moquette.persistence.TargetEntry; import io.moquette.spi.impl.Utils; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpRequest; @@ -28,6 +30,7 @@ import win.liyufan.im.RateLimiter; import win.liyufan.im.Utility; import java.io.UnsupportedEncodingException; +import java.util.concurrent.Executor; abstract public class RobotAction extends Action { protected static final Logger LOG = LoggerFactory.getLogger(RobotAction.class); @@ -35,6 +38,10 @@ abstract public class RobotAction extends Action { protected WFCMessage.Robot robot; + protected interface Callback { + void onSuccess(byte[] response); + } + @Override public ErrorCode preAction(Request request, Response response) { String nonce = request.getHeader("nonce"); @@ -89,7 +96,7 @@ abstract public class RobotAction extends Action { return localSign.equals(sign) ? ErrorCode.ERROR_CODE_SUCCESS : ErrorCode.ERROR_CODE_AUTH_FAILURE; } - protected void sendResponse(Response response, ErrorCode errorCode, Object data) { + protected void sendResponse(ErrorCode errorCode, Object data) { response.setStatus(HttpResponseStatus.OK); if (errorCode == null) { errorCode = ErrorCode.ERROR_CODE_SUCCESS; @@ -117,4 +124,30 @@ abstract public class RobotAction extends Action { } return null; } + + protected void sendApiRequest(String topic, byte[] message, Callback callback) { + ServerAPIHelper.sendRequest(robot.getUid(), null, topic, message, robot.getUid(), TargetEntry.Type.TARGET_TYPE_USER, new ServerAPIHelper.Callback() { + @Override + public void onSuccess(byte[] response) { + callback.onSuccess(response); + } + + @Override + public void onError(ErrorCode errorCode) { + sendResponse(errorCode, null); + } + + @Override + public void onTimeout() { + sendResponse(ErrorCode.ERROR_CODE_TIMEOUT, null); + } + + @Override + public Executor getResponseExecutor() { + return command -> { + ctx.executor().execute(command); + }; + } + }, false, true); + } } 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 30646410..034befb1 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 @@ -18,8 +18,6 @@ import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; import cn.wildfirechat.pojos.SendMessageData; import cn.wildfirechat.pojos.SendMessageResult; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; @@ -27,8 +25,6 @@ import io.netty.handler.codec.http.HttpResponseStatus; import cn.wildfirechat.common.ErrorCode; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Robot_Message_Send) @HttpMethod("POST") public class SendMessageAction extends RobotAction { @@ -44,38 +40,18 @@ public class SendMessageAction extends RobotAction { SendMessageData sendMessageData = getRequestBody(request.getNettyRequest(), SendMessageData.class); sendMessageData.setSender(robot.getUid()); if (SendMessageData.isValide(sendMessageData)) { - RPCCenter.getInstance().sendRequest(sendMessageData.getSender(), null, IMTopic.SendMessageTopic, sendMessageData.toProtoMessage().toByteArray(), sendMessageData.getSender(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - long messageId = byteBuf.readLong(); - long timestamp = byteBuf.readLong(); - sendResponse(response, null, new SendMessageResult(messageId, timestamp)); - } else { - sendResponse(response, errorCode, null); - } + sendApiRequest(IMTopic.SendMessageTopic, sendMessageData.toProtoMessage().toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + long messageId = byteBuf.readLong(); + long timestamp = byteBuf.readLong(); + sendResponse(null, new SendMessageResult(messageId, timestamp)); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/TransferGroupAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/TransferGroupAction.java index 29dd55e4..30b5deb7 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/robot/TransferGroupAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/robot/TransferGroupAction.java @@ -14,21 +14,16 @@ import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.pojos.InputTransferGroup; import com.google.gson.Gson; import com.xiaoleilu.loServer.RestResult; -import com.xiaoleilu.loServer.action.admin.AdminAction; import com.xiaoleilu.loServer.annotation.HttpMethod; import com.xiaoleilu.loServer.annotation.Route; import com.xiaoleilu.loServer.handler.Request; import com.xiaoleilu.loServer.handler.Response; -import io.moquette.persistence.RPCCenter; -import io.moquette.persistence.TargetEntry; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.http.FullHttpRequest; import io.netty.handler.codec.http.HttpResponseStatus; import win.liyufan.im.IMTopic; -import java.util.concurrent.Executor; - @Route(APIPath.Robot_Group_Transfer) @HttpMethod("POST") public class TransferGroupAction extends RobotAction { @@ -44,36 +39,16 @@ public class TransferGroupAction extends RobotAction { InputTransferGroup inputDismissGroup = getRequestBody(request.getNettyRequest(), InputTransferGroup.class); inputDismissGroup.setOperator(robot.getUid()); if (inputDismissGroup.isValide()) { - RPCCenter.getInstance().sendRequest(inputDismissGroup.getOperator(), null, IMTopic.TransferGroupTopic, inputDismissGroup.toProtoGroupRequest().toByteArray(), inputDismissGroup.getOperator(), TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { - @Override - public void onSuccess(byte[] result) { - ByteBuf byteBuf = Unpooled.buffer(); - byteBuf.writeBytes(result); - ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); - if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { - sendResponse(response, null, null); - } else { - sendResponse(response, errorCode, null); - } + sendApiRequest(IMTopic.TransferGroupTopic, inputDismissGroup.toProtoGroupRequest().toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { + sendResponse(null, null); + } else { + sendResponse(errorCode, null); } - - @Override - public void onError(ErrorCode errorCode) { - sendResponse(response, errorCode, null); - } - - @Override - public void onTimeout() { - sendResponse(response, ErrorCode.ERROR_CODE_TIMEOUT, null); - } - - @Override - public Executor getResponseExecutor() { - return command -> { - ctx.executor().execute(command); - }; - } - }, false); + }); return false; } else { response.setStatus(HttpResponseStatus.OK); diff --git a/broker/src/main/java/com/xiaoleilu/loServer/handler/ActionHandler.java b/broker/src/main/java/com/xiaoleilu/loServer/handler/ActionHandler.java index eca67cf2..6a661478 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/handler/ActionHandler.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/handler/ActionHandler.java @@ -134,6 +134,7 @@ abstract public class ActionHandler extends SimpleChannelInboundHandler 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; diff --git a/broker/src/main/java/io/moquette/imhandler/ChannelListenMember.java b/broker/src/main/java/io/moquette/imhandler/ChannelListenMember.java index e0eb0ea8..673fdb41 100644 --- a/broker/src/main/java/io/moquette/imhandler/ChannelListenMember.java +++ b/broker/src/main/java/io/moquette/imhandler/ChannelListenMember.java @@ -25,7 +25,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, false); + mServer.onApiMessage(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false, 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 bcc7d937..8404691b 100644 --- a/broker/src/main/java/io/moquette/imhandler/CreateChannelHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/CreateChannelHandler.java @@ -40,7 +40,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, 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; diff --git a/broker/src/main/java/io/moquette/imhandler/DistoryChannelHandler.java b/broker/src/main/java/io/moquette/imhandler/DistoryChannelHandler.java index fea12640..6f370d24 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, false); + mServer.onApiMessage(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, false, false); } return errorCode; } diff --git a/broker/src/main/java/io/moquette/persistence/RPCCenter.java b/broker/src/main/java/io/moquette/persistence/ServerAPIHelper.java similarity index 65% rename from broker/src/main/java/io/moquette/persistence/RPCCenter.java rename to broker/src/main/java/io/moquette/persistence/ServerAPIHelper.java index 9b98cad6..2c6ba72d 100644 --- a/broker/src/main/java/io/moquette/persistence/RPCCenter.java +++ b/broker/src/main/java/io/moquette/persistence/ServerAPIHelper.java @@ -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 requestMap = new ConcurrentHashMap<>(); - private AtomicInteger aiRequestId = new AtomicInteger(1); - public ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2); + private static Server server; + public static ConcurrentHashMap 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); diff --git a/broker/src/main/java/io/moquette/persistence/remote/RequestInfo.java b/broker/src/main/java/io/moquette/persistence/remote/RequestInfo.java index f0425d8e..072dc0db 100644 --- a/broker/src/main/java/io/moquette/persistence/remote/RequestInfo.java +++ b/broker/src/main/java/io/moquette/persistence/remote/RequestInfo.java @@ -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(() -> { diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index d7f79a64..b6247aa7 100755 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -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() { 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 a1fa82d4..90df7ff5 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -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(); 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 e148d91f..75aeccfd 100755 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -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) {