@Test public void testTopicRouteData2TopicPublishInfo() { TopicRouteData topicRouteData = new TopicRouteData(); topicRouteData.setFilterServerTable(new HashMap<String, List<String>>()); List<BrokerData> brokerDataList = new ArrayList<BrokerData>(); BrokerData brokerData = new BrokerData(); brokerData.setBrokerName("BrokerA"); brokerData.setCluster("DefaultCluster"); HashMap<Long, String> brokerAddrs = new HashMap<Long, String>(); brokerAddrs.put(0L, "127.0.0.1:10911"); brokerData.setBrokerAddrs(brokerAddrs); brokerDataList.add(brokerData); topicRouteData.setBrokerDatas(brokerDataList); List<QueueData> queueDataList = new ArrayList<QueueData>(); QueueData queueData = new QueueData(); queueData.setBrokerName("BrokerA"); queueData.setPerm(6); queueData.setReadQueueNums(3); queueData.setWriteQueueNums(4); queueData.setTopicSynFlag(0); queueDataList.add(queueData); topicRouteData.setQueueDatas(queueDataList); TopicPublishInfo topicPublishInfo = MQClientInstance.topicRouteData2TopicPublishInfo(topic, topicRouteData); assertThat(topicPublishInfo.isHaveTopicRouterInfo()).isFalse(); assertThat(topicPublishInfo.getMessageQueueList().size()).isEqualTo(4); }
public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) { TopicPublishInfo info = new TopicPublishInfo(); info.setTopicRouteData(route); if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) { String[] brokers = route.getOrderTopicConf().split(";"); for (String broker : brokers) { String[] item = broker.split(":"); int nums = Integer.parseInt(item[1]); for (int i = 0; i < nums; i++) { MessageQueue mq = new MessageQueue(topic, item[0], i); info.getMessageQueueList().add(mq); } } info.setOrderTopic(true); } else { List<QueueData> qds = route.getQueueDatas(); Collections.sort(qds); for (QueueData qd : qds) { if (PermName.isWriteable(qd.getPerm())) { BrokerData brokerData = null; for (BrokerData bd : route.getBrokerDatas()) { if (bd.getBrokerName().equals(qd.getBrokerName())) { brokerData = bd; break; } } if (null == brokerData) { continue; } if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) { continue; } for (int i = 0; i < qd.getWriteQueueNums(); i++) { MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i); info.getMessageQueueList().add(mq); } } } info.setOrderTopic(false); } return info; }
MQClientInstance { public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) { TopicPublishInfo info = new TopicPublishInfo(); info.setTopicRouteData(route); if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) { String[] brokers = route.getOrderTopicConf().split(";"); for (String broker : brokers) { String[] item = broker.split(":"); int nums = Integer.parseInt(item[1]); for (int i = 0; i < nums; i++) { MessageQueue mq = new MessageQueue(topic, item[0], i); info.getMessageQueueList().add(mq); } } info.setOrderTopic(true); } else { List<QueueData> qds = route.getQueueDatas(); Collections.sort(qds); for (QueueData qd : qds) { if (PermName.isWriteable(qd.getPerm())) { BrokerData brokerData = null; for (BrokerData bd : route.getBrokerDatas()) { if (bd.getBrokerName().equals(qd.getBrokerName())) { brokerData = bd; break; } } if (null == brokerData) { continue; } if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) { continue; } for (int i = 0; i < qd.getWriteQueueNums(); i++) { MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i); info.getMessageQueueList().add(mq); } } } info.setOrderTopic(false); } return info; } }
MQClientInstance { public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) { TopicPublishInfo info = new TopicPublishInfo(); info.setTopicRouteData(route); if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) { String[] brokers = route.getOrderTopicConf().split(";"); for (String broker : brokers) { String[] item = broker.split(":"); int nums = Integer.parseInt(item[1]); for (int i = 0; i < nums; i++) { MessageQueue mq = new MessageQueue(topic, item[0], i); info.getMessageQueueList().add(mq); } } info.setOrderTopic(true); } else { List<QueueData> qds = route.getQueueDatas(); Collections.sort(qds); for (QueueData qd : qds) { if (PermName.isWriteable(qd.getPerm())) { BrokerData brokerData = null; for (BrokerData bd : route.getBrokerDatas()) { if (bd.getBrokerName().equals(qd.getBrokerName())) { brokerData = bd; break; } } if (null == brokerData) { continue; } if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) { continue; } for (int i = 0; i < qd.getWriteQueueNums(); i++) { MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i); info.getMessageQueueList().add(mq); } } } info.setOrderTopic(false); } return info; } MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId);  MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook); }
MQClientInstance { public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) { TopicPublishInfo info = new TopicPublishInfo(); info.setTopicRouteData(route); if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) { String[] brokers = route.getOrderTopicConf().split(";"); for (String broker : brokers) { String[] item = broker.split(":"); int nums = Integer.parseInt(item[1]); for (int i = 0; i < nums; i++) { MessageQueue mq = new MessageQueue(topic, item[0], i); info.getMessageQueueList().add(mq); } } info.setOrderTopic(true); } else { List<QueueData> qds = route.getQueueDatas(); Collections.sort(qds); for (QueueData qd : qds) { if (PermName.isWriteable(qd.getPerm())) { BrokerData brokerData = null; for (BrokerData bd : route.getBrokerDatas()) { if (bd.getBrokerName().equals(qd.getBrokerName())) { brokerData = bd; break; } } if (null == brokerData) { continue; } if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) { continue; } for (int i = 0; i < qd.getWriteQueueNums(); i++) { MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i); info.getMessageQueueList().add(mq); } } } info.setOrderTopic(false); } return info; } MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId);  MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook); static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route); static Set<MessageQueue> topicRouteData2TopicSubscribeInfo(final String topic, final TopicRouteData route); void start(); String getClientId(); void updateTopicRouteInfoFromNameServer(); Map<MessageQueue, Long> parseOffsetTableFromBroker(Map<MessageQueue, Long> offsetTable, String namespace); void checkClientInBroker(); void sendHeartbeatToAllBrokerWithLock(); void adjustThreadPool(); boolean updateTopicRouteInfoFromNameServer(final String topic); boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
        DefaultMQProducer defaultMQProducer); void shutdown(); boolean registerConsumer(final String group, final MQConsumerInner consumer); void unregisterConsumer(final String group); boolean registerProducer(final String group, final DefaultMQProducerImpl producer); void unregisterProducer(final String group); boolean registerAdminExt(final String group, final MQAdminExtInner admin); void unregisterAdminExt(final String group); void rebalanceImmediately(); void doRebalance(); MQProducerInner selectProducer(final String group); MQConsumerInner selectConsumer(final String group); FindBrokerResult findBrokerAddressInAdmin(final String brokerName); String findBrokerAddressInPublish(final String brokerName); FindBrokerResult findBrokerAddressInSubscribe(
        final String brokerName,
        final long brokerId,
        final boolean onlyThisBroker
    ); int findBrokerVersion(String brokerName, String brokerAddr); List<String> findConsumerIdList(final String topic, final String group); String findBrokerAddrByTopic(final String topic); void resetOffset(String topic, String group, Map<MessageQueue, Long> offsetTable); Map<MessageQueue, Long> getConsumerStatus(String topic, String group); TopicRouteData getAnExistTopicRouteData(final String topic); MQClientAPIImpl getMQClientAPIImpl(); MQAdminImpl getMQAdminImpl(); long getBootTimestamp(); ScheduledExecutorService getScheduledExecutorService(); PullMessageService getPullMessageService(); DefaultMQProducer getDefaultMQProducer(); ConcurrentMap<String, TopicRouteData> getTopicRouteTable(); ConsumeMessageDirectlyResult consumeMessageDirectly(final MessageExt msg,
        final String consumerGroup,
        final String brokerName); ConsumerRunningInfo consumerRunningInfo(final String consumerGroup); ConsumerStatsManager getConsumerStatsManager(); NettyClientConfig getNettyClientConfig(); ClientConfig getClientConfig(); }
