You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2018/06/22 09:11:22 UTC
[rocketmq] 02/08: #325 all nameserer share topicroute Data
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
commit e8f1020e1c7956535e9a85a14082396e86191e5f
Author: 傅冲 <yu...@alibaba-inc.com>
AuthorDate: Fri Jun 1 12:39:21 2018 +0800
#325 all nameserer share topicroute Data
---
.../apache/rocketmq/broker/out/BrokerOuterAPI.java | 39 ++++++++++------------
1 file changed, 17 insertions(+), 22 deletions(-)
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
index 2825a34..c6421c7 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
@@ -125,14 +125,26 @@ public class BrokerOuterAPI {
final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
+
+ final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
+ requestHeader.setBrokerAddr(brokerAddr);
+ requestHeader.setBrokerId(brokerId);
+ requestHeader.setBrokerName(brokerName);
+ requestHeader.setClusterName(clusterName);
+ requestHeader.setHaServerAddr(haServerAddr);
+ requestHeader.setCompressed(compressed);
+
+ RegisterBrokerBody requestBody = new RegisterBrokerBody();
+ requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
+ requestBody.setFilterServerList(filterServerList);
+ final byte[] body = requestBody.encode(compressed);
final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run() {
try {
- RegisterBrokerResult result = registerBroker(namesrvAddr, clusterName, brokerAddr, brokerName, brokerId,
- haServerAddr, topicConfigWrapper, filterServerList, oneway, timeoutMills, compressed);
+ RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
if (result != null) {
registerBrokerResultList.add(result);
}
@@ -158,31 +170,14 @@ public class BrokerOuterAPI {
private RegisterBrokerResult registerBroker(
final String namesrvAddr,
- final String clusterName,
- final String brokerAddr,
- final String brokerName,
- final long brokerId,
- final String haServerAddr,
- final TopicConfigSerializeWrapper topicConfigWrapper,
- final List<String> filterServerList,
final boolean oneway,
final int timeoutMills,
- final boolean compressed
+ final RegisterBrokerRequestHeader requestHeader,
+ final byte[] body
) throws RemotingCommandException, MQBrokerException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
InterruptedException {
- RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
- requestHeader.setBrokerAddr(brokerAddr);
- requestHeader.setBrokerId(brokerId);
- requestHeader.setBrokerName(brokerName);
- requestHeader.setClusterName(clusterName);
- requestHeader.setHaServerAddr(haServerAddr);
- requestHeader.setCompressed(compressed);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);
-
- RegisterBrokerBody requestBody = new RegisterBrokerBody();
- requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
- requestBody.setFilterServerList(filterServerList);
- request.setBody(requestBody.encode(requestHeader.isCompressed()));
+ request.setBody(body);
if (oneway) {
try {