<commit_before>/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.namesrv.routeinfo;

import com.alibaba.rocketmq.common.DataVersion;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.TopicConfig;
import com.alibaba.rocketmq.common.constant.LoggerName;
import com.alibaba.rocketmq.common.constant.PermName;
import com.alibaba.rocketmq.common.namesrv.RegisterBrokerResult;
import com.alibaba.rocketmq.common.protocol.body.ClusterInfo;
import com.alibaba.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
import com.alibaba.rocketmq.common.protocol.body.TopicList;
import com.alibaba.rocketmq.common.protocol.route.BrokerData;
import com.alibaba.rocketmq.common.protocol.route.QueueData;
import com.alibaba.rocketmq.common.protocol.route.TopicRouteData;
import com.alibaba.rocketmq.common.sysflag.TopicSysFlag;
import com.alibaba.rocketmq.remoting.common.RemotingUtil;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;


/**
 * 运行过程中的路由信息，数据只在内存，宕机后数据消失，但是Broker会定期推送最新数据
 * 
 * @author shijia.wxr<<EMAIL>>
 * @since 2013-7-2
 */
public class RouteInfoManager {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.NamesrvLoggerName);
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;


    public RouteInfoManager() {
        this.topicQueueTable = new HashMap<String, List<QueueData>>(1024);
        this.brokerAddrTable = new HashMap<String, BrokerData>(128);
        this.clusterAddrTable = new HashMap<String, Set<String>>(32);
        this.brokerLiveTable = new HashMap<String, BrokerLiveInfo>(256);
        this.filterServerTable = new HashMap<String, List<String>>(256);
    }


    public byte[] getAllClusterInfo() {
        ClusterInfo clusterInfoSerializeWrapper = new ClusterInfo();
        clusterInfoSerializeWrapper.setBrokerAddrTable(this.brokerAddrTable);
        clusterInfoSerializeWrapper.setClusterAddrTable(this.clusterAddrTable);
        return clusterInfoSerializeWrapper.encode();
    }

    private static boolean contains(String[] array, String element) {
        if (null == array|| array.length == 0) {
            return false;
        }

        for (String e : array) {

            if (null == e && null == element) {
                return true;
            }

            if (null != e && e.equals(element)) {
                return true;
            }
        }

        return false;
    }

    public void deleteTopic(final String topic, final String[] brokerAddresses) {
        if (null == brokerAddresses) {
            log.info("Enter #deleteTopic: topic: {}, brokerAddresses: {}", topic, null);
        } else {
            log.info("Enter #deleteTopic: topic: {}, brokerAddresses: {}", topic, Arrays.asList(brokerAddresses));
        }

        try {
            try {
                this.lock.writeLock().lockInterruptibly();
                if (null == brokerAddresses || brokerAddresses.length == 0) {
                    log.info("Requested to delete topic: {}", topic);
                    this.topicQueueTable.remove(topic);
                } else {
                    log.info("Requested to delete topic: {} from brokers: {}", topic, Arrays.asList(brokerAddresses));
                    List<QueueData> queueDataList = topicQueueTable.get(topic);
                    if (null != queueDataList && !queueDataList.isEmpty()) {
                        Iterator<QueueData> iterator = queueDataList.iterator();
                        while (iterator.hasNext()) {
                            QueueData queueData = iterator.next();
                            if (contains(brokerAddresses, queueData.getBrokerName())) {
                                iterator.remove();
                            }
                        }
                    } else {
                        topicQueueTable.remove(topic);
                    }
                }
                log.info("Topic {} is deleted", topic);
            }
            finally {
                this.lock.writeLock().unlock();
            }
        }
        catch (Exception e) {
            log.error("deleteTopic Exception", e);
        }
    }


    public byte[] getAllTopicList() {
        TopicList topicList = new TopicList();
        try {
            try {
                this.lock.readLock().lockInterruptibly();
                topicList.getTopicList().addAll(this.topicQueueTable.keySet());<commit_msg>Add more log while removing topic.<commit_after>/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.namesrv.routeinfo;

import com.alibaba.rocketmq.common.DataVersion;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.TopicConfig;
import com.alibaba.rocketmq.common.constant.LoggerName;
import com.alibaba.rocketmq.common.constant.PermName;
import com.alibaba.rocketmq.common.namesrv.RegisterBrokerResult;
import com.alibaba.rocketmq.common.protocol.body.ClusterInfo;
import com.alibaba.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
import com.alibaba.rocketmq.common.protocol.body.TopicList;
import com.alibaba.rocketmq.common.protocol.route.BrokerData;
import com.alibaba.rocketmq.common.protocol.route.QueueData;
import com.alibaba.rocketmq.common.protocol.route.TopicRouteData;
import com.alibaba.rocketmq.common.sysflag.TopicSysFlag;
import com.alibaba.rocketmq.remoting.common.RemotingUtil;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;


/**
 * 运行过程中的路由信息，数据只在内存，宕机后数据消失，但是Broker会定期推送最新数据
 * 
 * @author shijia.wxr<<EMAIL>>
 * @since 2013-7-2
 */
public class RouteInfoManager {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.NamesrvLoggerName);
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;


    public RouteInfoManager() {
        this.topicQueueTable = new HashMap<String, List<QueueData>>(1024);
        this.brokerAddrTable = new HashMap<String, BrokerData>(128);
        this.clusterAddrTable = new HashMap<String, Set<String>>(32);
        this.brokerLiveTable = new HashMap<String, BrokerLiveInfo>(256);
        this.filterServerTable = new HashMap<String, List<String>>(256);
    }


    public byte[] getAllClusterInfo() {
        ClusterInfo clusterInfoSerializeWrapper = new ClusterInfo();
        clusterInfoSerializeWrapper.setBrokerAddrTable(this.brokerAddrTable);
        clusterInfoSerializeWrapper.setClusterAddrTable(this.clusterAddrTable);
        return clusterInfoSerializeWrapper.encode();
    }

    private static boolean contains(String[] array, String element) {
        if (null == array|| array.length == 0) {
            return false;
        }

        for (String e : array) {

            if (null == e && null == element) {
                return true;
            }

            if (null != e && e.equals(element)) {
                return true;
            }
        }

        return false;
    }

    public void deleteTopic(final String topic, final String[] brokerAddresses) {
        if (null == brokerAddresses) {
            log.info("Enter #deleteTopic: topic: {}, brokerAddresses: {}", topic, null);
        } else {
            log.info("Enter #deleteTopic: topic: {}, brokerAddresses: {}", topic, Arrays.asList(brokerAddresses));
        }

        try {
            try {
                this.lock.writeLock().lockInterruptibly();
                if (null == brokerAddresses || brokerAddresses.length == 0) {
                    log.info("Requested to delete topic: {}", topic);
                    this.topicQueueTable.remove(topic);
                    log.info("Topic {} is deleted completely.", topic);
                } else {
                    log.info("Requested to delete topic: {} from brokers: {}", topic, Arrays.asList(brokerAddresses));
                    List<QueueData> queueDataList = topicQueueTable.get(topic);
                    if (null != queueDataList && !queueDataList.isEmpty()) {
                        int count = 0;
                        Iterator<QueueData> iterator = queueDataList.iterator();
                        while (iterator.hasNext()) {
                            QueueData queueData = iterator.next();
                            if (contains(brokerAddresses, queueData.getBrokerName())) {
                                iterator.remove();
                                count++;
                            }
                        }
                        if (count < queueDataList.size()) {
                            log.info("Topic {} has {} queues removed.", topic, count);
                        } else {
                            //As there is no queue left.
                            topicQueueTable.remove(topic);
                            log.info("Topic {} is deleted completely", topic);
                        }
                    } else {
                        topicQueueTable.remove(topic);
                        log.info("Topic {} is deleted completely", topic);
                    }
                }
            }
            finally {
                this.lock.writeLock().unlock();
            }
        }
        catch (Exception e) {
            log.error("deleteTopic Exception", e);
        }
    }


    public byte[] getAllTopicList() {
        TopicList topicList = new TopicList();
        try {
            try {
                this.lock.readLock().lockInterruptibly();
                topicList.getTopicList().addAll(this.topicQueueTable.keySet());
{'fraction_non_alphanumeric': 0.10605933974091099, 'fraction_numerical': 0.0062682824905975765, 'mean_word_length': 2.425708559977097, 'pattern_counts': {'":': 0, '<': 44, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '538712', 'n_tokens_mistral': 3299, 'n_tokens_neox': 3151, 'n_words': 720}