MQClientInstance { public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) { TopicPublishInfo info = new TopicPublishInfo(); info.setTopicRouteData(route); if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) { String[] brokers = route.getOrderTopicConf().split(";"); for (String broker : brokers) { String[] item = broker.split(":"); int nums = Integer.parseInt(item[1]); for (int i = 0; i < nums; i++) { MessageQueue mq = new MessageQueue(topic, item[0], i); info.getMessageQueueList().add(mq); } } info.setOrderTopic(true); } else { List<QueueData> qds = route.getQueueDatas(); Collections.sort(qds); for (QueueData qd : qds) { if (PermName.isWriteable(qd.getPerm())) { BrokerData brokerData = null; for (BrokerData bd : route.getBrokerDatas()) { if (bd.getBrokerName().equals(qd.getBrokerName())) { brokerData = bd; break; } } if (null == brokerData) { continue; } if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) { continue; } for (int i = 0; i < qd.getWriteQueueNums(); i++) { MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i); info.getMessageQueueList().add(mq); } } } info.setOrderTopic(false); } return info; } MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId);  MQClientInstance(ClientConfig clientConfig, int instanceIndex, String clientId, RPCHook rpcHook); static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route); static Set<MessageQueue> topicRouteData2TopicSubscribeInfo(final String topic, final TopicRouteData route); void start(); String getClientId(); void updateTopicRouteInfoFromNameServer(); Map<MessageQueue, Long> parseOffsetTableFromBroker(Map<MessageQueue, Long> offsetTable, String namespace); void checkClientInBroker(); void sendHeartbeatToAllBrokerWithLock(); void adjustThreadPool(); boolean updateTopicRouteInfoFromNameServer(final String topic); boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,
        DefaultMQProducer defaultMQProducer); void shutdown(); boolean registerConsumer(final String group, final MQConsumerInner consumer); void unregisterConsumer(final String group); boolean registerProducer(final String group, final DefaultMQProducerImpl producer); void unregisterProducer(final String group); boolean registerAdminExt(final String group, final MQAdminExtInner admin); void unregisterAdminExt(final String group); void rebalanceImmediately(); void doRebalance(); MQProducerInner selectProducer(final String group); MQConsumerInner selectConsumer(final String group); FindBrokerResult findBrokerAddressInAdmin(final String brokerName); String findBrokerAddressInPublish(final String brokerName); FindBrokerResult findBrokerAddressInSubscribe(
        final String brokerName,
        final long brokerId,
        final boolean onlyThisBroker
    ); int findBrokerVersion(String brokerName, String brokerAddr); List<String> findConsumerIdList(final String topic, final String group); String findBrokerAddrByTopic(final String topic); void resetOffset(String topic, String group, Map<MessageQueue, Long> offsetTable); Map<MessageQueue, Long> getConsumerStatus(String topic, String group); TopicRouteData getAnExistTopicRouteData(final String topic); MQClientAPIImpl getMQClientAPIImpl(); MQAdminImpl getMQAdminImpl(); long getBootTimestamp(); ScheduledExecutorService getScheduledExecutorService(); PullMessageService getPullMessageService(); DefaultMQProducer getDefaultMQProducer(); ConcurrentMap<String, TopicRouteData> getTopicRouteTable(); ConsumeMessageDirectlyResult consumeMessageDirectly(final MessageExt msg,
        final String consumerGroup,
        final String brokerName); ConsumerRunningInfo consumerRunningInfo(final String consumerGroup); ConsumerStatsManager getConsumerStatsManager(); NettyClientConfig getNettyClientConfig(); ClientConfig getClientConfig();  }