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

Add load history message function

Browse Source
This commit is contained in:
heavyrain lee
2019-04-19 08:15:27 +08:00
parent b82f7dea3d
commit 0f05d63a3b
7 changed files with 848 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/imhandler/LoadRemoteMessagesHandler.java
+30
View File
@@ -0,0 +1,30 @@
/*
* This file is part of the Wildfire Chat package.
* (c) Heavyrain2012 <[email protected]>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
package io.moquette.imhandler;
import cn.wildfirechat.proto.ProtoConstants;
import cn.wildfirechat.proto.WFCMessage;
import io.moquette.spi.impl.Qos1PublishHandler;
import io.netty.buffer.ByteBuf;
import win.liyufan.im.ErrorCode;
import win.liyufan.im.IMTopic;
@Handler(value = IMTopic.LoadRemoteMessagesTopic)
public class LoadRemoteMessagesHandler extends IMHandler<WFCMessage.LoadRemoteMessages> {
@Override
public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, 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());
byte[] data = result.toByteArray();
LOG.info("User {} load message with count({}), payload size({})", fromUser, result.getMessageCount(), data.length);
ackPayload.ensureWritable(data.length).writeBytes(data);
return errorCode;
}
}
broker/src/main/java/io/moquette/persistence/DatabaseStore.java
+69
View File
@@ -8,6 +8,7 @@
package io.moquette.persistence;
import cn.wildfirechat.proto.ProtoConstants;
import cn.wildfirechat.proto.WFCMessage;
import cn.wildfirechat.server.ThreadPoolExecutorWrapper;
import com.hazelcast.core.HazelcastInstance;
@@ -532,6 +533,74 @@ public class DatabaseStore {
return null;
}
List<WFCMessage.Message> loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count) {
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);
}
}
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";
if (conversation.getType() == ProtoConstants.ConversationType.ConversationType_Private) {
sql += " _type = ? and _line = ? and _mid > ? and ((_target = ? and _from = ?) or (_target = ? and _from = ?)";
} else {
sql += " _type = ? and _line = ? and _mid > ? and _target = ?";
}
sql += " limit ?";
Connection connection = null;
PreparedStatement statement = null;
ResultSet resultSet = null;
List<WFCMessage.Message> out = new ArrayList<>();
try {
connection = DBUtil.getConnection();
statement = connection.prepareStatement(sql);
int index = 1;
statement.setInt(index++, conversation.getType());
statement.setInt(index++, conversation.getLine());
statement.setLong(index++, beforeUid);
statement.setString(index++, conversation.getTarget());
if (conversation.getType() == ProtoConstants.ConversationType.ConversationType_Private) {
statement.setString(index++, user);
statement.setString(index++, user);
statement.setString(index++, conversation.getTarget());
}
statement.setInt(index++, count);
resultSet = statement.executeQuery();
while (resultSet.next()) {
WFCMessage.Message.Builder builder = WFCMessage.Message.newBuilder();
index = 1;
builder.setMessageId(resultSet.getLong(index++));
builder.setFromUser(resultSet.getString(index++));
WFCMessage.Conversation.Builder cb = WFCMessage.Conversation.newBuilder();
cb.setType(resultSet.getInt(index++));
cb.setTarget(resultSet.getString(index++));
cb.setLine(resultSet.getInt(index++));
builder.setConversation(cb.build());
Blob blob = resultSet.getBlob(index++);
WFCMessage.MessageContent messageContent = WFCMessage.MessageContent.parseFrom(blob.getBinaryStream());
builder.setContent(messageContent);
builder.setServerTimestamp(resultSet.getTimestamp(index++).getTime());
WFCMessage.Message message = builder.build();
out.add(message);
}
} catch (SQLException | IOException e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
} finally {
DBUtil.closeDB(connection, statement, resultSet);
}
return null;
}
void persistUserMessage(final String userId, final long messageId, final long messageSeq) {
mScheduler.execute(()->{
Connection connection = null;
broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
+11 -1
View File
@@ -155,7 +155,6 @@ public class MemoryMessagesStore implements IMessagesStore {
IMap<Long, MessageBundle> mIMap = hzInstance.getMap(MESSAGES_MAP);
MessageBundle messageBundle = new MessageBundle(message.getMessageId(), fromUser, fromClientId, message);
if (message.getContent().getPersistFlag() == Transparent) {
mIMap.put(message.getMessageId(), messageBundle, 10, TimeUnit.SECONDS);
@@ -347,6 +346,17 @@ public class MemoryMessagesStore implements IMessagesStore {
return builder.build();
}
@Override
public WFCMessage.PullMessageResult loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count) {
WFCMessage.PullMessageResult.Builder builder = WFCMessage.PullMessageResult.newBuilder();
List<WFCMessage.Message> messages = databaseStore.loadRemoteMessages(user, conversation, beforeUid, count);
builder.setCurrent(0).setHead(0);
if(messages != null) {
builder.addAllMessage(messages);
}
return builder.build();
}
@Override
public Collection<UserClientEntry> getChatroomMembers(String chatroomId) {
MultiMap<String, UserClientEntry> chatroomMembers = m_Server.getHazelcastInstance().getMultiMap(CHATROOM_MEMBER_IDS);
broker/src/main/java/io/moquette/spi/IMessagesStore.java
+1
View File
@@ -100,6 +100,7 @@ public interface IMessagesStore {
WFCMessage.Message storeMessage(String fromUser, String fromClientId, WFCMessage.Message message);
int getNotifyReceivers(String fromUser, WFCMessage.Message message, Set<String> notifyReceivers);
WFCMessage.PullMessageResult fetchMessage(String user, String exceptClientId, long fromMessageId, int pullType);
WFCMessage.PullMessageResult loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count);
long insertUserMessages(String sender, int conversationType, String target, int line, int messageContentType, String userId, long messageId);
WFCMessage.GroupInfo createGroup(String operator, WFCMessage.GroupInfo groupInfo, List<WFCMessage.GroupMember> memberList);
ErrorCode addGroupMembers(String operator, String groupId, List<WFCMessage.GroupMember> memberList);
broker/src/main/java/win/liyufan/im/IMTopic.java
+2
View File
@@ -67,4 +67,6 @@ public interface IMTopic {
String ChannelPullTopic = "CHP";
String GetTokenTopic = "GETTOKEN";
String LoadRemoteMessagesTopic = "LRM";
}
broker/src/main/java/win/liyufan/im/MessageShardingUtil.java
+18
View File
@@ -60,4 +60,22 @@ public class MessageShardingUtil {
year %= 3;
return "t_messages_" + (year * 12 + month);
}
public static String getPreviousMessageTable(long mid) {
if (DBUtil.IsEmbedDB) {
return null;
}
mid >>= (nodeIdWidth + rotateIdWidth);
Date date = new Date(mid + T201801010000);
int month = date.getMonth();
int year = date.getYear();
year %= 3;
month = month + 1;
if (month == 12) {
month = 0;
year = (year + 3 -1)%3;
}
return "t_messages_" + (year * 12 + month);
}
}