From 6a25a83857b05737338e89973a65d2d1aea1fd04 Mon Sep 17 00:00:00 2001 From: heavyrian2012 Date: Mon, 2 May 2022 11:05:56 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=A0=E9=99=A4=E7=94=A8=E6=88=B7=E6=88=96?= =?UTF-8?q?=E5=88=A0=E9=99=A4=E9=A2=91=E9=81=93=E6=97=B6=EF=BC=8C=E5=88=A0?= =?UTF-8?q?=E9=99=A4=E5=AF=B9=E4=BA=8E=E7=9A=84=E8=AE=A2=E9=98=85=E5=85=B3?= =?UTF-8?q?=E7=B3=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...andler.java => DestroyChannelHandler.java} | 4 +- .../moquette/persistence/DatabaseStore.java | 51 +++++++++++++++++++ .../persistence/MemoryMessagesStore.java | 6 ++- .../java/io/moquette/spi/IMessagesStore.java | 2 +- 4 files changed, 58 insertions(+), 5 deletions(-) rename broker/src/main/java/io/moquette/imhandler/{DistroyChannelHandler.java => DestroyChannelHandler.java} (92%) diff --git a/broker/src/main/java/io/moquette/imhandler/DistroyChannelHandler.java b/broker/src/main/java/io/moquette/imhandler/DestroyChannelHandler.java similarity index 92% rename from broker/src/main/java/io/moquette/imhandler/DistroyChannelHandler.java rename to broker/src/main/java/io/moquette/imhandler/DestroyChannelHandler.java index 2d9d8a63..326b65da 100644 --- a/broker/src/main/java/io/moquette/imhandler/DistroyChannelHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/DestroyChannelHandler.java @@ -19,11 +19,11 @@ import static win.liyufan.im.IMTopic.PutUserSettingTopic; import static win.liyufan.im.UserSettingScope.kUserSettingMyChannels; @Handler(value = IMTopic.DestroyChannelInfoTopic) -public class DistroyChannelHandler extends GroupHandler { +public class DestroyChannelHandler extends GroupHandler { @Override public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, ProtoConstants.RequestSourceType requestSourceType, WFCMessage.IDBuf request, Qos1PublishHandler.IMCallback callback) { boolean isAdmin = requestSourceType == ProtoConstants.RequestSourceType.Request_From_Admin; - ErrorCode errorCode = m_messagesStore.distoryChannel(fromUser, request.getId(), isAdmin); + ErrorCode errorCode = m_messagesStore.destroyChannel(fromUser, request.getId(), isAdmin); if (errorCode == ErrorCode.ERROR_CODE_SUCCESS) { WFCMessage.ModifyUserSettingReq modifyUserSettingReq = WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingMyChannels).setKey(request.getId()).setValue("0").build(); mServer.onApiMessage(fromUser, null, modifyUserSettingReq.toByteArray(), 0, fromUser, PutUserSettingTopic, requestSourceType); diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index 7176593c..93b3618d 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -3332,6 +3332,57 @@ public class DatabaseStore { }); } + List getUserChannels(String userId) { + List out = new ArrayList<>(); + + Connection connection = null; + PreparedStatement statement = null; + ResultSet rs = null; + try { + connection = DBUtil.getConnection(); + String sql = "select _cid from t_channel_listener where _mid = ?"; + statement = connection.prepareStatement(sql); + + int index = 1; + statement.setString(index++, userId); + + 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; + } + + void clearChannelListener(final String channelId) { + LOG.info("Database remove channel {}", channelId); + Connection connection = null; + PreparedStatement statement = null; + try { + connection = DBUtil.getConnection(); + String sql = "delete from t_channel_listener where _cid=?"; + statement = connection.prepareStatement(sql); + statement.setString(1, channelId); + int count = statement.executeUpdate(); + LOG.info("Update rows {}", count); + } catch (SQLException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + Utility.printExecption(LOG, e, RDBS_Exception); + } finally { + DBUtil.closeDB(connection, statement); + } + } + 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 7c69bef6..ed41c700 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -3651,7 +3651,7 @@ public class MemoryMessagesStore implements IMessagesStore { @Override public void clearUserChannels(String userId) { - + databaseStore.getUserChannels(userId).forEach(s -> listenChannel(userId, s, false)); } @@ -3731,7 +3731,7 @@ public class MemoryMessagesStore implements IMessagesStore { } @Override - public ErrorCode distoryChannel(String operator, String channelId, boolean isAdmin) { + public ErrorCode destroyChannel(String operator, String channelId, boolean isAdmin) { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); IMap mIMap = hzInstance.getMap(CHANNELS); @@ -3751,6 +3751,8 @@ public class MemoryMessagesStore implements IMessagesStore { newInfoBuilder.setStatus(oldInfo.getStatus() | Channel_State_Mask_Deleted); newInfoBuilder.setUpdateDt(System.currentTimeMillis()); mIMap.put(channelId, newInfoBuilder.build()); + databaseStore.clearChannelListener(channelId); + hzInstance.getMultiMap(CHANNEL_LISTENERS).remove(channelId); callbackChannelInfoUpdateEvent(operator, channelId, Channel_Event_Destroy); return ErrorCode.ERROR_CODE_SUCCESS; diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 888d89af..73b6cb85 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -209,7 +209,7 @@ public interface IMessagesStore { void clearUserChannels(String userId); ErrorCode modifyChannelInfo(String operator, String channelId, int modifyType, String value); ErrorCode transferChannel(String operator, String channelId, String newOwner); - ErrorCode distoryChannel(String operator, String channelId, boolean isAdmin); + ErrorCode destroyChannel(String operator, String channelId, boolean isAdmin); List searchChannel(String keyword, boolean buzzy, int page); ErrorCode listenChannel(String operator, String channelId, boolean listen); WFCMessage.ChannelInfo getChannelInfo(String channelId);