From a074b9eac37fb716c6b29f01e60b80948ccad36d Mon Sep 17 00:00:00 2001 From: heavyrain2012 Date: Fri, 17 Dec 2021 21:24:56 +0800 Subject: [PATCH] =?UTF-8?q?=E9=A2=91=E9=81=93API=E5=8F=91=E9=80=81?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E6=97=B6=EF=BC=8C=E5=A6=82=E6=9E=9C=E6=98=AF?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=E5=A4=84=E7=90=86=E5=B0=B1=E4=B8=8D=E5=8F=91?= =?UTF-8?q?=E9=80=81=E7=BB=99=E9=A2=91=E9=81=93owner?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../java/io/moquette/imhandler/GroupHandler.java | 2 +- .../imhandler/HandleFriendRequestHandler.java | 6 +++--- .../main/java/io/moquette/imhandler/IMHandler.java | 8 ++++---- .../io/moquette/imhandler/RecallMessageHandler.java | 2 +- .../io/moquette/imhandler/SendMessageHandler.java | 10 +++++----- .../io/moquette/persistence/MemoryMessagesStore.java | 12 ++++++++++-- .../main/java/io/moquette/spi/IMessagesStore.java | 3 ++- .../java/io/moquette/spi/impl/MessagesPublisher.java | 6 ++++++ 8 files changed, 32 insertions(+), 17 deletions(-) diff --git a/broker/src/main/java/io/moquette/imhandler/GroupHandler.java b/broker/src/main/java/io/moquette/imhandler/GroupHandler.java index d972668a..d71796e5 100644 --- a/broker/src/main/java/io/moquette/imhandler/GroupHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/GroupHandler.java @@ -43,7 +43,7 @@ abstract public class GroupHandler extends IMHandler { try { long messageId = MessageShardingUtil.generateId(); builder.setMessageId(messageId); - saveAndPublish(fromUser, null, builder.build()); + saveAndPublish(fromUser, null, builder.build(), ProtoConstants.RequestSourceType.Request_From_User); } catch (Exception e) { e.printStackTrace(); } diff --git a/broker/src/main/java/io/moquette/imhandler/HandleFriendRequestHandler.java b/broker/src/main/java/io/moquette/imhandler/HandleFriendRequestHandler.java index 82ef542c..04f59be1 100644 --- a/broker/src/main/java/io/moquette/imhandler/HandleFriendRequestHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/HandleFriendRequestHandler.java @@ -40,7 +40,7 @@ public class HandleFriendRequestHandler extends IMHandler { public void afterAction(String clientID, String fromUser, String topic, Qos1PublishHandler.IMCallback callback) { } - protected long publish(String username, String clientID, WFCMessage.Message message) { + protected long publish(String username, String clientID, WFCMessage.Message message, ProtoConstants.RequestSourceType requestSourceType) { Set notifyReceivers = new LinkedHashSet<>(); WFCMessage.Message.Builder messageBuilder = message.toBuilder(); - int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers); + int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers, requestSourceType); mServer.getImBusinessScheduler().execute(() -> this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType)); return notifyReceivers.size(); } - protected long saveAndPublish(String username, String clientID, WFCMessage.Message message) { + protected long saveAndPublish(String username, String clientID, WFCMessage.Message message, ProtoConstants.RequestSourceType requestSourceType) { Set notifyReceivers = new LinkedHashSet<>(); message = m_messagesStore.storeMessage(username, clientID, message); WFCMessage.Message.Builder messageBuilder = message.toBuilder(); - int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers); + int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers, requestSourceType); mServer.getImBusinessScheduler().execute(() -> this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType)); return notifyReceivers.size(); } diff --git a/broker/src/main/java/io/moquette/imhandler/RecallMessageHandler.java b/broker/src/main/java/io/moquette/imhandler/RecallMessageHandler.java index 7691633c..5484936e 100644 --- a/broker/src/main/java/io/moquette/imhandler/RecallMessageHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/RecallMessageHandler.java @@ -34,7 +34,7 @@ public class RecallMessageHandler extends IMHandler { return ErrorCode.ERROR_CODE_NOT_EXIST; } - publish(fromUser, clientID, message); + publish(fromUser, clientID, message, requestSourceType); return errorCode; } diff --git a/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java b/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java index c28d82dd..3f5ea183 100644 --- a/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java @@ -203,15 +203,15 @@ public class SendMessageHandler extends IMHandler { @Override public void onSuccess(String content) { if(StringUtil.isNullOrEmpty(content)) { - saveAndPublish(fromUser, clientID, finalMsg); + saveAndPublish(fromUser, clientID, finalMsg, requestSourceType); } else { MessagePayload payload = new Gson().fromJson(content, MessagePayload.class); if (payload != null && payload.getType() > 0) { WFCMessage.Message newMsg = finalMsg.toBuilder().setContent(payload.toProtoMessageContent()).build(); - saveAndPublish(fromUser, clientID, newMsg); + saveAndPublish(fromUser, clientID, newMsg, requestSourceType); } else { LOG.error("Response content {} from censor is invalid payload, ignore response and send original message", content); - saveAndPublish(fromUser, clientID, finalMsg); + saveAndPublish(fromUser, clientID, finalMsg, requestSourceType); } } if(mWaitRemoteSensitiveServerResponse) { @@ -236,7 +236,7 @@ public class SendMessageHandler extends IMHandler { } } else { LOG.warn("Failed to censor message with status {}, errorMessage {}. Send the original message", statusCode, errorMessage); - saveAndPublish(fromUser, clientID, finalMsg); + saveAndPublish(fromUser, clientID, finalMsg, requestSourceType); if(mWaitRemoteSensitiveServerResponse) { errorCode = ERROR_CODE_SUCCESS; @@ -268,7 +268,7 @@ public class SendMessageHandler extends IMHandler { if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { if(!ignoreMsg) { - saveAndPublish(fromUser, clientID, message); + saveAndPublish(fromUser, clientID, message, requestSourceType); } ackPayload = ackPayload.capacity(20); ackPayload.writeLong(messageId); diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 7841ec31..82df6375 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -496,7 +496,7 @@ public class MemoryMessagesStore implements IMessagesStore { @Override - public int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder messageBuilder, Set notifyReceivers) { + public int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder messageBuilder, Set notifyReceivers, ProtoConstants.RequestSourceType requestSourceType) { WFCMessage.Message message = messageBuilder.build(); HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); int type = message.getConversation().getType(); @@ -556,8 +556,15 @@ public class MemoryMessagesStore implements IMessagesStore { } else if(type == ProtoConstants.ConversationType.ConversationType_Channel) { WFCMessage.ChannelInfo channelInfo = getChannelInfo(message.getConversation().getTarget()); if (channelInfo != null) { - notifyReceivers.add(fromUser); if (channelInfo.getOwner().equals(fromUser)) { + if(requestSourceType == ProtoConstants.RequestSourceType.Request_From_Channel + && !StringUtil.isNullOrEmpty(channelInfo.getCallback()) + && channelInfo.getAutomatic() == 1) { + LOG.info("Channel api message not send to the owner when automatic"); + } else { + notifyReceivers.add(fromUser); + } + if((channelInfo.getStatus() & ProtoConstants.ChannelState.Channel_State_Mask_Global) > 0) { if((message.getToList() != null && !message.getToList().isEmpty())) { notifyReceivers.addAll(message.getToList()); @@ -581,6 +588,7 @@ public class MemoryMessagesStore implements IMessagesStore { } } } else { + notifyReceivers.add(fromUser); if (StringUtil.isNullOrEmpty(channelInfo.getCallback()) || channelInfo.getAutomatic() == 0) { notifyReceivers.add(channelInfo.getOwner()); } diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 97f11f05..1bb82c75 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -18,6 +18,7 @@ package io.moquette.spi; import cn.wildfirechat.pojos.InputOutputUserInfo; import cn.wildfirechat.pojos.SystemSettingPojo; +import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; import com.xiaoleilu.loServer.model.FriendData; import cn.wildfirechat.pojos.InputOutputUserBlockStatus; @@ -101,7 +102,7 @@ public interface IMessagesStore { DatabaseStore getDatabaseStore(); WFCMessage.Message storeMessage(String fromUser, String fromClientId, WFCMessage.Message message); void storeSensitiveMessage(WFCMessage.Message message); - int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder message, Set notifyReceivers); + int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder message, Set notifyReceivers, ProtoConstants.RequestSourceType requestSourceType); Set getAllEnds(); WFCMessage.PullMessageResult fetchMessage(String user, String exceptClientId, long fromMessageId, int pullType); WFCMessage.PullMessageResult loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count, Collection contentTypes); 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 5799a6db..f4c73c77 100755 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -604,6 +604,12 @@ public class MessagesPublisher { Utility.printExecption(LOG, e, EVENT_CALLBACK_Exception); } }); + } else { + if (channelInfo == null) { + LOG.info("Channel {} not exist", message.getConversation().getTarget()); + } else { + LOG.info("Channel {} callback {}", message.getConversation().getTarget(), channelInfo.getCallback()); + } } } long messageId = message.getMessageId();