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

异常报警功能

Browse Source
This commit is contained in:
heavyrain2012
2020-12-12 16:27:15 +08:00
parent 28b732788c
commit 31d69e78cf
16 changed files with 310 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+33 -4
View File
@@ -49,6 +49,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.BiFunction;
import static cn.wildfirechat.common.IMExceptionEvent.EventType.EVENT_CALLBACK_Exception;
import static cn.wildfirechat.proto.ProtoConstants.PersistFlag.Transparent;
public class MessagesPublisher {
@@ -176,7 +177,14 @@ public class MessagesPublisher {
message = m_messagesStore.getMessage(messageHead);
}
final WFCMessage.Message finalMsg = message;
Server.getServer().getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(robot.getCallback(), new Gson().toJson(SendMessageData.fromProtoMessage(finalMsg), SendMessageData.class)));
Server.getServer().getCallbackScheduler().execute(() -> {
try {
HttpUtils.httpJsonPost(robot.getCallback(), new Gson().toJson(SendMessageData.fromProtoMessage(finalMsg), SendMessageData.class));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e, EVENT_CALLBACK_Exception);
}
});
continue;
}
}
@@ -540,7 +548,14 @@ 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())) {
Server.getServer().getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(channelInfo.getCallback() + "/message", new Gson().toJson(SendMessageData.fromProtoMessage(message), SendMessageData.class)));
Server.getServer().getCallbackScheduler().execute(() -> {
try {
HttpUtils.httpJsonPost(channelInfo.getCallback() + "/message", new Gson().toJson(SendMessageData.fromProtoMessage(message), SendMessageData.class));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e, EVENT_CALLBACK_Exception);
}
});
}
}
long messageId = message.getMessageId();
@@ -580,7 +595,14 @@ public class MessagesPublisher {
}
public void forwardMessage(final WFCMessage.Message message, String forwardUrl) {
Server.getServer().getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(forwardUrl, new Gson().toJson(OutputMessageData.fromProtoMessage(message), OutputMessageData.class)));
Server.getServer().getCallbackScheduler().execute(() -> {
try {
HttpUtils.httpJsonPost(forwardUrl, new Gson().toJson(OutputMessageData.fromProtoMessage(message), OutputMessageData.class));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e, EVENT_CALLBACK_Exception);
}
});
}
public void notifyChannelListenStatusChanged(WFCMessage.ChannelInfo channelInfo, String user, boolean listen) {
@@ -588,6 +610,13 @@ public class MessagesPublisher {
return;
}
Server.getServer().getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(channelInfo.getCallback() + "/subscribe", new Gson().toJson(new OutputNotifyChannelSubscribeStatus(user, channelInfo.getTargetId(), listen), OutputNotifyChannelSubscribeStatus.class)));
Server.getServer().getCallbackScheduler().execute(() -> {
try {
HttpUtils.httpJsonPost(channelInfo.getCallback() + "/subscribe", new Gson().toJson(new OutputNotifyChannelSubscribeStatus(user, channelInfo.getTargetId(), listen), OutputNotifyChannelSubscribeStatus.class));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e, EVENT_CALLBACK_Exception);
}
});
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+9 -2
View File
@@ -17,7 +17,6 @@
package io.moquette.spi.impl;
import cn.wildfirechat.common.ErrorCode;
import cn.wildfirechat.pojos.SendMessageData;
import cn.wildfirechat.pojos.UserOnlineStatus;
import cn.wildfirechat.proto.ProtoConstants;
import cn.wildfirechat.proto.WFCMessage;
@@ -54,6 +53,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos;
import static cn.wildfirechat.common.IMExceptionEvent.EventType.EVENT_CALLBACK_Exception;
import static io.moquette.spi.impl.Utils.readBytesAndRewind;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
@@ -252,7 +252,14 @@ public class ProtocolProcessor {
public void forwardOnlineStatusEvent(String userId, String clientId, int platform, int status, String packageName) {
if (!StringUtil.isNullOrEmpty(mUserOnlineStatusCallback)) {
mServer.getCallbackScheduler().execute(() -> HttpUtils.httpJsonPost(mUserOnlineStatusCallback, new Gson().toJson(new UserOnlineStatus(userId, clientId, platform, status, packageName))));
mServer.getCallbackScheduler().execute(() -> {
try {
HttpUtils.httpJsonPost(mUserOnlineStatusCallback, new Gson().toJson(new UserOnlineStatus(userId, clientId, platform, status, packageName)));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e, EVENT_CALLBACK_Exception);
}
});
}
}