删除用户或删除频道时，删除对于的订阅关系

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<WFCMessage.IDBuf> {
+public class DestroyChannelHandler extends GroupHandler<WFCMessage.IDBuf> {
     @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);

broker/src/main/java/io/moquette/persistence/DatabaseStore.java

@@ -3332,6 +3332,57 @@ public class DatabaseStore {
         });
     }
 
+    List<String> getUserChannels(String userId) {
+        List<String> 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<String> getSensitiveWord() {
         Connection connection = null;
         PreparedStatement statement = null;

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<String, WFCMessage.ChannelInfo> 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;

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<WFCMessage.ChannelInfo> searchChannel(String keyword, boolean buzzy, int page);
     ErrorCode listenChannel(String operator, String channelId, boolean listen);
     WFCMessage.ChannelInfo getChannelInfo(String channelId);