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

Load remote messages

Browse Source
This commit is contained in:
heavyrain lee
2019-05-01 23:47:28 +08:00
parent b47084f61f
commit ce16f87e12
2 changed files with 15 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/imhandler/LoadRemoteMessagesHandler.java
+5 -1
View File
@@ -21,7 +21,11 @@ public class LoadRemoteMessagesHandler extends IMHandler<WFCMessage.LoadRemoteMe
public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.LoadRemoteMessages request, Qos1PublishHandler.IMCallback callback) {
ErrorCode errorCode = ErrorCode.ERROR_CODE_SUCCESS;
WFCMessage.PullMessageResult result = m_messagesStore.loadRemoteMessages(fromUser, request.getConversation(), request.getBeforeUid(), request.getCount());
long beforeUid = request.getBeforeUid();
if (beforeUid == 0) {
beforeUid = Long.MAX_VALUE;
}
WFCMessage.PullMessageResult result = m_messagesStore.loadRemoteMessages(fromUser, request.getConversation(), beforeUid, request.getCount());
byte[] data = result.toByteArray();
LOG.info("User {} load message with count({}), payload size({})", fromUser, result.getMessageCount(), data.length);
ackPayload.ensureWritable(data.length).writeBytes(data);
broker/src/main/java/io/moquette/persistence/DatabaseStore.java
+10 -8
View File
@@ -537,23 +537,25 @@ public class DatabaseStore {
List<WFCMessage.Message> messages = loadRemoteMessagesFromTable(user, conversation, beforeUid, count, MessageShardingUtil.getMessageTable(beforeUid));
if (messages != null && messages.size() < count) {
String nexTable = MessageShardingUtil.getPreviousMessageTable(beforeUid);
List<WFCMessage.Message> nextMessages = loadRemoteMessagesFromTable(user, conversation, beforeUid, count-messages.size(), nexTable);
if (nextMessages != null) {
messages.addAll(nextMessages);
if (!StringUtil.isNullOrEmpty(nexTable)) {
List<WFCMessage.Message> nextMessages = loadRemoteMessagesFromTable(user, conversation, beforeUid, count - messages.size(), nexTable);
if (nextMessages != null) {
messages.addAll(nextMessages);
}
}
}
return messages;
}
List<WFCMessage.Message> loadRemoteMessagesFromTable(String user, WFCMessage.Conversation conversation, long beforeUid, int count, String table) {
String sql = "select `_from`, `_type`, `_target`, `_line`, `_data`, `_dt` from " + table +" where";
String sql = "select `_mid`, `_from`, `_type`, `_target`, `_line`, `_data`, `_dt` from " + table +" where";
if (conversation.getType() == ProtoConstants.ConversationType.ConversationType_Private) {
sql += " _type = ? and _line = ? and _mid > ? and ((_target = ? and _from = ?) or (_target = ? and _from = ?)";
sql += " _type = ? and _line = ? and _mid < ? and ((_target = ? and _from = ?) or (_target = ? and _from = ?)";
} else {
sql += " _type = ? and _line = ? and _mid > ? and _target = ?";
sql += " _type = ? and _line = ? and _mid < ? and _target = ?";
}
sql += " limit ?";
sql += "order by `_mid` DESC limit ?";
Connection connection = null;
PreparedStatement statement = null;
@@ -598,7 +600,7 @@ public class DatabaseStore {
} finally {
DBUtil.closeDB(connection, statement, resultSet);
}
return null;
return out;
}
void persistUserMessage(final String userId, final long messageId, final long messageSeq) {