import com.alibaba.rocketmq.remoting.exception.RemotingCommandException; //导入依赖的package包/类
@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    switch (request.getCode()) {
    case RequestCode.CHECK_TRANSACTION_STATE:
        return this.checkTransactionState(ctx, request);
    case RequestCode.NOTIFY_CONSUMER_IDS_CHANGED:
        return this.notifyConsumerIdsChanged(ctx, request);
    case RequestCode.RESET_CONSUMER_CLIENT_OFFSET:
        return this.resetOffset(ctx, request);
    case RequestCode.GET_CONSUMER_STATUS_FROM_CLIENT:
        return this.getConsumeStatus(ctx, request);

    case RequestCode.GET_CONSUMER_RUNNING_INFO:
        return this.getConsumerRunningInfo(ctx, request);

    case RequestCode.CONSUME_MESSAGE_DIRECTLY:
        return this.consumeMessageDirectly(ctx, request);
    default:
        break;
    }
    return null;
}
 

import com.alibaba.rocketmq.remoting.exception.RemotingCommandException; //导入依赖的package包/类
@Override
public void checkFields() throws RemotingCommandException {
    if (MessageSysFlag.TransactionNotType == this.commitOrRollback) {
        return;
    }

    if (MessageSysFlag.TransactionCommitType == this.commitOrRollback) {
        return;
    }

    if (MessageSysFlag.TransactionRollbackType == this.commitOrRollback) {
        return;
    }

    throw new RemotingCommandException("commitOrRollback field wrong");
}
 

import com.alibaba.rocketmq.remoting.exception.RemotingCommandException; //导入依赖的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.exception.RemotingCommandException; //导入依赖的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.exception.RemotingCommandException; //导入依赖的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.exception.RemotingCommandException; //导入依赖的package包/类
public RegisterFilterServerResponseHeader registerFilterServerToBroker(//
        final String brokerAddr,// 1
        final String filterServerAddr// 2
) throws RemotingCommandException, RemotingConnectException, RemotingSendRequestException,
        RemotingTimeoutException, InterruptedException, MQBrokerException {
    RegisterFilterServerRequestHeader requestHeader = new RegisterFilterServerRequestHeader();
    requestHeader.setFilterServerAddr(filterServerAddr);
    RemotingCommand request =
            RemotingCommand.createRequestCommand(RequestCode.REGISTER_FILTER_SERVER, requestHeader);

    RemotingCommand response = this.remotingClient.invokeSync(brokerAddr, request, 3000);
    assert response != null;
    switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        RegisterFilterServerResponseHeader responseHeader =
                (RegisterFilterServerResponseHeader) response
                    .decodeCommandCustomHeader(RegisterFilterServerResponseHeader.class);

        return responseHeader;
    }
    default:
        break;
    }

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

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

    try {
        boolean ok = this.filtersrvController.getFilterClassManager().registerFilterClass(requestHeader.getConsumerGroup(),//
            requestHeader.getTopic(),//
            requestHeader.getClassName(),//
            requestHeader.getClassCRC(), //
            request.getBody());//
        if (!ok) {
            throw new Exception("registerFilterClass error");
        }
    }
    catch (Exception e) {
        response.setCode(ResponseCode.SYSTEM_ERROR);
        response.setRemark(RemotingHelper.exceptionSimpleDesc(e));
        return response;
    }

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

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

    request.getExtFields().put("brokerName", this.brokerController.getBrokerConfig().getBrokerName());
    SelectMapedBufferResult selectMapedBufferResult = null;
    try {
        MessageId messageId = MessageDecoder.decodeMessageId(requestHeader.getMsgId());
        selectMapedBufferResult = this.brokerController.getMessageStore().selectOneMessageByOffset(messageId.getOffset());

        byte[] body = new byte[selectMapedBufferResult.getSize()];
        selectMapedBufferResult.getByteBuffer().get(body);
        request.setBody(body);
    }
    catch (UnknownHostException e) {
    }
    finally {
        if (selectMapedBufferResult != null) {
            selectMapedBufferResult.release();
        }
    }

    return this.callConsumer(RequestCode.CONSUME_MESSAGE_DIRECTLY, request, requestHeader.getConsumerGroup(),
        requestHeader.getClientId());
}
 

import com.alibaba.rocketmq.remoting.exception.RemotingCommandException; //导入依赖的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.exception.RemotingCommandException; //导入依赖的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.exception.RemotingCommandException; //导入依赖的package包/类
public RemotingCommand resetOffset(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final ResetOffsetRequestHeader requestHeader =
            (ResetOffsetRequestHeader) request.decodeCommandCustomHeader(ResetOffsetRequestHeader.class);
    log.info("[reset-offset] reset offset started by {}. topic={}, group={}, timestamp={}, isForce={}",
        new Object[] { RemotingHelper.parseChannelRemoteAddr(ctx.channel()), requestHeader.getTopic(), requestHeader.getGroup(),
                      requestHeader.getTimestamp(), requestHeader.isForce() });
    boolean isC = false;
    LanguageCode language = request.getLanguage();
    switch (language) {
    case CPP:
        isC = true;
        break;
    }
    return this.brokerController.getBroker2Client().resetOffset(requestHeader.getTopic(), requestHeader.getGroup(),
        requestHeader.getTimestamp(), requestHeader.isForce(), isC);
}
 

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

    Map<Integer, Long> correctionOffset =
            this.brokerController.getConsumerOffsetManager().queryMinOffsetInAllGroup(requestHeader.getTopic(),
                requestHeader.getFilterGroups());

    Map<Integer, Long> compareOffset =
            this.brokerController.getConsumerOffsetManager().queryOffset(requestHeader.getTopic(), requestHeader.getCompareGroup());

    if (compareOffset != null && !compareOffset.isEmpty()) {
        for (Integer queueId : compareOffset.keySet()) {
            correctionOffset.put(queueId, correctionOffset.get(queueId) > compareOffset.get(queueId) ? Long.MAX_VALUE
                    : correctionOffset.get(queueId));
        }
    }

    QueryCorrectionOffsetBody body = new QueryCorrectionOffsetBody();
    body.setCorrectionOffsets(correctionOffset);
    response.setBody(body.encode());
    response.setCode(ResponseCode.SUCCESS);
    response.setRemark(null);
    return response;
}
 

