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

添加部分频道相关server api

Browse Source
This commit is contained in:
heavyrian2012
2026-05-10 14:30:59 +08:00
parent 776c9e26c6
commit 42281c3ad3
19 changed files with 956 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/persistence/DatabaseStore.java
+254 -1
View File
@@ -3917,7 +3917,6 @@ public class DatabaseStore {
return null;
}
void persistChannelListener(final String groupId, final List<String> memberList) {
mScheduler.execute(()->{
Connection connection = null;
@@ -4182,6 +4181,260 @@ public class DatabaseStore {
}
}
void batchInsertChannelListener(String channelId, List<String> listeners, boolean listen) throws SQLException {
LOG.info("updateChannelListener {}", channelId);
Connection connection = null;
PreparedStatement statement = null;
try {
connection = DBUtil.getConnection();
int size = 0;
String sql = listen
? "insert into t_channel_listener (_cid, _mid, _dt) values(?, ?, ?)"
: "delete from t_channel_listener where _cid=? and _mid=?";
statement = connection.prepareStatement(sql);
for (String listener : listeners) {
int index = 1;
statement.setString(index++, channelId);
statement.setString(index++, listener);
if (listen) {
statement.setLong(index++, System.currentTimeMillis());
}
statement.addBatch();
if(++size%100 == 0) {
statement.executeBatch();
statement.clearBatch();
}
}
if(size%100 != 0) {
statement.executeBatch();
}
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
throw e;
} finally {
DBUtil.closeDB(connection, statement);
}
}
public List<String> findMissingMids(String cid, List<String> midList, boolean contain) {
String tempTable = "temp_check_mids_" + System.currentTimeMillis();
try (Connection conn = DBUtil.getConnection()) {
// Step 1: 创建临时表（仅当前连接可见，自动清理）
String createSql = "CREATE TEMPORARY TABLE " + tempTable + " (_mid VARCHAR(64) NOT NULL PRIMARY KEY)";
try (Statement st = conn.createStatement()) {
st.execute(createSql);
}
// Step 2: 批量插入待检查的 mid（去重用 INSERT IGNORE）
String insertSql = "INSERT IGNORE INTO " + tempTable + " (_mid) VALUES (?)";
int count = 0;
try (PreparedStatement ps = conn.prepareStatement(insertSql)) {
for (String mid : midList) {
ps.setString(1, mid);
ps.addBatch();
if (++count % 1000 == 0) {
ps.executeBatch(); // 每1000条刷盘
ps.clearBatch();
}
}
ps.executeBatch();
}
// Step 3: 查询差集（NOT EXISTS 利用复合索引）
String querySql = "SELECT t._mid FROM " + tempTable + " t " +
"WHERE " + (contain?"":" NOT ") + " EXISTS (" +
" SELECT 1 FROM t_channel_listener c " +
" WHERE c._cid = ? AND c._mid = t._mid" +
")";
List<String> missing = new ArrayList<>();
try (PreparedStatement ps = conn.prepareStatement(querySql)) {
ps.setString(1, cid);
try (ResultSet rs = ps.executeQuery()) {
while (rs.next()) {
missing.add(rs.getString(1));
}
}
}
return missing;
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
List<WFCMessage.ChannelInfo> getChannelInfoList(int count, int offset, boolean withDeleted) {
List<WFCMessage.ChannelInfo> out = new ArrayList<>();
Connection connection = null;
PreparedStatement statement = null;
ResultSet rs = null;
try {
connection = DBUtil.getConnection();
String sql = "select _cid, _name" +
", _portrait" +
", _owner" +
", _status" +
", _desc" +
", _extra" +
", _dt from t_channel";
if(!withDeleted) {
sql += " where _status <> 64 ";
}
sql += " order by _dt desc ";
sql += " limit " + count;
if (offset > 0) {
sql += " offset " + offset;
}
statement = connection.prepareStatement(sql);
rs = statement.executeQuery();
while (rs.next()) {
WFCMessage.ChannelInfo.Builder builder = WFCMessage.ChannelInfo.newBuilder();
int index = 1;
String value = rs.getString(index++);
value = (value == null ? "" : value);
builder.setTargetId(value);
value = rs.getString(index++);
value = (value == null ? "" : value);
builder.setName(value);
value = rs.getString(index++);
value = (value == null ? "" : value);
builder.setPortrait(value);
value = rs.getString(index++);
value = (value == null ? "" : value);
builder.setOwner(value);
int status = rs.getInt(index++);
builder.setStatus(status);
value = rs.getString(index++);
value = (value == null ? "" : value);
builder.setDesc(value);
value = rs.getString(index++);
value = (value == null ? "" : value);
builder.setExtra(value);
long longValue = rs.getLong(index++);
builder.setUpdateDt(longValue);
out.add(builder.build());
}
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
} finally {
DBUtil.closeDB(connection, statement, rs);
}
return out;
}
int getChannelTotalCount(boolean withDeleted) {
Connection connection = null;
PreparedStatement statement = null;
ResultSet rs = null;
try {
connection = DBUtil.getConnection();
String sql = "select count(*) from t_channel";
if(!withDeleted) {
sql += " where _status <> 64";
}
statement = connection.prepareStatement(sql);
rs = statement.executeQuery();
if (rs.next()) {
return rs.getInt(1);
}
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
} finally {
DBUtil.closeDB(connection, statement, rs);
}
return 0;
}
public int getChannelListenerCount(String channelId) {
Connection connection = null;
PreparedStatement statement = null;
ResultSet rs = null;
try {
connection = DBUtil.getConnection();
String sql = "select count(*) from t_channel_listener where _cid = ?";
statement = connection.prepareStatement(sql);
int index = 1;
statement.setString(index++, channelId);
rs = statement.executeQuery();
if (rs.next()) {
return rs.getInt(1);
}
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
} finally {
DBUtil.closeDB(connection, statement, rs);
}
return 0;
}
public List<String> getChannelListenerList(String channelId, int count, int offset) {
List<String> out = new ArrayList<>();
Connection connection = null;
PreparedStatement statement = null;
ResultSet rs = null;
try {
connection = DBUtil.getConnection();
String sql = "select _mid from t_channel_listener where _cid = ?";
if(count > 0) {
sql += " limit " + count;
}
if(offset > 0) {
sql += " offset " + offset;
}
statement = connection.prepareStatement(sql);
int index = 1;
statement.setString(index++, channelId);
rs = statement.executeQuery();
while (rs.next()) {
String value = rs.getString(1);
if (!StringUtil.isNullOrEmpty(value)) {
out.add(value);
}
}
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
Utility.printExecption(LOG, e, RDBS_Exception);
} finally {
DBUtil.closeDB(connection, statement, rs);
}
return out;
}
public Set<String> getSensitiveWord() {
Connection connection = null;
PreparedStatement statement = null;
broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
+79
View File
@@ -48,6 +48,7 @@ import win.liyufan.im.*;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
@@ -78,6 +79,7 @@ import static io.moquette.BrokerConstants.*;
import static io.moquette.server.Constants.MAX_CHATROOM_MESSAGE_QUEUE;
import static io.moquette.server.Constants.MAX_MESSAGE_QUEUE;
import static cn.wildfirechat.pojos.MyInfoType.*;
import static win.liyufan.im.IMTopic.PutUserSettingTopic;
import static win.liyufan.im.UserSettingScope.*;
public class MemoryMessagesStore implements IMessagesStore {
@@ -1341,6 +1343,20 @@ public class MemoryMessagesStore implements IMessagesStore {
return false;
}
@Override
public ErrorCode setChatroomState(String chatroomId, int status) {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
IMap<String, WFCMessage.ChatroomInfo> chatroomInfoMap = hzInstance.getMap(CHATROOMS);
WFCMessage.ChatroomInfo info = chatroomInfoMap.get(chatroomId);
if (info == null) {
return ErrorCode.ERROR_CODE_NOT_EXIST;
}
info = info.toBuilder().setState(status).build();
chatroomInfoMap.set(chatroomId, info);
return ErrorCode.ERROR_CODE_SUCCESS;
}
@Override
public long insertChatroomMessages(String target, int line, long messageId) {
// messageId是全局的，messageSeq是跟个人相关的，理论上messageId的增长数度远远大于seq。
@@ -2325,6 +2341,19 @@ public class MemoryMessagesStore implements IMessagesStore {
return out;
}
@Override
public List<WFCMessage.GroupInfo> getGroupInfos(List<WFCMessage.UserRequest> requests) {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
IMap<String, WFCMessage.GroupInfo> mIMap = hzInstance.getMap(GROUPS_MAP);
ArrayList<WFCMessage.GroupInfo> out = new ArrayList<>();
for (WFCMessage.UserRequest request : requests) {
WFCMessage.GroupInfo groupInfo = mIMap.get(request.getUid());
out.add(groupInfo);
}
return out;
}
@Override
public WFCMessage.GroupInfo getGroupInfo(String groupId) {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
@@ -4925,6 +4954,56 @@ public class MemoryMessagesStore implements IMessagesStore {
return getChannelListener(channelId);
}
@Override
public List<WFCMessage.ChannelInfo> getChannelInfoList(int count, int offset, boolean withDeleted) {
return databaseStore.getChannelInfoList(count, offset, withDeleted);
}
@Override
public int getChannelTotalCount(boolean withDeleted) {
return databaseStore.getChannelTotalCount(withDeleted);
}
@Override
public ErrorCode batchListenChannel(String channelId, List<String> userIds, boolean listen) {
IMap<String, WFCMessage.ChannelInfo> mIMap = m_Server.getHazelcastInstance().getMap(CHANNELS);
WFCMessage.ChannelInfo channelData = mIMap.get(channelId);
if(channelData == null || (channelData.getStatus() & Channel_State_Mask_Deleted) > 0) {
return ErrorCode.ERROR_CODE_NOT_EXIST;
}
List<String> missedIds = databaseStore.findMissingMids(channelId, userIds, !listen);
try {
databaseStore.batchInsertChannelListener(channelId, missedIds, listen);
} catch (SQLException e) {
throw new RuntimeException(e);
}
new Thread(() -> {
for (String missedId : missedIds) {
WFCMessage.ModifyUserSettingReq modifyUserSettingReq = WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingListenedChannels).setKey(channelId).setValue(listen? "1" : "0").build();
m_Server.onApiMessage(missedId, null, modifyUserSettingReq.toByteArray(), 0, missedId, PutUserSettingTopic, ProtoConstants.RequestSourceType.Request_From_User);
try {
Thread.sleep(50);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}).start();
return ErrorCode.ERROR_CODE_SUCCESS;
}
@Override
public int getChannelListenerCount(String channelId) {
return databaseStore.getChannelListenerCount(channelId);
}
@Override
public List<String> getChannelListenerList(String channelId, int count, int offset) {
return databaseStore.getChannelListenerList(channelId, count, offset);
}
@Override
public int getOnlineUserCount() {
return m_Server.getProcessor().getConnectionDescriptors().getActiveConnectionsNo();
broker/src/main/java/io/moquette/spi/IMessagesStore.java
+8
View File
@@ -113,6 +113,7 @@ public interface IMessagesStore {
ErrorCode modifyGroupMemberAlias(String operator, String groupId, String alias, String memberId, boolean isAdmin);
ErrorCode modifyGroupMemberExtra(String operator, String groupId, String extra, String memberId, boolean isAdmin);
List<WFCMessage.GroupInfo> getGroupInfos(List<WFCMessage.UserRequest> requests, String fromUser, boolean isAdmin);
List<WFCMessage.GroupInfo> getGroupInfos(List<WFCMessage.UserRequest> requests);
WFCMessage.GroupInfo getGroupInfo(String groupId);
Set<String> getUserGroupIds(String userId);
Set<String> getUserGroupIds(String userId, List<Integer> types);
@@ -170,6 +171,7 @@ public interface IMessagesStore {
int getChatroomMemberCount(String chatroomId);
Collection<String> getChatroomMemberClient(String userId);
boolean checkUserClientInChatroom(String user, String clientId, String chatroomId);
ErrorCode setChatroomState(String chatroomId, int status);
long insertChatroomMessages(String target, int line, long messageId);
Collection<UserClientEntry> getChatroomMembers(String chatroomId);
@@ -232,6 +234,12 @@ public interface IMessagesStore {
boolean canSendMessageInChannel(String user, String channelId);
boolean checkUserInChannel(String user, String channelId);
Collection<String> getChannelSubscriber(String channelId);
List<WFCMessage.ChannelInfo> getChannelInfoList(int count, int offset, boolean withDeleted);
int getChannelTotalCount(boolean withDeleted);
ErrorCode batchListenChannel(String channelId, List<String> userIds, boolean listen);
int getChannelListenerCount(String channelId);
List<String> getChannelListenerList(String channelId, int count, int offset);
int getOnlineUserCount();
GetOnlineUserResult getOnlineUsers(int offset, int count);