diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ChannelSubscriberBatchAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ChannelSubscriberBatchAction.java new file mode 100644 index 00000000..84e93453 --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ChannelSubscriberBatchAction.java @@ -0,0 +1,41 @@ +package com.xiaoleilu.loServer.action.admin; + + +import cn.wildfirechat.common.APIPath; +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.InputSubscribeChannelBatch; +import cn.wildfirechat.proto.ProtoConstants; +import cn.wildfirechat.proto.WFCMessage; + +import com.hazelcast.util.StringUtil; +import com.xiaoleilu.loServer.RestResult; +import com.xiaoleilu.loServer.annotation.HttpMethod; +import com.xiaoleilu.loServer.annotation.Route; +import com.xiaoleilu.loServer.handler.Request; +import com.xiaoleilu.loServer.handler.Response; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; + +import java.util.concurrent.Executor; + + +@Route(APIPath.Batch_Subscribe_Channel) +@HttpMethod("POST") +public class ChannelSubscriberBatchAction extends AdminAction { + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + InputSubscribeChannelBatch input = getRequestBody(request.getNettyRequest(), InputSubscribeChannelBatch.class); + if (input != null && !StringUtil.isNullOrEmpty(input.getChannelId()) && input.getUserIds() != null && !input.getUserIds().isEmpty() ) { + ErrorCode errorCode = messagesStore.batchListenChannel(input.getChannelId(), input.getUserIds(), input.getSubscribe()>0); + sendResponse(response, errorCode, null); + } else { + response.setStatus(HttpResponseStatus.OK); + RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); + response.setContent(gson.toJson(result)); + } + } + return true; + } +} diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetBatchGroupInfoAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetBatchGroupInfoAction.java new file mode 100755 index 00000000..7a7038ef --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetBatchGroupInfoAction.java @@ -0,0 +1,65 @@ +/* + * This file is part of the Wildfire Chat package. + * (c) Heavyrain2012 + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +package com.xiaoleilu.loServer.action.admin; + +import cn.wildfirechat.common.APIPath; +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.PojoGroupInfo; +import cn.wildfirechat.pojos.PojoGroupInfoList; +import cn.wildfirechat.proto.WFCMessage; +import com.xiaoleilu.loServer.RestResult; +import com.xiaoleilu.loServer.annotation.HttpMethod; +import com.xiaoleilu.loServer.annotation.Route; +import com.xiaoleilu.loServer.handler.Request; +import com.xiaoleilu.loServer.handler.Response; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; + +import java.util.ArrayList; +import java.util.List; + +@Route(APIPath.Group_Batch_Info) +@HttpMethod("POST") +public class GetBatchGroupInfoAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + List inputGetGroup = getRequestBody(request.getNettyRequest(), ArrayList.class); + if (inputGetGroup != null && !inputGetGroup.isEmpty()) { + List requests = new ArrayList<>(); + for (String s : inputGetGroup) { + requests.add(WFCMessage.UserRequest.newBuilder().setUid(s).build()); + } + List groupInfoDataList = messagesStore.getGroupInfos(requests); + PojoGroupInfoList pojoGroupInfoList = new PojoGroupInfoList(); + pojoGroupInfoList.groupInfoList = new ArrayList<>(); + + for (WFCMessage.GroupInfo groupInfo : groupInfoDataList) { + pojoGroupInfoList.groupInfoList.add(PojoGroupInfo.fromProto(groupInfo)); + } + + response.setStatus(HttpResponseStatus.OK); + RestResult result = RestResult.ok(pojoGroupInfoList); + response.setContent(gson.toJson(result)); + } else { + response.setStatus(HttpResponseStatus.OK); + RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); + response.setContent(gson.toJson(result)); + } + + } + return true; + } +} diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetChannelListAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetChannelListAction.java new file mode 100755 index 00000000..434e9323 --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetChannelListAction.java @@ -0,0 +1,57 @@ +/* + * This file is part of the Wildfire Chat package. + * (c) Heavyrain2012 + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +package com.xiaoleilu.loServer.action.admin; + +import cn.wildfirechat.common.APIPath; +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.InputCountOffset; +import cn.wildfirechat.pojos.OutputChannelInfoList; +import cn.wildfirechat.pojos.OutputGetChannelInfo; +import cn.wildfirechat.proto.WFCMessage; +import com.xiaoleilu.loServer.annotation.HttpMethod; +import com.xiaoleilu.loServer.annotation.Route; +import com.xiaoleilu.loServer.handler.Request; +import com.xiaoleilu.loServer.handler.Response; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.util.internal.StringUtil; + +import java.util.ArrayList; +import java.util.List; + +@Route(APIPath.List_Channel_Info) +@HttpMethod("POST") +public class GetChannelListAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + InputCountOffset inputCreateChannel = getRequestBody(request.getNettyRequest(), InputCountOffset.class); + if (inputCreateChannel != null + && inputCreateChannel.count > 0 && inputCreateChannel.offset >= 0) { + int count = messagesStore.getChannelTotalCount(false); + List list = messagesStore.getChannelInfoList(inputCreateChannel.count, inputCreateChannel.offset, false); + OutputChannelInfoList outputChannelInfoList = new OutputChannelInfoList(); + outputChannelInfoList.total = count; + outputChannelInfoList.list = new ArrayList<>(); + for (WFCMessage.ChannelInfo channelData : list) { + outputChannelInfoList.list.add(OutputGetChannelInfo.fromPbInfo(channelData)); + } + sendResponse(response, ErrorCode.ERROR_CODE_SUCCESS, outputChannelInfoList); + } else { + sendResponse(response, ErrorCode.INVALID_PARAMETER, null); + } + } + return true; + } +} diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetGroupInfoAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetGroupInfoAction.java index 08e1d9c6..7910acae 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetGroupInfoAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetGroupInfoAction.java @@ -46,25 +46,7 @@ public class GetGroupInfoAction extends AdminAction { if (groupInfo == null) { result = RestResult.resultOf(ErrorCode.ERROR_CODE_NOT_EXIST); } else { - PojoGroupInfo pojoGroupInfo = new PojoGroupInfo(); - pojoGroupInfo.setExtra(groupInfo.getExtra()); - pojoGroupInfo.setName(groupInfo.getName()); - pojoGroupInfo.setOwner(groupInfo.getOwner()); - pojoGroupInfo.setPortrait(groupInfo.getPortrait()); - pojoGroupInfo.setTarget_id(groupInfo.getTargetId()); - pojoGroupInfo.setType(groupInfo.getType()); - pojoGroupInfo.setMember_count(groupInfo.getMemberCount()); - pojoGroupInfo.setMute(groupInfo.getMute()); - pojoGroupInfo.setJoin_type(groupInfo.getJoinType()); - pojoGroupInfo.setPrivate_chat(groupInfo.getPrivateChat()); - pojoGroupInfo.setSearchable(groupInfo.getSearchable()); - pojoGroupInfo.setMax_member_count(groupInfo.getMemberCount()); - pojoGroupInfo.setHistory_message(groupInfo.getHistoryMessage()); - pojoGroupInfo.setSuper_group(groupInfo.getSuperGroup()>0); - pojoGroupInfo.setDeleted(groupInfo.getDeleted()>0); - pojoGroupInfo.setUpdate_dt(groupInfo.getUpdateDt()); - pojoGroupInfo.setMember_update_dt(groupInfo.getMemberUpdateDt()); - result = RestResult.ok(pojoGroupInfo); + result = RestResult.ok(PojoGroupInfo.fromProto(groupInfo)); } setResponseContent(result, response); } else { diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ListChannelSubscriberAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ListChannelSubscriberAction.java new file mode 100644 index 00000000..1cef7753 --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/ListChannelSubscriberAction.java @@ -0,0 +1,49 @@ +package com.xiaoleilu.loServer.action.admin; + + +import cn.wildfirechat.common.APIPath; +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.InputStringCountOffset; +import cn.wildfirechat.pojos.InputSubscribeChannel; +import cn.wildfirechat.pojos.OutputBooleanValue; +import cn.wildfirechat.pojos.PojoTotalList; +import com.xiaoleilu.loServer.RestResult; +import com.xiaoleilu.loServer.annotation.HttpMethod; +import com.xiaoleilu.loServer.annotation.Route; +import com.xiaoleilu.loServer.handler.Request; +import com.xiaoleilu.loServer.handler.Response; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.util.internal.StringUtil; + +import java.util.List; + +@Route(APIPath.List_Channel_Subscriber) +@HttpMethod("POST") +public class ListChannelSubscriberAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + InputStringCountOffset inputSubscribeChannel = getRequestBody(request.getNettyRequest(), InputStringCountOffset.class); + response.setStatus(HttpResponseStatus.OK); + if (inputSubscribeChannel != null + && !StringUtil.isNullOrEmpty(inputSubscribeChannel.target) + && inputSubscribeChannel.count > 0) { + int count = messagesStore.getChannelListenerCount(inputSubscribeChannel.target); + List list = messagesStore.getChannelListenerList(inputSubscribeChannel.target, inputSubscribeChannel.count, inputSubscribeChannel.offset); + RestResult result = RestResult.ok(new PojoTotalList(count, list)); + response.setContent(gson.toJson(result)); + } else { + RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); + response.setContent(gson.toJson(result)); + } + } + return true; + } +} diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SetChatroomMuteAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SetChatroomMuteAction.java new file mode 100755 index 00000000..f948fa19 --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/SetChatroomMuteAction.java @@ -0,0 +1,53 @@ +/* + * This file is part of the Wildfire Chat package. + * (c) Heavyrain2012 + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +package com.xiaoleilu.loServer.action.admin; + +import cn.wildfirechat.common.APIPath; +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.InputChatroomMute; + +import com.xiaoleilu.loServer.RestResult; +import com.xiaoleilu.loServer.annotation.HttpMethod; +import com.xiaoleilu.loServer.annotation.Route; +import com.xiaoleilu.loServer.handler.Request; +import com.xiaoleilu.loServer.handler.Response; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.util.internal.StringUtil; + +@Route(APIPath.Chatroom_MuteAll) +@HttpMethod("POST") +public class SetChatroomMuteAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + InputChatroomMute input = getRequestBody(request.getNettyRequest(), InputChatroomMute.class); + if (input != null + && !StringUtil.isNullOrEmpty(input.getChatroomId())) { + + ErrorCode code = messagesStore.setChatroomState(input.getChatroomId(), input.getStatus()); + response.setStatus(HttpResponseStatus.OK); + response.setContent(gson.toJson(RestResult.resultOf(code))); + sendResponse(response, ErrorCode.ERROR_CODE_SUCCESS, null); + } else { + response.setStatus(HttpResponseStatus.OK); + RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); + response.setContent(gson.toJson(result)); + } + + } + return true; + } +} diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/UserSettingGetAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/UserSettingGetAction.java new file mode 100755 index 00000000..b5b7c040 --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/UserSettingGetAction.java @@ -0,0 +1,57 @@ +/* + * This file is part of the Wildfire Chat package. + * (c) Heavyrain2012 + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +package com.xiaoleilu.loServer.action.admin; + +import cn.wildfirechat.common.APIPath; +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.UserSettingPojo; +import cn.wildfirechat.proto.WFCMessage; +import com.xiaoleilu.loServer.RestResult; +import com.xiaoleilu.loServer.annotation.HttpMethod; +import com.xiaoleilu.loServer.annotation.Route; +import com.xiaoleilu.loServer.handler.Request; +import com.xiaoleilu.loServer.handler.Response; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.util.internal.StringUtil; + +@Route(APIPath.User_Get_Setting) +@HttpMethod("POST") +public class UserSettingGetAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + UserSettingPojo input = getRequestBody(request.getNettyRequest(), UserSettingPojo.class); + if (input != null && (!StringUtil.isNullOrEmpty(input.getUserId()))) { + WFCMessage.UserSettingEntry entry = messagesStore.getUserSetting(input.getUserId(), input.getScope(), input.getKey()); + response.setStatus(HttpResponseStatus.OK); + RestResult result; + if (entry == null) { + result = RestResult.resultOf(ErrorCode.ERROR_CODE_NOT_EXIST); + } else { + input.setValue(entry.getValue()); + result = RestResult.ok(input); + } + response.setContent(gson.toJson(result)); + } else { + response.setStatus(HttpResponseStatus.OK); + RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); + response.setContent(gson.toJson(result)); + } + + } + return true; + } +} diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/UserSettingPutAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/UserSettingPutAction.java new file mode 100755 index 00000000..be000689 --- /dev/null +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/UserSettingPutAction.java @@ -0,0 +1,60 @@ +/* + * This file is part of the Wildfire Chat package. + * (c) Heavyrain2012 + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +package com.xiaoleilu.loServer.action.admin; + +import cn.wildfirechat.common.APIPath; +import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.pojos.UserSettingPojo; +import cn.wildfirechat.proto.ProtoConstants; +import cn.wildfirechat.proto.WFCMessage; +import com.xiaoleilu.loServer.RestResult; +import com.xiaoleilu.loServer.annotation.HttpMethod; +import com.xiaoleilu.loServer.annotation.Route; +import com.xiaoleilu.loServer.handler.Request; +import com.xiaoleilu.loServer.handler.Response; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.util.internal.StringUtil; + +import static win.liyufan.im.IMTopic.PutUserSettingTopic; +import static win.liyufan.im.UserSettingScope.kUserSettingMyChannels; + +@Route(APIPath.User_Put_Setting) +@HttpMethod("POST") +public class UserSettingPutAction extends AdminAction { + + @Override + public boolean isTransactionAction() { + return true; + } + + @Override + public boolean action(Request request, Response response) { + if (request.getNettyRequest() instanceof FullHttpRequest) { + UserSettingPojo userSettingPojo = getRequestBody(request.getNettyRequest(), UserSettingPojo.class); + if (userSettingPojo != null && !StringUtil.isNullOrEmpty(userSettingPojo.getUserId())) { + WFCMessage.ModifyUserSettingReq modifyUserSettingReq = WFCMessage.ModifyUserSettingReq.newBuilder().setScope(userSettingPojo.getScope()).setKey(userSettingPojo.getKey()).setValue(userSettingPojo.getValue()).build(); + sendApiMessage(response, userSettingPojo.getUserId(), PutUserSettingTopic, modifyUserSettingReq.toByteArray(), result -> { + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeBytes(result); + ErrorCode errorCode = ErrorCode.fromCode(byteBuf.readByte()); + return new Result(errorCode); + }); + return false; + } else { + response.setStatus(HttpResponseStatus.OK); + RestResult result = RestResult.resultOf(ErrorCode.INVALID_PARAMETER); + response.setContent(gson.toJson(result)); + } + } + return true; + } +} diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index ec647cd3..89e17185 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -3917,7 +3917,6 @@ public class DatabaseStore { return null; } - void persistChannelListener(final String groupId, final List memberList) { mScheduler.execute(()->{ Connection connection = null; @@ -4182,6 +4181,260 @@ public class DatabaseStore { } } + void batchInsertChannelListener(String channelId, List 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 findMissingMids(String cid, List 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 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 getChannelInfoList(int count, int offset, boolean withDeleted) { + List 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 getChannelListenerList(String channelId, int count, int offset) { + List 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 getSensitiveWord() { Connection connection = null; PreparedStatement statement = null; diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 67c31d0d..d1cc6fde 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -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 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 getGroupInfos(List requests) { + HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); + IMap mIMap = hzInstance.getMap(GROUPS_MAP); + ArrayList 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 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 userIds, boolean listen) { + IMap 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 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 getChannelListenerList(String channelId, int count, int offset) { + return databaseStore.getChannelListenerList(channelId, count, offset); + } + @Override public int getOnlineUserCount() { return m_Server.getProcessor().getConnectionDescriptors().getActiveConnectionsNo(); diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index eed1e5ab..28010bdc 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -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 getGroupInfos(List requests, String fromUser, boolean isAdmin); + List getGroupInfos(List requests); WFCMessage.GroupInfo getGroupInfo(String groupId); Set getUserGroupIds(String userId); Set getUserGroupIds(String userId, List types); @@ -170,6 +171,7 @@ public interface IMessagesStore { int getChatroomMemberCount(String chatroomId); Collection 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 getChatroomMembers(String chatroomId); @@ -232,6 +234,12 @@ public interface IMessagesStore { boolean canSendMessageInChannel(String user, String channelId); boolean checkUserInChannel(String user, String channelId); Collection getChannelSubscriber(String channelId); + List getChannelInfoList(int count, int offset, boolean withDeleted); + int getChannelTotalCount(boolean withDeleted); + ErrorCode batchListenChannel(String channelId, List userIds, boolean listen); + int getChannelListenerCount(String channelId); + List getChannelListenerList(String channelId, int count, int offset); + int getOnlineUserCount(); GetOnlineUserResult getOnlineUsers(int offset, int count); diff --git a/common/src/main/java/cn/wildfirechat/common/APIPath.java b/common/src/main/java/cn/wildfirechat/common/APIPath.java index 7dbf985e..3095f91f 100644 --- a/common/src/main/java/cn/wildfirechat/common/APIPath.java +++ b/common/src/main/java/cn/wildfirechat/common/APIPath.java @@ -120,8 +120,11 @@ public interface APIPath { String Create_Channel = "/admin/channel/create"; String Destroy_Channel = "/admin/channel/destroy"; String Get_Channel_Info = "/admin/channel/get"; + String List_Channel_Info = "/admin/channel/list"; String Subscribe_Channel = "/admin/channel/subscribe"; + String Batch_Subscribe_Channel = "/admin/channel/batch_subscribe"; String Check_User_Subscribe_Channel = "/admin/channel/is_subscribed"; + String List_Channel_Subscriber = "/admin/channel/list_subscriber"; String Get_System_Setting = "/admin/system/get_setting"; String Put_System_Setting = "/admin/system/put_setting"; String GET_CUSTOMER = "/admin/customer"; diff --git a/common/src/main/java/cn/wildfirechat/pojos/InputStringCountOffset.java b/common/src/main/java/cn/wildfirechat/pojos/InputStringCountOffset.java new file mode 100644 index 00000000..e7fe897f --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/InputStringCountOffset.java @@ -0,0 +1,16 @@ +package cn.wildfirechat.pojos; + +public class InputStringCountOffset { + public String target; + public int count; + public int offset; + + public InputStringCountOffset() { + } + + public InputStringCountOffset(String target, int count, int offset) { + this.target = target; + this.count = count; + this.offset = offset; + } +} diff --git a/common/src/main/java/cn/wildfirechat/pojos/InputSubscribeChannelBatch.java b/common/src/main/java/cn/wildfirechat/pojos/InputSubscribeChannelBatch.java new file mode 100644 index 00000000..29614362 --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/InputSubscribeChannelBatch.java @@ -0,0 +1,50 @@ +/* + * This file is part of the Wildfire Chat package. + * (c) Heavyrain2012 + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +package cn.wildfirechat.pojos; + +import java.util.List; + +public class InputSubscribeChannelBatch { + private String channelId; + private List userIds; + private int subscribe; //1，订阅；0，取消订阅 + + public InputSubscribeChannelBatch() { + } + + public InputSubscribeChannelBatch(String channelId, List userIds, int subscribe) { + this.channelId = channelId; + this.userIds = userIds; + this.subscribe = subscribe; + } + + public String getChannelId() { + return channelId; + } + + public void setChannelId(String channelId) { + this.channelId = channelId; + } + + public List getUserIds() { + return userIds; + } + + public void setUserIds(List userIds) { + this.userIds = userIds; + } + + public int getSubscribe() { + return subscribe; + } + + public void setSubscribe(int subscribe) { + this.subscribe = subscribe; + } +} diff --git a/common/src/main/java/cn/wildfirechat/pojos/OutputChannelInfoList.java b/common/src/main/java/cn/wildfirechat/pojos/OutputChannelInfoList.java new file mode 100755 index 00000000..908bd6f1 --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/OutputChannelInfoList.java @@ -0,0 +1,21 @@ +/* + * This file is part of the Wildfire Chat package. + * (c) Heavyrain2012 + * + * For the full copyright and license information, please view the LICENSE + * file that was distributed with this source code. + */ + +package cn.wildfirechat.pojos; + + +import cn.wildfirechat.proto.WFCMessage; + +import java.util.ArrayList; +import java.util.List; + +public class OutputChannelInfoList { + public List list; + public int total; +} + diff --git a/common/src/main/java/cn/wildfirechat/pojos/PojoGroupInfo.java b/common/src/main/java/cn/wildfirechat/pojos/PojoGroupInfo.java index 99dfef40..cca89129 100644 --- a/common/src/main/java/cn/wildfirechat/pojos/PojoGroupInfo.java +++ b/common/src/main/java/cn/wildfirechat/pojos/PojoGroupInfo.java @@ -8,6 +8,8 @@ package cn.wildfirechat.pojos; +import cn.wildfirechat.proto.WFCMessage; + /** * 群组信息类 *

@@ -175,4 +177,26 @@ public class PojoGroupInfo { public void setMember_update_dt(long member_update_dt) { this.member_update_dt = member_update_dt; } + + static public PojoGroupInfo fromProto(WFCMessage.GroupInfo groupInfo) { + PojoGroupInfo pojoGroupInfo = new PojoGroupInfo(); + pojoGroupInfo.setExtra(groupInfo.getExtra()); + pojoGroupInfo.setName(groupInfo.getName()); + pojoGroupInfo.setOwner(groupInfo.getOwner()); + pojoGroupInfo.setPortrait(groupInfo.getPortrait()); + pojoGroupInfo.setTarget_id(groupInfo.getTargetId()); + pojoGroupInfo.setType(groupInfo.getType()); + pojoGroupInfo.setMember_count(groupInfo.getMemberCount()); + pojoGroupInfo.setMute(groupInfo.getMute()); + pojoGroupInfo.setJoin_type(groupInfo.getJoinType()); + pojoGroupInfo.setPrivate_chat(groupInfo.getPrivateChat()); + pojoGroupInfo.setSearchable(groupInfo.getSearchable()); + pojoGroupInfo.setMax_member_count(groupInfo.getMemberCount()); + pojoGroupInfo.setHistory_message(groupInfo.getHistoryMessage()); + pojoGroupInfo.setSuper_group(groupInfo.getSuperGroup()>0); + pojoGroupInfo.setDeleted(groupInfo.getDeleted()>0); + pojoGroupInfo.setUpdate_dt(groupInfo.getUpdateDt()); + pojoGroupInfo.setMember_update_dt(groupInfo.getMemberUpdateDt()); + return pojoGroupInfo; + } } diff --git a/common/src/main/java/cn/wildfirechat/pojos/PojoTotalList.java b/common/src/main/java/cn/wildfirechat/pojos/PojoTotalList.java new file mode 100644 index 00000000..cae44662 --- /dev/null +++ b/common/src/main/java/cn/wildfirechat/pojos/PojoTotalList.java @@ -0,0 +1,16 @@ +package cn.wildfirechat.pojos; + +import java.util.List; + +public class PojoTotalList { + public int total; + public List list; + + public PojoTotalList() { + } + + public PojoTotalList(int total, List list) { + this.total = total; + this.list = list; + } +} diff --git a/sdk/src/main/java/cn/wildfirechat/sdk/ChannelAdmin.java b/sdk/src/main/java/cn/wildfirechat/sdk/ChannelAdmin.java index e57bf00f..a4c0895b 100644 --- a/sdk/src/main/java/cn/wildfirechat/sdk/ChannelAdmin.java +++ b/sdk/src/main/java/cn/wildfirechat/sdk/ChannelAdmin.java @@ -2,9 +2,12 @@ package cn.wildfirechat.sdk; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.pojos.*; +import cn.wildfirechat.pojos.mesh.PojoStringList; import cn.wildfirechat.sdk.model.IMResult; import cn.wildfirechat.sdk.utilities.AdminHttpUtils; +import java.util.List; + /** * 频道管理类 *

@@ -53,6 +56,21 @@ public class ChannelAdmin { return AdminHttpUtils.httpJsonPost(path, inputChannelId, OutputGetChannelInfo.class); } + /** + * 获取频道信息列表 + * @param count 数量 + * @param offset offset + * @return 频道信息列表 + * @throws Exception 请求失败时抛出异常 + */ + public static IMResult getChannelInfoList(int count, int offset) throws Exception { + String path = APIPath.List_Channel_Info; + InputCountOffset inputCountOffset = new InputCountOffset(); + inputCountOffset.count = count; + inputCountOffset.offset = offset; + return AdminHttpUtils.httpJsonPost(path, inputCountOffset, OutputChannelInfoList.class); + } + /** * 订阅频道 * @param channelId 频道ID @@ -66,6 +84,19 @@ public class ChannelAdmin { return AdminHttpUtils.httpJsonPost(path, input, Void.class); } + /** + * 批量用户订阅频道 + * @param channelId 频道ID + * @param userIds 用户ID列表 + * @return 订阅结果 + * @throws Exception 请求失败时抛出异常 + */ + public static IMResult batchSubscribeChannel(String channelId, List userIds) throws Exception { + String path = APIPath.Batch_Subscribe_Channel; + InputSubscribeChannelBatch input = new InputSubscribeChannelBatch(channelId, userIds, 1); + return AdminHttpUtils.httpJsonPost(path, input, Void.class); + } + /** * 取消订阅频道 * @param channelId 频道ID @@ -79,6 +110,19 @@ public class ChannelAdmin { return AdminHttpUtils.httpJsonPost(path, input, Void.class); } + /** + * 批量用户取消订阅频道 + * @param channelId 频道ID + * @param userIds 用户ID列表 + * @return 取消订阅结果 + * @throws Exception 请求失败时抛出异常 + */ + public static IMResult batchUnsubscribeChannel(String channelId, List userIds) throws Exception { + String path = APIPath.Batch_Subscribe_Channel; + InputSubscribeChannelBatch input = new InputSubscribeChannelBatch(channelId, userIds, 0); + return AdminHttpUtils.httpJsonPost(path, input, Void.class); + } + /** * 检查用户是否订阅了频道 * @param userId 用户ID @@ -91,4 +135,18 @@ public class ChannelAdmin { InputSubscribeChannel input = new InputSubscribeChannel(channelId, userId, 0); return AdminHttpUtils.httpJsonPost(path, input, OutputBooleanValue.class); } + + /** + * 获取频道订阅用户列表 + * @param channelId 频道ID + * @param count + * @param offset + * @return 订阅用户id列表 + * @throws Exception 请求失败时抛出异常 + */ + public static IMResult getChannelSubscribers(String channelId, int count, int offset) throws Exception { + String path = APIPath.List_Channel_Subscriber; + InputStringCountOffset input = new InputStringCountOffset(channelId, count, offset); + return AdminHttpUtils.httpJsonPost(path, input, PojoTotalList.class); + } } diff --git a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java index 6a7b0817..d430de17 100644 --- a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java +++ b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java @@ -270,7 +270,7 @@ public class Main { System.out.println("get user info success"); } else { System.out.println("get user info by mobile failure"); - System.exit(-1); +// System.exit(-1); } } else { System.out.println("get user info by mobile failure"); @@ -1022,7 +1022,7 @@ public class Main { //验证群组已被解散 resultGetGroupInfo = GroupAdmin.getGroupInfo(groupInfo.getTarget_id()); - if (resultGetGroupInfo != null && resultGetGroupInfo.getErrorCode() == ErrorCode.ERROR_CODE_NOT_EXIST) { + if (resultGetGroupInfo != null && (resultGetGroupInfo.getErrorCode() == ErrorCode.ERROR_CODE_NOT_EXIST || resultGetGroupInfo.result.isDeleted())) { System.out.println("group dismissed verified"); } else { System.out.println("group dismiss verify failure"); @@ -1187,14 +1187,15 @@ public class Main { System.out.println("failure"); } - //测试删除多播消息 - IMResult deleteMultiCastResult = MessageAdmin.deleteMultiCastMessage("user1", resultMulticastMessage.result.getMessageUid(), multicastReceivers); - if (deleteMultiCastResult != null && deleteMultiCastResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { - System.out.println("delete multicast message success"); - } else { - System.out.println("delete multicast message failure"); + if (commercialServer) { + //测试删除多播消息 + IMResult deleteMultiCastResult = MessageAdmin.deleteMultiCastMessage("user1", resultMulticastMessage.result.getMessageUid(), multicastReceivers); + if (deleteMultiCastResult != null && deleteMultiCastResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { + System.out.println("delete multicast message success"); + } else { + System.out.println("delete multicast message failure"); + } } - } //*********************************************** @@ -1591,6 +1592,40 @@ public class Main { System.exit(-1); } + IMResult channelInfoListIMResult = ChannelAdmin.getChannelInfoList(10, 0); + if(channelInfoListIMResult != null && channelInfoListIMResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { + System.out.println("get channel list success"); + } else { + System.out.println("get channel list failure"); + System.exit(-1); + } + + voidIMResult = ChannelAdmin.batchSubscribeChannel(inputCreateChannel.getTargetId(), Arrays.asList("user1", "user2", "user3")); + if(voidIMResult != null && voidIMResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { + System.out.println("batch subscribe channel success"); + } else { + System.out.println("batch subscribe channel failure"); + System.exit(-1); + } + + + IMResult totalListIMResult = ChannelAdmin.getChannelSubscribers(inputCreateChannel.getTargetId(), 10, 1); + if(totalListIMResult != null && totalListIMResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { + System.out.println("get channel subscribe list success"); + } else { + System.out.println("get channel subscribe list failure"); + System.exit(-1); + } + + voidIMResult = ChannelAdmin.batchUnsubscribeChannel(inputCreateChannel.getTargetId(), Arrays.asList("user1", "user2", "user3")); + if(voidIMResult != null && voidIMResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { + System.out.println("batch unsubscribe channel success"); + } else { + System.out.println("batch unsubscribe channel failure"); + System.exit(-1); + } + + voidIMResult = ChannelAdmin.destroyChannel(inputCreateChannel.getTargetId()); if(voidIMResult != null && voidIMResult.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { System.out.println("success");