import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
    switch (request.getCode()) {
    case RequestCode.HEART_BEAT:
        return this.heartBeat(ctx, request);
    case RequestCode.UNREGISTER_CLIENT:
        return this.unregisterClient(ctx, request);
    case RequestCode.GET_CONSUMER_LIST_BY_GROUP:
        return this.getConsumerListByGroup(ctx, request);
    case RequestCode.UPDATE_CONSUMER_OFFSET:
        return this.updateConsumerOffset(ctx, request);
    case RequestCode.QUERY_CONSUMER_OFFSET:
        return this.queryConsumerOffset(ctx, request);
    default:
        break;  
    }
    return null;
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
private RemotingCommand consumeMessageDirectly(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final ConsumeMessageDirectlyResultRequestHeader requestHeader =
            (ConsumeMessageDirectlyResultRequestHeader) request
                .decodeCommandCustomHeader(ConsumeMessageDirectlyResultRequestHeader.class);

    final MessageExt msg = MessageDecoder.decode(ByteBuffer.wrap(request.getBody()));

    ConsumeMessageDirectlyResult result =
            this.mqClientFactory.consumeMessageDirectly(msg, requestHeader.getConsumerGroup(), requestHeader.getBrokerName());

    if (null != result) {
        response.setCode(ResponseCode.SUCCESS);
        response.setBody(result.encode());
    }
    else {
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark(String.format("The Consumer Group <%s> not exist in this consumer", requestHeader.getConsumerGroup()));
    }

    return response;
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
private RemotingCommand getConsumerRunningInfo(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetConsumerRunningInfoRequestHeader requestHeader =
            (GetConsumerRunningInfoRequestHeader) request.decodeCommandCustomHeader(GetConsumerRunningInfoRequestHeader.class);

    ConsumerRunningInfo consumerRunningInfo = this.mqClientFactory.consumerRunningInfo(requestHeader.getConsumerGroup());
    if (null != consumerRunningInfo) {
        if (requestHeader.isJstackEnable()) {
            Map<Thread, StackTraceElement[]> map = Thread.getAllStackTraces();
            consumerRunningInfo.setStackTraceElementMap(map);
            String jstack = UtilAll.jstack(map);
            consumerRunningInfo.setJstack(jstack);
        }

        response.setCode(ResponseCode.SUCCESS);
        response.setBody(consumerRunningInfo.encode());
    }
    else {
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark(String.format("The Consumer Group <%s> not exist in this consumer", requestHeader.getConsumerGroup()));
    }

    return response;
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
private RemotingCommand searchOffsetByTimestamp(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(SearchOffsetResponseHeader.class);
    final SearchOffsetResponseHeader responseHeader = (SearchOffsetResponseHeader) response.readCustomHeader();
    final SearchOffsetRequestHeader requestHeader =
            (SearchOffsetRequestHeader) request.decodeCommandCustomHeader(SearchOffsetRequestHeader.class);

    long offset =
            this.brokerController.getMessageStore().getOffsetInQueueByTime(requestHeader.getTopic(), requestHeader.getQueueId(),
                requestHeader.getTimestamp());

    responseHeader.setOffset(offset);

    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
private PullResult processPullResponse(final RemotingCommand response) throws MQBrokerException, RemotingCommandException {
    PullStatus pullStatus = PullStatus.NO_NEW_MSG;
    switch (response.getCode()) {
    case ResponseCode.SUCCESS:
        pullStatus = PullStatus.FOUND;
        break;
    case ResponseCode.PULL_NOT_FOUND:
        pullStatus = PullStatus.NO_NEW_MSG;
        break;
    case ResponseCode.PULL_RETRY_IMMEDIATELY:
        pullStatus = PullStatus.NO_MATCHED_MSG;
        break;
    case ResponseCode.PULL_OFFSET_MOVED:
        pullStatus = PullStatus.OFFSET_ILLEGAL;
        break;

    default:
        throw new MQBrokerException(response.getCode(), response.getRemark());
    }

    PullMessageResponseHeader responseHeader =
            (PullMessageResponseHeader) response.decodeCommandCustomHeader(PullMessageResponseHeader.class);

    return new PullResultExt(pullStatus, responseHeader.getNextBeginOffset(), responseHeader.getMinOffset(),
        responseHeader.getMaxOffset(), null, responseHeader.getSuggestWhichBrokerId(), response.getBody());
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public void executeSendMessageHookAfter(final RemotingCommand response, final SendMessageContext context) {
    if (hasSendMessageHook()) {
        for (SendMessageHook hook : this.sendMessageHookList) {
            try {
                if (response != null) {
                    final SendMessageResponseHeader responseHeader =
                            (SendMessageResponseHeader) response.readCustomHeader();
                    context.setMsgId(responseHeader.getMsgId());
                    context.setQueueId(responseHeader.getQueueId());
                    context.setQueueOffset(responseHeader.getQueueOffset());
                    context.setCode(response.getCode());
                    context.setErrorMsg(response.getRemark());
                }
                hook.sendMessageAfter(context);
            }
            catch (Throwable e) {

            }
        }
    }
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public void unlockBatchMQ(//
        final String addr,//
        final UnlockBatchRequestBody requestBody,//
        final long timeoutMillis,//
        final boolean oneway//
) throws RemotingException, MQBrokerException, InterruptedException {
    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNLOCK_BATCH_MQ, null);

    request.setBody(requestBody.encode());

    if (oneway) {
        this.remotingClient.invokeOneway(addr, request, timeoutMillis);
    }
    else {
        RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
        switch (response.getCode()) {
        case ResponseCode.SUCCESS: {
            return;
        }
        default:
            break;
        }

        throw new MQBrokerException(response.getCode(), response.getRemark());
    }
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public TopicStatsTable getTopicStatsInfo(final String addr, final String topic, final long timeoutMillis) throws InterruptedException,
        RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException {
    GetTopicStatsInfoRequestHeader requestHeader = new GetTopicStatsInfoRequestHeader();
    requestHeader.setTopic(topic);

    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_TOPIC_STATS_INFO, requestHeader);

    RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        TopicStatsTable topicStatsTable = TopicStatsTable.decode(response.getBody(), TopicStatsTable.class);
        return topicStatsTable;
    }
    default:
        break;
    }

    throw new MQBrokerException(response.getCode(), response.getRemark());
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public ProducerConnection getProducerConnectionList(final String addr, final String producerGroup, final long timeoutMillis)
        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException,
        MQBrokerException {
    GetProducerConnectionListRequestHeader requestHeader = new GetProducerConnectionListRequestHeader();
    requestHeader.setProducerGroup(producerGroup);

    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_PRODUCER_CONNECTION_LIST, requestHeader);

    RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        return ProducerConnection.decode(response.getBody(), ProducerConnection.class);
    }
    default:
        break;
    }

    throw new MQBrokerException(response.getCode(), response.getRemark());
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public ConsumerConnection getConsumerConnectionList(final String addr, final String consumerGroup, final long timeoutMillis)
        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException,
        MQBrokerException {
    GetConsumerConnectionListRequestHeader requestHeader = new GetConsumerConnectionListRequestHeader();
    requestHeader.setConsumerGroup(consumerGroup);

    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_CONNECTION_LIST, requestHeader);

    RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        ConsumerConnection consumerConnection = ConsumerConnection.decode(response.getBody(), ConsumerConnection.class);
        return consumerConnection;
    }
    default:
        break;
    }

    throw new MQBrokerException(response.getCode(), response.getRemark());
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public KVTable getBrokerRuntimeInfo(final String addr, final long timeoutMillis) throws RemotingConnectException,
        RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException {

    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_RUNTIME_INFO, null);

    RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        return KVTable.decode(response.getBody(), KVTable.class);
    }
    default:
        break;
    }

    throw new MQBrokerException(response.getCode(), response.getRemark());
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public GroupList queryTopicConsumeByWho(final String addr, final String topic, final long timeoutMillis)
        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException,
        MQBrokerException {
    QueryTopicConsumeByWhoRequestHeader requestHeader = new QueryTopicConsumeByWhoRequestHeader();
    requestHeader.setTopic(topic);

    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_TOPIC_CONSUME_BY_WHO, requestHeader);

    RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        GroupList groupList = GroupList.decode(response.getBody(), GroupList.class);
        return groupList;
    }
    default:
        break;
    }

    throw new MQBrokerException(response.getCode(), response.getRemark());
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public List<QueueTimeSpan> queryConsumeTimeSpan(final String addr, final String topic, final String group, final long timeoutMillis)
        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException,
        MQBrokerException {
    QueryConsumeTimeSpanRequestHeader requestHeader = new QueryConsumeTimeSpanRequestHeader();
    requestHeader.setTopic(topic);
    requestHeader.setGroup(group);

    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_CONSUME_TIME_SPAN, requestHeader);

    RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        QueryConsumeTimeSpanBody consumeTimeSpanBody = GroupList.decode(response.getBody(), QueryConsumeTimeSpanBody.class);
        return consumeTimeSpanBody.getConsumeTimeSpanSet();
    }
    default:
        break;
    }

    throw new MQBrokerException(response.getCode(), response.getRemark());
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
private RemotingCommand getKVListByNamespace(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetKVListByNamespaceRequestHeader requestHeader =
            (GetKVListByNamespaceRequestHeader) request.decodeCommandCustomHeader(GetKVListByNamespaceRequestHeader.class);

    byte[] jsonValue = this.namesrvController.getKvConfigManager().getKVListByNamespace(//
        requestHeader.getNamespace());
    if (null != jsonValue) {
        response.setBody(jsonValue);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

    response.setCode(ResponseCode.QUERY_NOT_FOUND);
    response.setRemark("No config item, Namespace: " + requestHeader.getNamespace());
    return response;
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
private RemotingCommand wipeWritePermOfBroker(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(WipeWritePermOfBrokerResponseHeader.class);
    final WipeWritePermOfBrokerResponseHeader responseHeader = (WipeWritePermOfBrokerResponseHeader) response.readCustomHeader();
    final WipeWritePermOfBrokerRequestHeader requestHeader =
            (WipeWritePermOfBrokerRequestHeader) request.decodeCommandCustomHeader(WipeWritePermOfBrokerRequestHeader.class);

    int wipeTopicCnt = this.namesrvController.getRouteInfoManager().wipeWritePermOfBrokerByLock(requestHeader.getBrokerName());

    log.info("wipe write perm of broker[{}], client: {}, {}", //
        requestHeader.getBrokerName(), //
        RemotingHelper.parseChannelRemoteAddr(ctx.channel()), //
        wipeTopicCnt);

    responseHeader.setWipeTopicCount(wipeTopicCnt);
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetRouteInfoRequestHeader requestHeader =
            (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);

    TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());

    if (topicRouteData != null) {
        String orderTopicConf =
                this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,
                    requestHeader.getTopic());
        topicRouteData.setOrderTopicConf(orderTopicConf);

        byte[] content = topicRouteData.encode();
        response.setBody(content);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

    response.setCode(ResponseCode.TOPIC_NOT_EXIST);
    response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic()
            + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
    return response;
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public RemotingCommand getKVConfig(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(GetKVConfigResponseHeader.class);
    final GetKVConfigResponseHeader responseHeader = (GetKVConfigResponseHeader) response.readCustomHeader();
    final GetKVConfigRequestHeader requestHeader =
            (GetKVConfigRequestHeader) request.decodeCommandCustomHeader(GetKVConfigRequestHeader.class);

    String value = this.namesrvController.getKvConfigManager().getKVConfig(//
        requestHeader.getNamespace(),//
        requestHeader.getKey()//
        );

    if (value != null) {
        responseHeader.setValue(value);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

    response.setCode(ResponseCode.QUERY_NOT_FOUND);
    response.setRemark("No config item, Namespace: " + requestHeader.getNamespace() + " Key: " + requestHeader.getKey());
    return response;
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
private RemotingCommand lockBatchMQ(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    LockBatchRequestBody requestBody = LockBatchRequestBody.decode(request.getBody(), LockBatchRequestBody.class);

    Set<MessageQueue> lockOKMQSet = this.brokerController.getRebalanceLockManager().tryLockBatch(//
        requestBody.getConsumerGroup(),//
        requestBody.getMqSet(),//
        requestBody.getClientId());

    LockBatchResponseBody responseBody = new LockBatchResponseBody();
    responseBody.setLockOKMQSet(lockOKMQSet);

    response.setBody(responseBody.encode());
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
public void notifyConsumerIdsChanged(//
        final Channel channel,//
        final String consumerGroup//
) {
    if (null == consumerGroup) {
        log.error("notifyConsumerIdsChanged consumerGroup is null");
        return;
    }

    NotifyConsumerIdsChangedRequestHeader requestHeader = new NotifyConsumerIdsChangedRequestHeader();
    requestHeader.setConsumerGroup(consumerGroup);
    RemotingCommand request =
            RemotingCommand.createRequestCommand(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, requestHeader);

    try {
        this.brokerController.getRemotingServer().invokeOneway(channel, request, 10);
    }
    catch (Exception e) {
        log.error("notifyConsumerIdsChanged exception, " + consumerGroup, e);
    }
}
 

import com.alibaba.rocketmq.remoting.protocol.RemotingCommand; //导入依赖的package包/类
private RemotingCommand getBrokerConfig(ChannelHandlerContext ctx, RemotingCommand request) {

        final RemotingCommand response = RemotingCommand.createResponseCommand(GetBrokerConfigResponseHeader.class);
        final GetBrokerConfigResponseHeader responseHeader = (GetBrokerConfigResponseHeader) response.readCustomHeader();

        String content = this.brokerController.encodeAllConfig();
        if (content != null && content.length() > 0) {
            try {
                response.setBody(content.getBytes(MixAll.DEFAULT_CHARSET));
            }
            catch (UnsupportedEncodingException e) {
                log.error("", e);

                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("UnsupportedEncodingException " + e);
                return response;
            }
        }

        responseHeader.setVersion(this.brokerController.getConfigDataVersion());

        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }