rocketmq页面工具

/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You 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 org.apache.rocketmq.dashboard.service.client; import com.google.common.base.Throwables; import java.io.UnsupportedEncodingException; import java.util.List; import java.util.Map; import java.util.Properties; import java.util.Set; import org.apache.rocketmq.client.QueryResult; import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.impl.MQAdminImpl; import org.apache.rocketmq.common.AclConfig; import org.apache.rocketmq.common.PlainAccessConfig; import org.apache.rocketmq.common.TopicConfig; import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats; import org.apache.rocketmq.remoting.protocol.admin.RollbackStats; import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable; import org.apache.rocketmq.common.message.MessageClientIDSetter; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.message.MessageQueue; import org.apache.rocketmq.common.message.MessageRequestMode; import org.apache.rocketmq.remoting.protocol.RequestCode; import org.apache.rocketmq.remoting.protocol.ResponseCode; import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo; import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData; import org.apache.rocketmq.remoting.protocol.body.ClusterAclVersionInfo; import org.apache.rocketmq.remoting.protocol.body.ClusterInfo; import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult; import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList; import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection; import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo; import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache; import org.apache.rocketmq.remoting.protocol.body.GroupList; import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo; import org.apache.rocketmq.remoting.protocol.body.KVTable; import org.apache.rocketmq.remoting.protocol.body.ProducerConnection; import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo; import org.apache.rocketmq.remoting.protocol.body.QueryConsumeQueueResponseBody; import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan; import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper; import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper; import org.apache.rocketmq.remoting.protocol.body.TopicList; import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader; import org.apache.rocketmq.remoting.protocol.header.controller.GetMetaDataResponseHeader; import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData; import org.apache.rocketmq.remoting.protocol.route.TopicRouteData; import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail; import org.apache.rocketmq.remoting.protocol.subscription.GroupForbidden; import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig; import org.apache.rocketmq.dashboard.util.JsonUtil; import org.apache.rocketmq.remoting.RemotingClient; import org.apache.rocketmq.remoting.exception.RemotingCommandException; import org.apache.rocketmq.remoting.exception.RemotingConnectException; import org.apache.rocketmq.remoting.exception.RemotingException; import org.apache.rocketmq.remoting.exception.RemotingSendRequestException; import org.apache.rocketmq.remoting.exception.RemotingTimeoutException; import org.apache.rocketmq.remoting.protocol.RemotingCommand; import org.apache.rocketmq.tools.admin.MQAdminExt; import org.apache.rocketmq.tools.admin.api.BrokerOperatorResult; import org.apache.rocketmq.tools.admin.api.MessageTrack; import org.apache.rocketmq.tools.admin.common.AdminToolResult; import org.joor.Reflect; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Service; import static org.apache.rocketmq.remoting.protocol.RemotingSerializable.decode; @Service public class MQAdminExtImpl implements MQAdminExt { private Logger logger = LoggerFactory.getLogger(MQAdminExtImpl.class); public MQAdminExtImpl() { } @Override public void updateBrokerConfig(String brokerAddr, Properties properties) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, UnsupportedEncodingException, InterruptedException, MQBrokerException, MQClientException { MQAdminInstance.threadLocalMQAdminExt().updateBrokerConfig(brokerAddr, properties); } @Override public void createAndUpdateTopicConfig(String addr, TopicConfig config) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { MQAdminInstance.threadLocalMQAdminExt().createAndUpdateTopicConfig(addr, config); } @Override public void createAndUpdatePlainAccessConfig(String addr, PlainAccessConfig plainAccessConfig) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { MQAdminInstance.threadLocalMQAdminExt().createAndUpdatePlainAccessConfig(addr, plainAccessConfig); } @Override public void deletePlainAccessConfig(String addr, String accessKey) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { MQAdminInstance.threadLocalMQAdminExt().deletePlainAccessConfig(addr, accessKey); } @Override public void updateGlobalWhiteAddrConfig(String addr, String globalWhiteAddrs) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { MQAdminInstance.threadLocalMQAdminExt().updateGlobalWhiteAddrConfig(addr, globalWhiteAddrs); } @Override public ClusterAclVersionInfo examineBrokerClusterAclVersionInfo( String addr) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { return null; } @Override public AclConfig examineBrokerClusterAclConfig( String addr) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { return MQAdminInstance.threadLocalMQAdminExt().examineBrokerClusterAclConfig(addr); } @Override public void createAndUpdateSubscriptionGroupConfig(String addr, SubscriptionGroupConfig config) throws RemotingException, MQBrokerException, InterruptedException, MQClientException { MQAdminInstance.threadLocalMQAdminExt().createAndUpdateSubscriptionGroupConfig(addr, config); } @Override public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) throws MQBrokerException { RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient(); RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, null); RemotingCommand response = null; try { response = remotingClient.invokeSync(addr, request, 3000); } catch (Exception err) { Throwables.throwIfUnchecked(err); throw new RuntimeException(err); } assert response != null