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();