diff --git a/broker/config/wildfirechat.conf b/broker/config/wildfirechat.conf index d23ba04a..2cd73984 100755 --- a/broker/config/wildfirechat.conf +++ b/broker/config/wildfirechat.conf @@ -179,15 +179,43 @@ push.ios.server.address http://localhost:8085/ios/push #********************************************************************* ##用户在线状态事件回调地址 ##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! -#user.online_status_callback http://localhost:8888/user/online_event +#user.online_status_callback http://localhost:8888/im_event/user/online + +##用户信息变动事件回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#relation.relation_update_callback http://localhost:8888/im_event/user/relation + +##用户信息变动事件回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#user.user_info_update_callback http://localhost:8888/im_event/user/info ##消息转发地址 ##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! -#message.forward.url http://localhost:8087/message/forward +#message.forward.url http://localhost:8888/im_event/message ##设备信息转发地址 ##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! -#things_message.forward.url http://localhost:8087/device/forward +#things_message.forward.url http://localhost:8888/im_event/things/message + +##群组信息变动事件回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#group.group_info_update_callback http://localhost:8888/im_event/group/info + +##群组信息变动事件回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#group.group_member_update_callback http://localhost:8888/im_event/group/member + +##频道信息变动回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#channel.channel_info_update_callback http://localhost:8888/im_event/channel/info + +##聊天室信息变动回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#chatroom.chatroom_info_update_callback http://localhost:8888/im_event/chatroom/info + +##聊天室成员变动回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#chatroom.chatroom_member_update_callback http://localhost:8888/im_event/chatroom/member #********************************************************************* # Sensitive configuration diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 88194079..e0432848 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -94,6 +94,17 @@ public final class BrokerConstants { public static final String USER_ONLINE_STATUS_CALLBACK = "user.online_status_callback"; + public static final String GROUP_INFO_UPDATE_CALLBACK = "group.group_info_update_callback"; + public static final String GROUP_MEMBER_UPDATE_CALLBACK = "group.group_member_update_callback"; + public static final String RELATION_UPDATE_CALLBACK = "relation.relation_update_callback"; + public static final String USER_INFO_UPDATE_CALLBACK = "user.user_info_update_callback"; + + + public static final String CHANNEL_INFO_UPDATE_CALLBACK = "channel.channel_info_update_callback"; + public static final String CHATROOM_INFO_UPDATE_CALLBACK = "chatroom.chatroom_info_update_callback"; + public static final String CHATROOM_MEMBER_UPDATE_CALLBACK = "chatroom.chatroom_member_update_callback"; + + public static final String HZ_Cluster_Node_External_IP = "node_external_ip"; public static final String HZ_Cluster_Node_External_Long_Port = "node_external_long_port"; public static final String HZ_Cluster_Node_External_Short_Port = "node_external_short_port"; diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 2a039cbe..80e2a4e1 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -16,14 +16,14 @@ package io.moquette.persistence; -import cn.wildfirechat.pojos.SystemSettingPojo; +import cn.wildfirechat.pojos.*; import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; +import com.google.gson.Gson; import com.google.protobuf.ByteString; import com.hazelcast.core.*; import com.hazelcast.util.StringUtil; import com.xiaoleilu.loServer.model.FriendData; -import cn.wildfirechat.pojos.InputOutputUserBlockStatus; import cn.wildfirechat.common.ErrorCode; import io.moquette.BrokerConstants; import io.moquette.imhandler.IMHandler; @@ -53,6 +53,11 @@ import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Consumer; import static cn.wildfirechat.proto.ProtoConstants.ChannelState.Channel_State_Mask_Deleted; +import static cn.wildfirechat.proto.ProtoConstants.ChannelUpdateEventType.*; +import static cn.wildfirechat.proto.ProtoConstants.ChatroomMemberUpdateEventType.Chatroom_Member_Event_Join; +import static cn.wildfirechat.proto.ProtoConstants.ChatroomMemberUpdateEventType.Chatroom_Member_Event_Leave; +import static cn.wildfirechat.proto.ProtoConstants.ChatroomUpdateEventType.Chatroom_Event_Create; +import static cn.wildfirechat.proto.ProtoConstants.ChatroomUpdateEventType.Chatroom_Event_Destroy; import static cn.wildfirechat.proto.ProtoConstants.GroupMemberType.*; import static cn.wildfirechat.proto.ProtoConstants.ModifyChannelInfoType.*; import static cn.wildfirechat.proto.ProtoConstants.ModifyGroupInfoType.*; @@ -138,6 +143,14 @@ public class MemoryMessagesStore implements IMessagesStore { private long mRecallTimeLimit = 300; + private String mGroupInfoUpdateCallback; + private String mGroupMemberUpdateCallback; + private String mRelationUpdateCallback; + private String mUserInfoUpdateCallback; + private String mChannelInfoUpdateCallback; + private String mChatroomInfoUpdateCallback; + private String mChatroomMemberUpdateCallback; + MemoryMessagesStore(Server server, DatabaseStore databaseStore) { m_Server = server; this.databaseStore = databaseStore; @@ -243,6 +256,48 @@ public class MemoryMessagesStore implements IMessagesStore { mRecallTimeLimit = Long.parseLong(m_Server.getConfig().getProperty(BrokerConstants.MESSAGES_RECALL_TIME_LIMIT)); } catch (Exception e) { } + + try { + mGroupInfoUpdateCallback = server.getConfig().getProperty(GROUP_INFO_UPDATE_CALLBACK); + } catch (Exception e) { + + } + + try { + mGroupMemberUpdateCallback = server.getConfig().getProperty(GROUP_MEMBER_UPDATE_CALLBACK); + } catch (Exception e) { + + } + + try { + mRelationUpdateCallback = server.getConfig().getProperty(RELATION_UPDATE_CALLBACK); + } catch (Exception e) { + + } + + try { + mUserInfoUpdateCallback = server.getConfig().getProperty(USER_INFO_UPDATE_CALLBACK); + } catch (Exception e) { + + } + + try { + mChannelInfoUpdateCallback = server.getConfig().getProperty(CHANNEL_INFO_UPDATE_CALLBACK); + } catch (Exception e) { + + } + + try { + mChatroomInfoUpdateCallback = server.getConfig().getProperty(CHATROOM_INFO_UPDATE_CALLBACK); + } catch (Exception e) { + + } + + try { + mChatroomMemberUpdateCallback = server.getConfig().getProperty(CHATROOM_MEMBER_UPDATE_CALLBACK); + } catch (Exception e) { + + } } private void printMissConfigLog(String config, String defaultValue) { @@ -822,10 +877,80 @@ public class MemoryMessagesStore implements IMessagesStore { mIMap.set(groupId, groupInfo); databaseStore.persistGroupMember(groupId, updatedMemberList); + callbackGroupEvent(fromUser, groupInfo.getTargetId(), ProtoConstants.GroupUpdateEventType.Group_Event_Create); return groupInfo; } + private void callbackGroupEvent(String operatorId, String groupId, int type) { + if (!StringUtil.isNullOrEmpty(mGroupInfoUpdateCallback)) { + GroupUpdateEvent event = new GroupUpdateEvent(); + event.operatorId = operatorId; + event.groupId = groupId; + event.type = type; + m_Server.getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(mGroupInfoUpdateCallback, new Gson().toJson(event))); + } + } + + private void callbackGroupMemberEvent(String operatorId, String groupId, List memberIds, int type, String value) { + if (!StringUtil.isNullOrEmpty(mGroupMemberUpdateCallback)) { + GroupMemberUpdateEvent event = new GroupMemberUpdateEvent(); + event.operatorId = operatorId; + event.groupId = groupId; + event.memberIds = memberIds; + event.type = type; + event.value = value; + m_Server.getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(mGroupMemberUpdateCallback, new Gson().toJson(event))); + } + } + + private void callbackRelationEvent(String userId, String targetId, int type, String value) { + if (!StringUtil.isNullOrEmpty(mRelationUpdateCallback)) { + RelationUpdateEvent event = new RelationUpdateEvent(); + event.userId = userId; + event.targetId = targetId; + event.type = type; + event.value = value; + m_Server.getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(mRelationUpdateCallback, new Gson().toJson(event))); + } + } + + private void callbackUserInfoEvent(WFCMessage.User user) { + if (!StringUtil.isNullOrEmpty(mUserInfoUpdateCallback)) { + m_Server.getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(mUserInfoUpdateCallback, new Gson().toJson(InputOutputUserInfo.fromPbUser(user)))); + } + } + + private void callbackChannelInfoUpdateEvent(String operatorId, String channelId, int type) { + if (!StringUtil.isNullOrEmpty(mChannelInfoUpdateCallback)) { + ChannelUpdateEvent event = new ChannelUpdateEvent(); + event.operatorId = operatorId; + event.channelId = channelId; + event.type = type; + m_Server.getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(mChannelInfoUpdateCallback, new Gson().toJson(event))); + } + } + + private void callbackChatroomInfoUpdateEvent(String chatroomId, int type) { + if (!StringUtil.isNullOrEmpty(mChatroomInfoUpdateCallback)) { + ChatroomUpdateEvent event = new ChatroomUpdateEvent(); + event.chatroomId = chatroomId; + event.type = type; + m_Server.getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(mChatroomInfoUpdateCallback, new Gson().toJson(event))); + } + } + + private void callbackChatroomMemberEvent(String operatorId, String chatroomId, List memberIds, int type) { + if (!StringUtil.isNullOrEmpty(mChatroomMemberUpdateCallback)) { + ChatroomMemberUpdateEvent event = new ChatroomMemberUpdateEvent(); + event.operatorId = operatorId; + event.chatroomId = chatroomId; + event.memberIds = memberIds; + event.type = type; + m_Server.getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(mChatroomMemberUpdateCallback, new Gson().toJson(event))); + } + } + @Override public ErrorCode addGroupMembers(String operator, boolean isAdmin, String groupId, List memberList) { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); @@ -940,6 +1065,15 @@ public class MemoryMessagesStore implements IMessagesStore { mIMap.put(groupId, groupInfo.toBuilder().setMemberUpdateDt(updateDt).setUpdateDt(updateDt).setMemberCount(count).build()); databaseStore.persistGroupMember(groupId, memberList); databaseStore.updateGroupMemberCountDt(groupId, count, updateDt); + + List memberIds = new ArrayList<>(); + for (WFCMessage.GroupMember member : memberList) { + if (member.getType() != GroupMemberType_Removed) { + memberIds.add(member.getMemberId()); + } + } + + callbackGroupMemberEvent(operator, groupId, memberIds, ProtoConstants.GroupMemberUpdateEventType.Group_Member_Event_Join, null); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -948,6 +1082,8 @@ public class MemoryMessagesStore implements IMessagesStore { removeGroupMember(groupId, memberList); removeFavGroup(groupId, memberList); removeGroupUserSettings(groupId, memberList); + + callbackGroupMemberEvent(operator, groupId, memberList, ProtoConstants.GroupMemberUpdateEventType.Group_Member_Event_Kickoff, null); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -1026,6 +1162,8 @@ public class MemoryMessagesStore implements IMessagesStore { removeGroupMember(groupId, Arrays.asList(operator)); removeFavGroup(groupId, Arrays.asList(operator)); removeGroupUserSettings(groupId, Arrays.asList(operator)); + + callbackGroupMemberEvent(operator, groupId, Arrays.asList(operator), ProtoConstants.GroupMemberUpdateEventType.Group_Member_Event_Leave, null); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -1093,6 +1231,7 @@ public class MemoryMessagesStore implements IMessagesStore { removeFavGroup(groupId, ids); removeGroupUserSettings(groupId, ids); + callbackGroupEvent(operator, groupId, ProtoConstants.GroupUpdateEventType.Group_Event_Destroy); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -1168,6 +1307,16 @@ public class MemoryMessagesStore implements IMessagesStore { newInfoBuilder.setUpdateDt(System.currentTimeMillis()); mIMap.put(groupId, newInfoBuilder.build()); + + if (modifyType == Modify_Group_Mute) { + if (newInfoBuilder.getMute() > 0) { + callbackGroupEvent(operator, groupId, ProtoConstants.GroupUpdateEventType.Group_Event_Mute); + } else { + callbackGroupEvent(operator, groupId, ProtoConstants.GroupUpdateEventType.Group_Event_Unmute); + } + } else { + callbackGroupEvent(operator, groupId, ProtoConstants.GroupUpdateEventType.Group_Event_Update); + } return ErrorCode.ERROR_CODE_SUCCESS; } @@ -1252,6 +1401,7 @@ public class MemoryMessagesStore implements IMessagesStore { return ErrorCode.ERROR_CODE_NOT_IN_GROUP; } + callbackGroupMemberEvent(operator, groupId, Arrays.asList(memberId), ProtoConstants.GroupMemberUpdateEventType.Group_Member_Event_Alias, alias); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -1407,6 +1557,7 @@ public class MemoryMessagesStore implements IMessagesStore { } } + callbackGroupEvent(operator, groupId, ProtoConstants.GroupUpdateEventType.Group_Event_Transfer); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -1445,6 +1596,8 @@ public class MemoryMessagesStore implements IMessagesStore { } mIMap.put(groupId, groupInfo.toBuilder().setUpdateDt(updateDt).setMemberUpdateDt(updateDt).build()); + + callbackGroupMemberEvent(operator, groupId, userList, ProtoConstants.GroupMemberUpdateEventType.Group_Member_Event_Type_Update, (type == 0 ? ProtoConstants.GroupMemberType.GroupMemberType_Normal : ProtoConstants.GroupMemberType.GroupMemberType_Manager) + ""); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -1733,6 +1886,9 @@ public class MemoryMessagesStore implements IMessagesStore { user = builder.build(); databaseStore.updateUser(user); mUserMap.set(userId, user); + + callbackUserInfoEvent(user); + return ErrorCode.ERROR_CODE_SUCCESS; } else { return ErrorCode.ERROR_CODE_NOT_MODIFIED; @@ -1836,6 +1992,8 @@ public class MemoryMessagesStore implements IMessagesStore { databaseStore.updateUser(user); mUserMap.put(user.getUid(), user); databaseStore.updateUserPassword(user.getUid(), password); + + callbackUserInfoEvent(user); } @Override @@ -1930,6 +2088,7 @@ public class MemoryMessagesStore implements IMessagesStore { chatroomInfo = builder.build(); } chatroomInfoMap.put(chatroomId, chatroomInfo); + callbackChatroomInfoUpdateEvent(chatroomId, Chatroom_Event_Create); } @Override @@ -1940,6 +2099,7 @@ public class MemoryMessagesStore implements IMessagesStore { if (room != null) { room = room.toBuilder().setUpdateDt(System.currentTimeMillis()).setState(ProtoConstants.ChatroomState.Chatroom_State_End).build(); chatroomInfoMap.put(chatroomId, room); + callbackChatroomInfoUpdateEvent(chatroomId, Chatroom_Event_Destroy); } } @@ -2326,6 +2486,8 @@ public class MemoryMessagesStore implements IMessagesStore { heads[1] = friendData1.getTimestamp(); friendsMap.remove(userId); friendsMap.remove(request.getTargetUid()); + + callbackRelationEvent(userId, request.getTargetUid(), 0, "" + request.getStatus()); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -2388,6 +2550,8 @@ public class MemoryMessagesStore implements IMessagesStore { msgBuilder.setConversation(WFCMessage.Conversation.newBuilder().setTarget(userId).setLine(0).setType(ProtoConstants.ConversationType.ConversationType_Private).build()); msgBuilder.setContent(WFCMessage.MessageContent.newBuilder().setType(1).setSearchableContent(existRequest.getReason()).build()); + + callbackRelationEvent(userId, request.getTargetUid(), 0, "1"); } return ErrorCode.ERROR_CODE_SUCCESS; } @@ -2436,6 +2600,7 @@ public class MemoryMessagesStore implements IMessagesStore { heads[0] = friendData.getTimestamp(); + callbackRelationEvent(fromUser, targetUserId, 2, state+""); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -2489,6 +2654,7 @@ public class MemoryMessagesStore implements IMessagesStore { friendsMap.remove(fromUser); + callbackRelationEvent(fromUser, targetUserId, 1, alias); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -2507,6 +2673,7 @@ public class MemoryMessagesStore implements IMessagesStore { chatroomMessages.put(userId, new TreeMap<>()); mWriteLock.unlock(); + callbackChatroomMemberEvent(userId, chatroomId, Arrays.asList(userId), Chatroom_Member_Event_Join); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -2519,6 +2686,8 @@ public class MemoryMessagesStore implements IMessagesStore { mWriteLock.lock(); chatroomMessages.remove(userId); mWriteLock.unlock(); + + callbackChatroomMemberEvent(userId, chatroomId, Arrays.asList(userId), Chatroom_Member_Event_Leave); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -2577,6 +2746,7 @@ public class MemoryMessagesStore implements IMessagesStore { friendsMap.remove(userId); friendsMap.remove(friendUid); + callbackRelationEvent(userId, friendUid, 0, "0"); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -2807,6 +2977,7 @@ public class MemoryMessagesStore implements IMessagesStore { mIMap.put(channelInfo.getTargetId(), channelInfo); + callbackChannelInfoUpdateEvent(operator, channelInfo.getTargetId(), Channel_Event_Create); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -2859,6 +3030,7 @@ public class MemoryMessagesStore implements IMessagesStore { newInfoBuilder.setUpdateDt(System.currentTimeMillis()); mIMap.put(channelId, newInfoBuilder.build()); + callbackChannelInfoUpdateEvent(operator, channelId, Channel_Event_Update); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -2885,6 +3057,8 @@ public class MemoryMessagesStore implements IMessagesStore { newInfoBuilder.setOwner(newOwner); newInfoBuilder.setUpdateDt(System.currentTimeMillis()); mIMap.put(channelId, newInfoBuilder.build()); + + callbackChannelInfoUpdateEvent(operator, channelId, Channel_Event_Transfer); return ErrorCode.ERROR_CODE_SUCCESS; } @@ -2909,6 +3083,8 @@ public class MemoryMessagesStore implements IMessagesStore { newInfoBuilder.setStatus(oldInfo.getStatus() | Channel_State_Mask_Deleted); newInfoBuilder.setUpdateDt(System.currentTimeMillis()); mIMap.put(channelId, newInfoBuilder.build()); + + callbackChannelInfoUpdateEvent(operator, channelId, Channel_Event_Destroy); return ErrorCode.ERROR_CODE_SUCCESS; } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 63f83e12..db3593d2 100755 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -94,6 +94,7 @@ public class Server { private ThreadPoolExecutorWrapper dbScheduler; private ThreadPoolExecutorWrapper imBusinessScheduler; + private ThreadPoolExecutorWrapper callbackScheduler; private IConfig mConfig; @@ -243,6 +244,7 @@ public class Server { int threadNum = Runtime.getRuntime().availableProcessors() * 2; dbScheduler = new ThreadPoolExecutorWrapper(Executors.newScheduledThreadPool(threadNum), threadNum, "db"); imBusinessScheduler = new ThreadPoolExecutorWrapper(Executors.newScheduledThreadPool(threadNum), threadNum, "business"); + callbackScheduler = new ThreadPoolExecutorWrapper(Executors.newScheduledThreadPool(1), 1, "callback"); final String handlerProp = System.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME); if (handlerProp != null) { @@ -447,6 +449,7 @@ public class Server { dbScheduler.shutdown(); imBusinessScheduler.shutdown(); + callbackScheduler.shutdown(); LOG.info("Moquette server has been stopped."); } @@ -507,4 +510,8 @@ public class Server { public ThreadPoolExecutorWrapper getImBusinessScheduler() { return imBusinessScheduler; } + + public ThreadPoolExecutorWrapper getCallbackScheduler() { + return callbackScheduler; + } } diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index 9c2e978b..1abc15ab 100755 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -29,6 +29,7 @@ import com.xiaoleilu.loServer.model.FriendData; import io.moquette.persistence.*; import io.moquette.persistence.MemorySessionStore.Session; import io.moquette.server.ConnectionDescriptorStore; +import io.moquette.server.Server; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.netty.buffer.ByteBuf; @@ -175,7 +176,7 @@ public class MessagesPublisher { message = m_messagesStore.getMessage(messageHead); } final WFCMessage.Message finalMsg = message; - executorCallback.execute(() -> HttpUtils.httpJsonPost(robot.getCallback(), new Gson().toJson(SendMessageData.fromProtoMessage(finalMsg), SendMessageData.class))); + Server.getServer().getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(robot.getCallback(), new Gson().toJson(SendMessageData.fromProtoMessage(finalMsg), SendMessageData.class))); continue; } } @@ -531,7 +532,7 @@ public class MessagesPublisher { if (message.getConversation().getType() == ProtoConstants.ConversationType.ConversationType_Channel) { WFCMessage.ChannelInfo channelInfo = m_messagesStore.getChannelInfo(message.getConversation().getTarget()); if (channelInfo != null && !StringUtil.isNullOrEmpty(channelInfo.getCallback())) { - executorCallback.execute(() -> HttpUtils.httpJsonPost(channelInfo.getCallback() + "/message", new Gson().toJson(SendMessageData.fromProtoMessage(message), SendMessageData.class))); + Server.getServer().getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(channelInfo.getCallback() + "/message", new Gson().toJson(SendMessageData.fromProtoMessage(message), SendMessageData.class))); } } long messageId = message.getMessageId(); @@ -571,13 +572,14 @@ public class MessagesPublisher { } public void forwardMessage(final WFCMessage.Message message, String forwardUrl) { - executorCallback.execute(() -> HttpUtils.httpJsonPost(forwardUrl, new Gson().toJson(OutputMessageData.fromProtoMessage(message), OutputMessageData.class))); + Server.getServer().getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(forwardUrl, new Gson().toJson(OutputMessageData.fromProtoMessage(message), OutputMessageData.class))); } public void notifyChannelListenStatusChanged(WFCMessage.ChannelInfo channelInfo, String user, boolean listen) { if (channelInfo == null || StringUtil.isNullOrEmpty(channelInfo.getCallback())) { return; } - executorCallback.execute(() -> HttpUtils.httpJsonPost(channelInfo.getCallback() + "/subscribe", new Gson().toJson(new OutputNotifyChannelSubscribeStatus(user, channelInfo.getTargetId(), listen), OutputNotifyChannelSubscribeStatus.class))); + + Server.getServer().getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(channelInfo.getCallback() + "/subscribe", new Gson().toJson(new OutputNotifyChannelSubscribeStatus(user, channelInfo.getTargetId(), listen), OutputNotifyChannelSubscribeStatus.class))); } } 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 6163ed71..77309555 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -252,7 +252,7 @@ public class ProtocolProcessor { public void forwardOnlineStatusEvent(String userId, String clientId, int platform, int status, String packageName) { if (!StringUtil.isNullOrEmpty(mUserOnlineStatusCallback)) { - executorCallback.execute(() -> HttpUtils.httpJsonPost(mUserOnlineStatusCallback, new Gson().toJson(new UserOnlineStatus(userId, clientId, platform, status, packageName)))); + mServer.getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(mUserOnlineStatusCallback, new Gson().toJson(new UserOnlineStatus(userId, clientId, platform, status, packageName)))); } } diff --git a/common/src/main/java/cn/wildfirechat/pojos/ChannelUpdateEvent.java b/common/src/main/java/cn/wildfirechat/pojos/ChannelUpdateEvent.java new file mode 100644 index 00000000..d6f28ea8 --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/ChannelUpdateEvent.java @@ -0,0 +1,7 @@ +package cn.wildfirechat.pojos; + +public class ChannelUpdateEvent { + public String operatorId; + public String channelId; + public int type; +} diff --git a/common/src/main/java/cn/wildfirechat/pojos/ChatroomMemberUpdateEvent.java b/common/src/main/java/cn/wildfirechat/pojos/ChatroomMemberUpdateEvent.java new file mode 100644 index 00000000..374f8509 --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/ChatroomMemberUpdateEvent.java @@ -0,0 +1,10 @@ +package cn.wildfirechat.pojos; + +import java.util.List; + +public class ChatroomMemberUpdateEvent { + public String operatorId; + public String chatroomId; + public List memberIds; + public int type; +} diff --git a/common/src/main/java/cn/wildfirechat/pojos/ChatroomUpdateEvent.java b/common/src/main/java/cn/wildfirechat/pojos/ChatroomUpdateEvent.java new file mode 100644 index 00000000..803b3212 --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/ChatroomUpdateEvent.java @@ -0,0 +1,6 @@ +package cn.wildfirechat.pojos; + +public class ChatroomUpdateEvent { + public String chatroomId; + public int type; +} diff --git a/common/src/main/java/cn/wildfirechat/pojos/GroupMemberUpdateEvent.java b/common/src/main/java/cn/wildfirechat/pojos/GroupMemberUpdateEvent.java new file mode 100644 index 00000000..58856847 --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/GroupMemberUpdateEvent.java @@ -0,0 +1,11 @@ +package cn.wildfirechat.pojos; + +import java.util.List; + +public class GroupMemberUpdateEvent { + public String operatorId; + public String groupId; + public List memberIds; + public int type; + public String value; +} diff --git a/common/src/main/java/cn/wildfirechat/pojos/GroupUpdateEvent.java b/common/src/main/java/cn/wildfirechat/pojos/GroupUpdateEvent.java new file mode 100644 index 00000000..7bd8936b --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/GroupUpdateEvent.java @@ -0,0 +1,7 @@ +package cn.wildfirechat.pojos; + +public class GroupUpdateEvent { + public String operatorId; + public String groupId; + public int type; +} diff --git a/common/src/main/java/cn/wildfirechat/pojos/RelationUpdateEvent.java b/common/src/main/java/cn/wildfirechat/pojos/RelationUpdateEvent.java new file mode 100644 index 00000000..9bce3958 --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/RelationUpdateEvent.java @@ -0,0 +1,8 @@ +package cn.wildfirechat.pojos; + +public class RelationUpdateEvent { + public String userId; + public String targetId; + public int type; + public String value; +} diff --git a/common/src/main/java/cn/wildfirechat/pojos/UserOnlineStatus.java b/common/src/main/java/cn/wildfirechat/pojos/UserOnlineStatus.java index 7c760ec8..242402ed 100644 --- a/common/src/main/java/cn/wildfirechat/pojos/UserOnlineStatus.java +++ b/common/src/main/java/cn/wildfirechat/pojos/UserOnlineStatus.java @@ -12,6 +12,9 @@ public class UserOnlineStatus { public long timestamp; public String packageName; + public UserOnlineStatus() { + } + public UserOnlineStatus(String userId, String clientId, int platform, int status, String packageName) { this.userId = userId; this.clientId = clientId; diff --git a/common/src/main/java/cn/wildfirechat/proto/ProtoConstants.java b/common/src/main/java/cn/wildfirechat/proto/ProtoConstants.java index 86ea8067..4d74b1cd 100644 --- a/common/src/main/java/cn/wildfirechat/proto/ProtoConstants.java +++ b/common/src/main/java/cn/wildfirechat/proto/ProtoConstants.java @@ -164,7 +164,42 @@ public class ProtoConstants { int Message_Ignore = 1; } + public interface GroupUpdateEventType { + int Group_Event_Create = 0; + int Group_Event_Update = 1; + int Group_Event_Transfer = 2; + int Group_Event_Mute = 3; + int Group_Event_Unmute = 4; + int Group_Event_Destroy = 5; + } + public interface GroupMemberUpdateEventType { + int Group_Member_Event_Join = 0; + int Group_Member_Event_Leave = 1; + int Group_Member_Event_Kickoff = 2; + int Group_Member_Event_Type_Update = 3; + int Group_Member_Event_Alias = 4; + } + + public interface ChannelUpdateEventType { + int Channel_Event_Create = 0; + int Channel_Event_Update = 1; + int Channel_Event_Transfer = 2; + int Channel_Event_Destroy = 3; + } + + public interface ChatroomUpdateEventType { + int Chatroom_Event_Create = 0; + int Chatroom_Event_Destroy = 1; + } + + public interface ChatroomMemberUpdateEventType { + int Chatroom_Member_Event_Join = 0; + int Chatroom_Member_Event_Leave = 1; + int Chatroom_Member_Event_Kickoff = 2; + int Chatroom_Member_Event_Mute = 3; + int Chatroom_Member_Event_Unmute = 4; + } public static final int MESSAGE_CONTENT_TYPE_CREATE_GROUP = 104; public static final int MESSAGE_CONTENT_TYPE_ADD_GROUP_MEMBER = 105; diff --git a/distribution/src/main/resources/wildfirechat.conf b/distribution/src/main/resources/wildfirechat.conf index 697a87d8..bc78fd2a 100755 --- a/distribution/src/main/resources/wildfirechat.conf +++ b/distribution/src/main/resources/wildfirechat.conf @@ -173,12 +173,43 @@ push.ios.server.address http://localhost:8085/ios/push #********************************************************************* ##用户在线状态事件回调地址 ##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! -#user.online_status_callback http://localhost:8888/user/online_event +#user.online_status_callback http://localhost:8888/im_event/user/online + +##用户信息变动事件回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#relation.relation_update_callback http://localhost:8888/im_event/user/relation + +##用户信息变动事件回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#user.user_info_update_callback http://localhost:8888/im_event/user/info ##消息转发地址 ##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! -#message.forward.url http://localhost:8087/message/forward +#message.forward.url http://localhost:8087/im_event/message +##设备信息转发地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#things_message.forward.url http://localhost:8087/im_event/things/message + +##群组信息变动事件回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#group.group_info_update_callback http://localhost:8888/im_event/group/info + +##群组信息变动事件回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#group.group_member_update_callback http://localhost:8888/im_event/group/member + +##频道信息变动回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#channel.channel_info_update_callback http://localhost:8888/im_event/channel/info + +##聊天室信息变动回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#chatroom.chatroom_info_update_callback http://localhost:8888/im_event/chatroom/info + +##聊天室成员变动回调地址 +##注意回调是单线程回调,接收服务必须在同一内网,且异步处理快速返回,否则会有延迟问题!! +#chatroom.chatroom_member_update_callback http://localhost:8888/im_event/chatroom/member #********************************************************************* # Netty Configuration