import com.alibaba.rocketmq.remoting.exception.RemotingCommandException; //导入依赖的package包/类
@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    SendMessageContext mqtraceContext = null;
    switch (request.getCode()) {
    case RequestCode.CONSUMER_SEND_MSG_BACK: //客户端消费失败，重新打回的消息
        return this.consumerSendMsgBack(ctx, request);
    default: //其他消息
        SendMessageRequestHeader requestHeader = parseRequestHeader(request); //客户端producer发送的SEND_MESSAGE 等消息都在这里面解析
        if (requestHeader == null) {
            return null;
        }
        mqtraceContext = buildMsgContext(ctx, requestHeader);
        this.executeSendMessageHookBefore(ctx, request, mqtraceContext);
        //消息处理  把接收的晓得写入commitlog等也是在这里面
        final RemotingCommand response = this.sendMessage(ctx, request, mqtraceContext, requestHeader);
        this.executeSendMessageHookAfter(response, mqtraceContext);
        return response;
    }
}
 

import com.alibaba.rocketmq.remoting.exception.RemotingCommandException; //导入依赖的package包/类
public RemotingCommand checkTransactionState(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final CheckTransactionStateRequestHeader requestHeader =
            (CheckTransactionStateRequestHeader) request.decodeCommandCustomHeader(CheckTransactionStateRequestHeader.class);
    final ByteBuffer byteBuffer = ByteBuffer.wrap(request.getBody());
    final MessageExt messageExt = MessageDecoder.decode(byteBuffer);
    if (messageExt != null) {
        final String group = messageExt.getProperty(MessageConst.PROPERTY_PRODUCER_GROUP);
        if (group != null) {
            MQProducerInner producer = this.mqClientFactory.selectProducer(group);
            if (producer != null) {
                final String addr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                producer.checkTransactionState(addr, messageExt, requestHeader);
            }
            else {
                log.debug("checkTransactionState, pick producer by group[{}] failed", group);
            }
        }
        else {
            log.warn("checkTransactionState, pick producer group failed");
        }
    }
    else {
        log.warn("checkTransactionState, decode message failed");
    }

    return null;
}
 

import com.alibaba.rocketmq.remoting.exception.RemotingCommandException; //导入依赖的package包/类
public RemotingCommand notifyConsumerIdsChanged(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    try {
        final NotifyConsumerIdsChangedRequestHeader requestHeader =
                (NotifyConsumerIdsChangedRequestHeader) request.decodeCommandCustomHeader(NotifyConsumerIdsChangedRequestHeader.class);
        log.info("receive broker's notification[{}], the consumer group: {} changed, rebalance immediately",//
            RemotingHelper.parseChannelRemoteAddr(ctx.channel()),//
            requestHeader.getConsumerGroup());
        this.mqClientFactory.rebalanceImmediately();
    }
    catch (Exception e) {
        log.error("notifyConsumerIdsChanged exception", RemotingHelper.exceptionSimpleDesc(e));
    }
    return null;
}
 

import com.alibaba.rocketmq.remoting.exception.RemotingCommandException; //导入依赖的package包/类
public RemotingCommand resetOffset(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final ResetOffsetRequestHeader requestHeader =
            (ResetOffsetRequestHeader) request.decodeCommandCustomHeader(ResetOffsetRequestHeader.class);
    log.info("invoke reset offset operation from broker. brokerAddr={}, topic={}, group={}, timestamp={}",
        new Object[] { RemotingHelper.parseChannelRemoteAddr(ctx.channel()), requestHeader.getTopic(), requestHeader.getGroup(),
                      requestHeader.getTimestamp() });
    Map<MessageQueue, Long> offsetTable = new HashMap<MessageQueue, Long>();
    if (request.getBody() != null) {
        ResetOffsetBody body = ResetOffsetBody.decode(request.getBody(), ResetOffsetBody.class);
        offsetTable = body.getOffsetTable();
    }
    this.mqClientFactory.resetOffset(requestHeader.getTopic(), requestHeader.getGroup(), offsetTable);
    return null;
}
 

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

    Map<MessageQueue, Long> offsetTable = this.mqClientFactory.getConsumerStatus(requestHeader.getTopic(), requestHeader.getGroup());
    GetConsumerStatusBody body = new GetConsumerStatusBody();
    body.setMessageQueueTable(offsetTable);
    response.setBody(body.encode());
    response.setCode(ResponseCode.SUCCESS);
    return response;
}
 

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

    this.namesrvController.getKvConfigManager().putKVConfig(//
        requestHeader.getNamespace(),//
        requestHeader.getKey(),//
        requestHeader.getValue()//
        );

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

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

    this.namesrvController.getKvConfigManager().deleteKVConfig(//
        requestHeader.getNamespace(),//
        requestHeader.getKey()//
        );

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

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

    this.namesrvController.getRouteInfoManager().unregisterBroker(//
        requestHeader.getClusterName(), // 1
        requestHeader.getBrokerAddr(), // 2
        requestHeader.getBrokerName(), // 3
        requestHeader.getBrokerId());

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