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

频道API发送消息时，如果是自动处理就不发送给频道owner

Browse Source
This commit is contained in:
heavyrain2012
2021-12-17 21:24:56 +08:00
parent f1b2c546a3
commit a074b9eac3
8 changed files with 32 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/imhandler/GroupHandler.java
+1 -1
View File
@@ -43,7 +43,7 @@ abstract public class GroupHandler<T> extends IMHandler<T> {
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();
}
broker/src/main/java/io/moquette/imhandler/HandleFriendRequestHandler.java
+3 -3
View File
@@ -40,7 +40,7 @@ public class HandleFriendRequestHandler extends IMHandler<WFCMessage.HandleFrien
builder.setMessageId(messageId);
builder.setServerTimestamp(timestamp);
if(!StringUtil.isNullOrEmpty(builder.getContent().getSearchableContent())) {
saveAndPublish(request.getTargetUid(), null, builder.build());
saveAndPublish(request.getTargetUid(), null, builder.build(), requestSourceType);
}
MemorySessionStore.Session session = m_sessionsStore.getSession(clientID);
@@ -63,7 +63,7 @@ public class HandleFriendRequestHandler extends IMHandler<WFCMessage.HandleFrien
messageId = MessageShardingUtil.generateId();
builder.setMessageId(messageId);
saveAndPublish(request.getTargetUid(), null, builder.build());
saveAndPublish(request.getTargetUid(), null, builder.build(), requestSourceType);
if (m_messagesStore.isNewFriendWelcomeMessage()) {
contentBuilder.setType(93);
@@ -75,7 +75,7 @@ public class HandleFriendRequestHandler extends IMHandler<WFCMessage.HandleFrien
messageId = MessageShardingUtil.generateId();
builder.setMessageId(messageId);
builder.setServerTimestamp(++timestamp);
saveAndPublish(request.getTargetUid(), null, builder.build());
saveAndPublish(request.getTargetUid(), null, builder.build(), requestSourceType);
} catch (Exception e) {
e.printStackTrace();
}
broker/src/main/java/io/moquette/imhandler/IMHandler.java
+4 -4
View File
@@ -246,21 +246,21 @@ abstract public class IMHandler<T> {
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<String> 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<String> 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();
}
broker/src/main/java/io/moquette/imhandler/RecallMessageHandler.java
+1 -1
View File
@@ -34,7 +34,7 @@ public class RecallMessageHandler extends IMHandler<WFCMessage.INT64Buf> {
return ErrorCode.ERROR_CODE_NOT_EXIST;
}
publish(fromUser, clientID, message);
publish(fromUser, clientID, message, requestSourceType);
return errorCode;
}
broker/src/main/java/io/moquette/imhandler/SendMessageHandler.java
+5 -5
View File
@@ -203,15 +203,15 @@ public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
@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<WFCMessage.Message> {
}
} 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<WFCMessage.Message> {
if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) {
if(!ignoreMsg) {
saveAndPublish(fromUser, clientID, message);
saveAndPublish(fromUser, clientID, message, requestSourceType);
}
ackPayload = ackPayload.capacity(20);
ackPayload.writeLong(messageId);
broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
+10 -2
View File
@@ -496,7 +496,7 @@ public class MemoryMessagesStore implements IMessagesStore {
@Override
public int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder messageBuilder, Set<String> notifyReceivers) {
public int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder messageBuilder, Set<String> 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());
}
broker/src/main/java/io/moquette/spi/IMessagesStore.java
+2 -1
View File
@@ -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<String> notifyReceivers);
int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder message, Set<String> notifyReceivers, ProtoConstants.RequestSourceType requestSourceType);
Set<String> 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<Integer> contentTypes);
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+6
View File
@@ -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();