/**
* Copyright (C) 2013 kangliqiang ,kangliq@163.com
*
* Licensed 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.
*/
#include <assert.h>

#include "MQClientAPIImpl.h"
#include "MQClientException.h"
#include "SocketUtil.h"
#include "UtilAll.h"
#include "TcpRemotingClient.h"
#include "MQProtos.h"
#include "PullResultExt.h"
#include "ConsumerInvokeCallback.h"
#include "NamesrvUtil.h"
#include "VirtualEnvUtil.h"
#include "ClientRemotingProcessor.h"
#include "CommandCustomHeader.h"
#include "TopicList.h"
#include "ProducerInvokeCallback.h"
#include "MessageDecoder.h"
#include "MessageSysFlag.h"
#include "GetConsumerListByGroupResponseBody.h"


namespace rmq
{


MQClientAPIImpl::MQClientAPIImpl(ClientConfig& clientConfig,
								const RemoteClientConfig& remoteClientConfig,
                                ClientRemotingProcessor* pClientRemotingProcessor)
    : m_pClientRemotingProcessor(pClientRemotingProcessor)
{
    m_pRemotingClient = new TcpRemotingClient(remoteClientConfig);

    m_pRemotingClient->registerProcessor(CHECK_TRANSACTION_STATE_VALUE, m_pClientRemotingProcessor);
    m_pRemotingClient->registerProcessor(NOTIFY_CONSUMER_IDS_CHANGED_VALUE, m_pClientRemotingProcessor);
    m_pRemotingClient->registerProcessor(RESET_CONSUMER_CLIENT_OFFSET_VALUE, m_pClientRemotingProcessor);
    m_pRemotingClient->registerProcessor(GET_CONSUMER_STATUS_FROM_CLIENT_VALUE, m_pClientRemotingProcessor);
    m_pRemotingClient->registerProcessor(GET_CONSUMER_RUNNING_INFO_VALUE, m_pClientRemotingProcessor);
    m_pRemotingClient->registerProcessor(CONSUME_MESSAGE_DIRECTLY_VALUE, m_pClientRemotingProcessor);
}

MQClientAPIImpl::~MQClientAPIImpl()
{
}

std::string MQClientAPIImpl::getProjectGroupPrefix()
{
    return m_projectGroupPrefix;
}

std::vector<std::string> MQClientAPIImpl::getNameServerAddressList()
{
    return m_pRemotingClient->getNameServerAddressList();
}

TcpRemotingClient* MQClientAPIImpl::getRemotingClient()
{
    return m_pRemotingClient;
}

std::string MQClientAPIImpl::fetchNameServerAddr()
{
    try
    {
        std::string addrs = m_topAddressing.fetchNSAddr();
        if (!addrs.empty())
        {
            if (addrs != m_nameSrvAddr)
            {
            	RMQ_INFO("name server address changed, %s -> %s",
            		m_nameSrvAddr.c_str(), addrs.c_str());
                updateNameServerAddressList(addrs);
                m_nameSrvAddr = addrs;
                return m_nameSrvAddr;
            }
        }
    }
    catch (...)
    {
		RMQ_ERROR("fetchNameServerAddr Exception");
    }

    return m_nameSrvAddr;
}

void MQClientAPIImpl::updateNameServerAddressList(const std::string& addrs)
{
    m_nameSrvAddr = addrs;
    std::vector<std::string> av;
    UtilAll::Split(av, addrs, ";");
    if (av.size() > 0)
    {
    	m_pRemotingClient->updateNameServerAddressList(av);
    }
}

void MQClientAPIImpl::start()
{
    m_pRemotingClient->start();

    try
    {
        std::string localAddress = getLocalAddress();
        m_projectGroupPrefix = getProjectGroupByIp(localAddress, 3000);
    }
    catch (std::exception e)
    {
    }
}

void MQClientAPIImpl::shutdown()
{
    m_pRemotingClient->shutdown();
}

void MQClientAPIImpl::createSubscriptionGroup(const std::string& addr,
        SubscriptionGroupConfig config,
        int timeoutMillis)
{
    //TODO
}


void MQClientAPIImpl::createTopic(const std::string& addr,
                                  const std::string& defaultTopic,
                                  TopicConfig topicConfig,
                                  int timeoutMillis)
{
    std::string topicWithProjectGroup = topicConfig.getTopicName();
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        topicWithProjectGroup =
            VirtualEnvUtil::buildWithProjectGroup(topicConfig.getTopicName(), m_projectGroupPrefix);
    }

    CreateTopicRequestHeader* requestHeader = new CreateTopicRequestHeader();
    requestHeader->topic = (topicWithProjectGroup);
    requestHeader->defaultTopic = (defaultTopic);
    requestHeader->readQueueNums = (topicConfig.getReadQueueNums());
    requestHeader->writeQueueNums = (topicConfig.getWriteQueueNums());
    requestHeader->perm = (topicConfig.getPerm());
    requestHeader->topicFilterType = (topicConfig.getTopicFilterType());
    requestHeader->topicSysFlag = (topicConfig.getTopicSysFlag());
    requestHeader->order = (topicConfig.isOrder());

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(UPDATE_AND_CREATE_TOPIC_VALUE, requestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case SUCCESS_VALUE:
            {
                return;
            }
            default:
                break;
        }
        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    THROW_MQEXCEPTION(MQClientException, "createTopic failed", -1);
}

SendResult MQClientAPIImpl::sendMessage(const std::string& addr,
                                        const std::string& brokerName,
                                        Message& msg,
                                        SendMessageRequestHeader* pRequestHeader,
                                        int timeoutMillis,
                                        CommunicationMode communicationMode,
                                        SendCallback* pSendCallback)
{
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        msg.setTopic(VirtualEnvUtil::buildWithProjectGroup(msg.getTopic(), m_projectGroupPrefix));
        pRequestHeader->producerGroup = (VirtualEnvUtil::buildWithProjectGroup(pRequestHeader->producerGroup,
                                         m_projectGroupPrefix));
        pRequestHeader->topic = (VirtualEnvUtil::buildWithProjectGroup(pRequestHeader->topic,
                                 m_projectGroupPrefix));
    }

	bool sendSmartMsg = true;
	RemotingCommandPtr request = NULL;
    if (sendSmartMsg)
    {
        SendMessageRequestHeaderV2* pRequestHeaderV2 = SendMessageRequestHeaderV2::createSendMessageRequestHeaderV2(pRequestHeader);
        request = RemotingCommand::createRequestCommand(SEND_MESSAGE_V2_VALUE, pRequestHeaderV2);
        delete pRequestHeader;
    }
    else
    {
        request = RemotingCommand::createRequestCommand(SEND_MESSAGE_VALUE, pRequestHeader);
    }

    if (msg.getCompressBody() != NULL)
    {
    	request->setBody((char*)msg.getCompressBody(), msg.getCompressBodyLen(), false);
    }
    else
    {
    	request->setBody((char*)msg.getBody(), msg.getBodyLen(), false);
    }

    SendResult result;
    switch (communicationMode)
    {
        case ONEWAY:
            m_pRemotingClient->invokeOneway(addr, request, timeoutMillis);
            return result;
        case ASYNC:
            sendMessageAsync(addr, brokerName, msg, timeoutMillis, request, pSendCallback);
            return result;
        case SYNC:
        {
            SendResult* r = sendMessageSync(addr, brokerName, msg, timeoutMillis, request);
            if (r)
            {
                result = *r;
                delete r;
            }
            return result;
        }
        default:
            break;
    }
    return result;
}

PullResult* MQClientAPIImpl::pullMessage(const std::string& addr,
        PullMessageRequestHeader* pRequestHeader,
        int timeoutMillis,
        CommunicationMode communicationMode,
        PullCallback* pPullCallback)
{

    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        pRequestHeader->consumerGroup = (VirtualEnvUtil::buildWithProjectGroup(
                                             pRequestHeader->consumerGroup, m_projectGroupPrefix));
        pRequestHeader->topic = (VirtualEnvUtil::buildWithProjectGroup(pRequestHeader->topic,
                                 m_projectGroupPrefix));
    }

    RemotingCommandPtr request = RemotingCommand::createRequestCommand(PULL_MESSAGE_VALUE, pRequestHeader);

    PullResult* result = NULL;
    switch (communicationMode)
    {
        case ONEWAY:
            break;
        case ASYNC:
            pullMessageAsync(addr, request, timeoutMillis, pPullCallback);
            break;
        case SYNC:
            result =  pullMessageSync(addr, request, timeoutMillis);
            break;
        default:
            assert(false);
            break;
    }

    return result;
}

MessageExt* MQClientAPIImpl::viewMessage(const std::string& addr,  long long phyoffset,  int timeoutMillis)
{
    ViewMessageRequestHeader* requestHeader = new ViewMessageRequestHeader();
    requestHeader->offset = phyoffset;

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(VIEW_MESSAGE_BY_ID_VALUE, requestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case SUCCESS_VALUE:
            {
                if (response->getBody() != NULL)
                {
                	int len = 0;
                	MessageExt* messageExt = MessageDecoder::decode((char*)response->getBody(),
                		response->getBodyLen(), len);
                    if (!UtilAll::isBlank(m_projectGroupPrefix))
				    {
				        messageExt->setTopic(VirtualEnvUtil::clearProjectGroup(messageExt->getTopic(),
				        	m_projectGroupPrefix));
				    }
                    return messageExt;
                }
            }
            default:
                break;
        }
        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    THROW_MQEXCEPTION(MQClientException, "viewMessage failed", -1);
}

long long MQClientAPIImpl::searchOffset(const std::string& addr,
                                        const std::string& topic,
                                        int queueId,
                                        long long timestamp,
                                        int timeoutMillis)
{
    std::string topicWithProjectGroup = topic;
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        topicWithProjectGroup = VirtualEnvUtil::buildWithProjectGroup(topic, m_projectGroupPrefix);
    }

	SearchOffsetRequestHeader* pRequestHeader = new SearchOffsetRequestHeader();
    pRequestHeader->topic = topicWithProjectGroup;
    pRequestHeader->queueId = queueId;
    pRequestHeader->timestamp = timestamp;

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(SEARCH_OFFSET_BY_TIMESTAMP_VALUE, pRequestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
	    switch (response->getCode())
	    {
	        case SUCCESS_VALUE:
	        {
	            SearchOffsetResponseHeader* ret = (SearchOffsetResponseHeader*)response->getCommandCustomHeader();
	            return ret->offset;
	        }
	        default:
	            break;
	    }
	    //THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
	}

    //THROW_MQEXCEPTION(MQClientException, "searchOffset failed", -1);
    return -1;
}

long long MQClientAPIImpl::getMaxOffset(const std::string& addr,
                                        const std::string& topic,
                                        int queueId,
                                        int timeoutMillis)
{
    std::string topicWithProjectGroup = topic;
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        topicWithProjectGroup = VirtualEnvUtil::buildWithProjectGroup(topic, m_projectGroupPrefix);
    }

	GetMaxOffsetRequestHeader* pRequestHeader = new GetMaxOffsetRequestHeader();
    pRequestHeader->topic = topicWithProjectGroup;
    pRequestHeader->queueId = queueId;

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(GET_MAX_OFFSET_VALUE, pRequestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
	    switch (response->getCode())
	    {
	        case SUCCESS_VALUE:
	        {
	            GetMaxOffsetResponseHeader* ret = (GetMaxOffsetResponseHeader*)response->getCommandCustomHeader();
	            return ret->offset;
	        }
	        default:
	            break;
	    }
	    //THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
	}

    //THROW_MQEXCEPTION(MQClientException, "getMaxOffset failed", -1);
    return -1;
}


std::list<std::string> MQClientAPIImpl::getConsumerIdListByGroup(const std::string& addr,
        const std::string& consumerGroup,
        int timeoutMillis)
{
    std::string consumerGroupWithProjectGroup = consumerGroup;
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        consumerGroupWithProjectGroup =
            VirtualEnvUtil::buildWithProjectGroup(consumerGroup, m_projectGroupPrefix);
    }

    GetConsumerListByGroupRequestHeader* requestHeader = new GetConsumerListByGroupRequestHeader();
    requestHeader->consumerGroup = consumerGroupWithProjectGroup;

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(GET_CONSUMER_LIST_BY_GROUP_VALUE, requestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case SUCCESS_VALUE:
            {
                if (response->getBody() != NULL)
                {
                    GetConsumerListByGroupResponseBody* body =
                        GetConsumerListByGroupResponseBody::decode((char*)response->getBody(), response->getBodyLen());
                    std::list<std::string> ret = body->getConsumerIdList();
                    delete body;
                    return ret;
                }
            }
            default:
                break;
        }

        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    THROW_MQEXCEPTION(MQClientException, "getConsumerIdListByGroup failed", -1);
}

long long MQClientAPIImpl::getMinOffset(const std::string& addr,
                                        const std::string& topic,
                                        int queueId,
                                        int timeoutMillis)
{
    std::string topicWithProjectGroup = topic;
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        topicWithProjectGroup = VirtualEnvUtil::buildWithProjectGroup(topic, m_projectGroupPrefix);
    }

	GetMinOffsetRequestHeader* pRequestHeader = new GetMinOffsetRequestHeader();
    pRequestHeader->topic = topicWithProjectGroup;
    pRequestHeader->queueId = queueId;

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(GET_MIN_OFFSET_VALUE, pRequestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
	    switch (response->getCode())
	    {
	        case SUCCESS_VALUE:
	        {
	            GetMinOffsetResponseHeader* ret = (GetMinOffsetResponseHeader*)response->getCommandCustomHeader();
	            return ret->offset;
	        }
	        default:
	            break;
	    }
	    //THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
	}

    //THROW_MQEXCEPTION(MQClientException, "getMinOffset failed", -1);
    return -1;
}

long long MQClientAPIImpl::getEarliestMsgStoretime(const std::string& addr,
        const std::string& topic,
        int queueId,
        int timeoutMillis)
{
    std::string topicWithProjectGroup = topic;
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        topicWithProjectGroup = VirtualEnvUtil::buildWithProjectGroup(topic, m_projectGroupPrefix);
    }

	GetEarliestMsgStoretimeRequestHeader* pRequestHeader = new GetEarliestMsgStoretimeRequestHeader();
    pRequestHeader->topic = topicWithProjectGroup;
    pRequestHeader->queueId = queueId;

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(GET_EARLIEST_MSG_STORETIME_VALUE, pRequestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
	    switch (response->getCode())
	    {
	        case SUCCESS_VALUE:
	        {
	            GetEarliestMsgStoretimeResponseHeader* ret = (GetEarliestMsgStoretimeResponseHeader*)response->getCommandCustomHeader();
	            return ret->timestamp;
	        }
	        default:
	            break;
	    }
	    THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
	}

    THROW_MQEXCEPTION(MQClientException, "getEarliestMsgStoretime failed", -1);
}

long long MQClientAPIImpl::queryConsumerOffset(const std::string& addr,
        QueryConsumerOffsetRequestHeader* pRequestHeader,
        int timeoutMillis)
{
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        pRequestHeader->consumerGroup = VirtualEnvUtil::buildWithProjectGroup(
                                            pRequestHeader->consumerGroup, m_projectGroupPrefix);
        pRequestHeader->topic = VirtualEnvUtil::buildWithProjectGroup(pRequestHeader->topic,
                                m_projectGroupPrefix);
    }

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(QUERY_CONSUMER_OFFSET_VALUE, pRequestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
	    switch (response->getCode())
	    {
	        case SUCCESS_VALUE:
	        {
	            QueryConsumerOffsetResponseHeader* ret = (QueryConsumerOffsetResponseHeader*)response->getCommandCustomHeader();
	            long long offset = ret->offset;
	            return offset;
	        }
	        default:
	            break;
	    }
	    THROW_MQEXCEPTION(MQBrokerException, response->getRemark(), response->getCode());
	}

    THROW_MQEXCEPTION(MQClientException, "queryConsumerOffset failed", -1);
    return -1;
}

void MQClientAPIImpl::updateConsumerOffset(const std::string& addr,
        UpdateConsumerOffsetRequestHeader* pRequestHeader,
        int timeoutMillis)
{
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        pRequestHeader->consumerGroup = VirtualEnvUtil::buildWithProjectGroup(
                                            pRequestHeader->consumerGroup, m_projectGroupPrefix);
        pRequestHeader->topic = VirtualEnvUtil::buildWithProjectGroup(
                                    pRequestHeader->topic, m_projectGroupPrefix);
    }

    RemotingCommandPtr request = RemotingCommand::createRequestCommand(UPDATE_CONSUMER_OFFSET_VALUE, pRequestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case SUCCESS_VALUE:
            {
                return;
            }
            default:
                break;
        }

        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    THROW_MQEXCEPTION(MQClientException, "updateConsumerOffset failed", -1);
}

void MQClientAPIImpl::updateConsumerOffsetOneway(const std::string& addr,
        UpdateConsumerOffsetRequestHeader* pRequestHeader,
        int timeoutMillis)
{
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        pRequestHeader->consumerGroup = VirtualEnvUtil::buildWithProjectGroup(
                                            pRequestHeader->consumerGroup, m_projectGroupPrefix);
        pRequestHeader->topic = VirtualEnvUtil::buildWithProjectGroup(pRequestHeader->topic,
                                m_projectGroupPrefix);
    }

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(UPDATE_CONSUMER_OFFSET_VALUE, pRequestHeader);

    m_pRemotingClient->invokeOneway(addr, request, timeoutMillis);
}

void MQClientAPIImpl::sendHearbeat(const std::string& addr, HeartbeatData* pHeartbeatData, int timeoutMillis)
{
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        std::set<ConsumerData>& consumerDatas = pHeartbeatData->getConsumerDataSet();
        std::set<ConsumerData>::iterator it = consumerDatas.begin();
        for (; it != consumerDatas.end(); it++)
        {
            ConsumerData& consumerData = (ConsumerData&)(*it);
            consumerData.groupName = VirtualEnvUtil::buildWithProjectGroup(consumerData.groupName,
                                     m_projectGroupPrefix);

            std::set<SubscriptionData>& subscriptionDatas = consumerData.subscriptionDataSet;
            std::set<SubscriptionData>::iterator itsub = subscriptionDatas.begin();
            for (; itsub != subscriptionDatas.end(); itsub++)
            {
                SubscriptionData& subscriptionData = (SubscriptionData&)(*itsub);
                subscriptionData.setTopic(VirtualEnvUtil::buildWithProjectGroup(
                                              subscriptionData.getTopic(), m_projectGroupPrefix));
            }
        }

        std::set<ProducerData>& producerDatas = pHeartbeatData->getProducerDataSet();
        std::set<ProducerData>::iterator itp = producerDatas.begin();
        for (; itp != producerDatas.end(); itp++)
        {
            ProducerData& producerData = (ProducerData&)(*itp);
            producerData.groupName = VirtualEnvUtil::buildWithProjectGroup(producerData.groupName,
                                     m_projectGroupPrefix);
        }
    }

    RemotingCommandPtr request = RemotingCommand::createRequestCommand(HEART_BEAT_VALUE, NULL);

    std::string body;
    pHeartbeatData->encode(body);
    request->setBody((char*)body.data(), body.length(), true);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case SUCCESS_VALUE:
            {
                return;
            }
            default:
                break;
        }

        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    THROW_MQEXCEPTION(MQClientException, "sendHearbeat failed", -1);
}

void MQClientAPIImpl::unregisterClient(const std::string& addr,
                                       const std::string& clientID,
                                       const std::string& producerGroup,
                                       const std::string& consumerGroup,
                                       int timeoutMillis)
{
    std::string producerGroupWithProjectGroup = producerGroup;
    std::string consumerGroupWithProjectGroup = consumerGroup;
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        producerGroupWithProjectGroup =
            VirtualEnvUtil::buildWithProjectGroup(producerGroup, m_projectGroupPrefix);
        consumerGroupWithProjectGroup =
            VirtualEnvUtil::buildWithProjectGroup(consumerGroup, m_projectGroupPrefix);
    }

    UnregisterClientRequestHeader* requestHeader = new UnregisterClientRequestHeader();
    requestHeader->clientID = (clientID);
    requestHeader->producerGroup = (producerGroupWithProjectGroup);
    requestHeader->consumerGroup = (consumerGroupWithProjectGroup);

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(UNREGISTER_CLIENT_VALUE, requestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case SUCCESS_VALUE:
                return;
            default:
                break;
        }

        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    THROW_MQEXCEPTION(MQClientException, "unregisterClient failed", -1);
}

void MQClientAPIImpl::endTransactionOneway(const std::string& addr,
        EndTransactionRequestHeader* pRequestHeader,
        const std::string& remark,
        int timeoutMillis)
{
    //TODO
}

void MQClientAPIImpl::queryMessage(const std::string& addr,
                                   QueryMessageRequestHeader* pRequestHeader,
                                   int timeoutMillis,
                                   InvokeCallback* pInvokeCallback)
{
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        pRequestHeader->topic = VirtualEnvUtil::buildWithProjectGroup(pRequestHeader->topic,
        	m_projectGroupPrefix);
    }

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(QUERY_MESSAGE_VALUE, pRequestHeader);

    m_pRemotingClient->invokeAsync(addr, request, timeoutMillis, pInvokeCallback);
    return;
}

bool MQClientAPIImpl::registerClient(const std::string& addr, HeartbeatData& heartbeat, int timeoutMillis)
{
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        std::set<ConsumerData>& consumerDatas = heartbeat.getConsumerDataSet();
        std::set<ConsumerData>::iterator it = consumerDatas.begin();

        for (; it != consumerDatas.end(); it++)
        {
            ConsumerData& consumerData = (ConsumerData&)(*it);

            consumerData.groupName = VirtualEnvUtil::buildWithProjectGroup(consumerData.groupName,
                                     m_projectGroupPrefix);
            std::set<SubscriptionData>& subscriptionDatas = consumerData.subscriptionDataSet;
            std::set<SubscriptionData>::iterator itsub = subscriptionDatas.begin();

            for (; itsub != subscriptionDatas.end(); itsub++)
            {
                SubscriptionData& subscriptionData = (SubscriptionData&)(*itsub);
                subscriptionData.setTopic(VirtualEnvUtil::buildWithProjectGroup(
                                              subscriptionData.getTopic(), m_projectGroupPrefix));
            }
        }

        std::set<ProducerData>& producerDatas = heartbeat.getProducerDataSet();
        std::set<ProducerData>::iterator itp = producerDatas.begin();
        for (; itp != producerDatas.end(); itp++)
        {
            ProducerData& producerData = (ProducerData&)(*itp);
            producerData.groupName = VirtualEnvUtil::buildWithProjectGroup(producerData.groupName,
                                     m_projectGroupPrefix);
        }
    }

    RemotingCommandPtr request = RemotingCommand::createRequestCommand(HEART_BEAT_VALUE, NULL);

    std::string body;
    heartbeat.encode(body);

    request->setBody((char*)body.data(), body.length(), true);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    return (response && response->getCode() == SUCCESS_VALUE);
}

void MQClientAPIImpl::consumerSendMessageBack(
		const std::string& addr,
		MessageExt& msg,
        const std::string& consumerGroup,
        int delayLevel,
        int timeoutMillis)
{
    std::string consumerGroupWithProjectGroup = consumerGroup;
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        consumerGroupWithProjectGroup =
            VirtualEnvUtil::buildWithProjectGroup(consumerGroup, m_projectGroupPrefix);
        msg.setTopic(VirtualEnvUtil::buildWithProjectGroup(msg.getTopic(), m_projectGroupPrefix));
    }

    ConsumerSendMsgBackRequestHeader* requestHeader = new ConsumerSendMsgBackRequestHeader();
    requestHeader->group = consumerGroupWithProjectGroup;
    requestHeader->offset = msg.getCommitLogOffset();
    requestHeader->delayLevel = delayLevel;

    RemotingCommandPtr request = RemotingCommand::createRequestCommand(CONSUMER_SEND_MSG_BACK_VALUE, requestHeader);

	std::string brokerAddr = addr.empty() ? socketAddress2IPPort(msg.getStoreHost()) : addr;
    RemotingCommandPtr response = m_pRemotingClient->invokeSync(brokerAddr, request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case SUCCESS_VALUE:
                return;
                break;
            default:
                break;
        }

        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    THROW_MQEXCEPTION(MQClientException, "consumerSendMessageBack failed", -1);
}

std::set<MessageQueue> MQClientAPIImpl::lockBatchMQ(const std::string& addr,
        LockBatchRequestBody* pRequestBody,
        int timeoutMillis)
{
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        pRequestBody->setConsumerGroup((VirtualEnvUtil::buildWithProjectGroup(
                                            pRequestBody->getConsumerGroup(), m_projectGroupPrefix)));
        std::set<MessageQueue>& messageQueues = pRequestBody->getMqSet();
        std::set<MessageQueue>::iterator it = messageQueues.begin();

        for (; it != messageQueues.end(); it++)
        {
            MessageQueue& messageQueue = (MessageQueue&)(*it);
            messageQueue.setTopic(VirtualEnvUtil::buildWithProjectGroup(messageQueue.getTopic(),
                                  m_projectGroupPrefix));
        }
    }

    RemotingCommandPtr request = RemotingCommand::createRequestCommand(LOCK_BATCH_MQ_VALUE, NULL);

    std::string body;
    pRequestBody->encode(body);
    request->setBody((char*)body.data(), body.length(), true);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case SUCCESS_VALUE:
            {
                LockBatchResponseBody* responseBody =
                    LockBatchResponseBody::decode(response->getBody(), response->getBodyLen());
                std::set<MessageQueue> messageQueues = responseBody->getLockOKMQSet();

                if (!UtilAll::isBlank(m_projectGroupPrefix))
                {
                    std::set<MessageQueue>::iterator it = messageQueues.begin();

                    for (; it != messageQueues.end(); it++)
                    {
                        MessageQueue& messageQueue = (MessageQueue&)(*it);
                        messageQueue.setTopic(VirtualEnvUtil::clearProjectGroup(messageQueue.getTopic(),
                                              m_projectGroupPrefix));
                    }
                }
                return messageQueues;
            }
            default:
                break;
        }

        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    THROW_MQEXCEPTION(MQClientException, "lockBatchMQ failed", -1);
}

void MQClientAPIImpl::unlockBatchMQ(const std::string& addr,
                                    UnlockBatchRequestBody* pRequestBody,
                                    int timeoutMillis,
                                    bool oneway)
{
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        pRequestBody->setConsumerGroup((VirtualEnvUtil::buildWithProjectGroup(
                                            pRequestBody->getConsumerGroup(), m_projectGroupPrefix)));
        std::set<MessageQueue>& messageQueues = pRequestBody->getMqSet();
        std::set<MessageQueue>::iterator it = messageQueues.begin();

        for (; it != messageQueues.end(); it++)
        {
            MessageQueue& messageQueue = (MessageQueue&)(*it);
            messageQueue.setTopic(VirtualEnvUtil::buildWithProjectGroup(messageQueue.getTopic(),
                                  m_projectGroupPrefix));
        }
    }

    RemotingCommandPtr request = RemotingCommand::createRequestCommand(UNLOCK_BATCH_MQ_VALUE, NULL);

    std::string body;
    pRequestBody->encode(body);
    request->setBody((char*)body.data(), body.length(), true);

    if (oneway)
    {
        m_pRemotingClient->invokeOneway(addr, request, timeoutMillis);
    }
    else
    {
        RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
        if (response)
        {
            switch (response->getCode())
            {
                case SUCCESS_VALUE:
                    return;
                default:
                    break;
            }

            THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
        }

        THROW_MQEXCEPTION(MQClientException, "unlockBatchMQ failed", -1);
    }
}

TopicStatsTable MQClientAPIImpl::getTopicStatsInfo(const std::string& addr,
        const std::string& topic,
        int timeoutMillis)
{
    //TODO
    TopicStatsTable t;
    return t;
}

ConsumeStats MQClientAPIImpl::getConsumeStats(const std::string& addr,
        const std::string& consumerGroup,
        int timeoutMillis)
{
    //TODO
    ConsumeStats cs;
    return cs;
}

ProducerConnection* MQClientAPIImpl::getProducerConnectionList(const std::string& addr,
        const std::string& producerGroup,
        int timeoutMillis)
{
    //TODO
    return NULL;
}

ConsumerConnection* MQClientAPIImpl::getConsumerConnectionList(const std::string& addr,
        const std::string& consumerGroup,
        int timeoutMillis)
{
    //TODO
    return NULL;
}

KVTable MQClientAPIImpl::getBrokerRuntimeInfo(const std::string& addr,  int timeoutMillis)
{
    //TODO
    KVTable kv;
    return kv;
}

void MQClientAPIImpl::updateBrokerConfig(const std::string& addr,
        const std::map<std::string, std::string>&  properties,
        int timeoutMillis)
{
    //TODO
}

ClusterInfo* MQClientAPIImpl::getBrokerClusterInfo(int timeoutMillis)
{
   //TODO
    return NULL;
}

TopicRouteData* MQClientAPIImpl::getDefaultTopicRouteInfoFromNameServer(const std::string& topic,
        int timeoutMillis)
{
    GetRouteInfoRequestHeader* requestHeader = new GetRouteInfoRequestHeader();
    requestHeader->topic = topic;

    RemotingCommandPtr request = RemotingCommand::createRequestCommand(GET_ROUTEINTO_BY_TOPIC_VALUE, requestHeader);
    RemotingCommandPtr response = m_pRemotingClient->invokeSync("", request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case TOPIC_NOT_EXIST_VALUE:
            {
				// TODO LOG
                break;
            }
            case SUCCESS_VALUE:
            {
                int bodyLen = response->getBodyLen();
                const char* body = response->getBody();
                if (body)
                {
                    TopicRouteData* ret = TopicRouteData::encode(body, bodyLen);
                    return ret;
                }
            }
            default:
                break;
        }

        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    return NULL;
}

TopicRouteData* MQClientAPIImpl::getTopicRouteInfoFromNameServer(const std::string& topic, int timeoutMillis)
{
    std::string topicWithProjectGroup = topic;
    if (!UtilAll::isBlank(m_projectGroupPrefix))
    {
        topicWithProjectGroup = VirtualEnvUtil::buildWithProjectGroup(topic, m_projectGroupPrefix);
    }

    GetRouteInfoRequestHeader* requestHeader = new GetRouteInfoRequestHeader();
    requestHeader->topic = topicWithProjectGroup;

    RemotingCommandPtr request = RemotingCommand::createRequestCommand(GET_ROUTEINTO_BY_TOPIC_VALUE, requestHeader);
    RemotingCommandPtr response = m_pRemotingClient->invokeSync("", request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case TOPIC_NOT_EXIST_VALUE:
            {
            	if (topic != MixAll::DEFAULT_TOPIC)
            	{
            		RMQ_WARN("get Topic [{%s}] RouteInfoFromNameServer is not exist value", topic.c_str());
            	}
                break;
            }
            case SUCCESS_VALUE:
            {
                int bodyLen = response->getBodyLen();
                const char* body = response->getBody();
                if (body)
                {
                    TopicRouteData* ret = TopicRouteData::encode(body, bodyLen);
                    return ret;
                }
            }
            default:
                break;
        }

        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    return NULL;
}

TopicList* MQClientAPIImpl::getTopicListFromNameServer(int timeoutMillis)
{
    RemotingCommandPtr request = RemotingCommand::createRequestCommand(GET_ALL_TOPIC_LIST_FROM_NAMESERVER_VALUE, NULL);
    RemotingCommandPtr response = m_pRemotingClient->invokeSync("", request, timeoutMillis);
    if (response)
    {
        switch (response->getCode())
        {
            case SUCCESS_VALUE:
            {
                char* body = (char*)response->getBody();
                if (body != NULL)
                {
                    TopicList* topicList = TopicList::decode(body, response->getBodyLen());

                    if (!UtilAll::isBlank(m_projectGroupPrefix))
                    {
                        std::set<std::string> newTopicSet;

                        const std::set<std::string>& topics = topicList->getTopicList();
                        std::set<std::string>::const_iterator it = topics.begin();
                        for (; it != topics.end(); it++)
                        {
                            std::string topic = *it;
                            newTopicSet.insert(VirtualEnvUtil::clearProjectGroup(topic, m_projectGroupPrefix));
                        }

                        topicList->setTopicList(newTopicSet);
                    }

                    return topicList;
                }
            }
            default:
                break;
        }

        THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
    }

    return NULL;
}

int MQClientAPIImpl::wipeWritePermOfBroker(const std::string& namesrvAddr,
        const std::string& brokerName,
        int timeoutMillis)
{
    //TODO
    return 0;
}

void MQClientAPIImpl::deleteTopicInBroker(const std::string& addr,
        const std::string& topic,
        int timeoutMillis)
{
    //TODO
}

void MQClientAPIImpl::deleteTopicInNameServer(const std::string& addr,
        const std::string& topic,
        int timeoutMillis)
{
    //TODO
}

void MQClientAPIImpl::deleteSubscriptionGroup(const std::string& addr,
        const std::string& groupName,
        int timeoutMillis)
{
    //TODO
}

std::string MQClientAPIImpl::getKVConfigValue(const std::string& projectNamespace,
        const std::string& key,
        int timeoutMillis)
{
	GetKVConfigRequestHeader* pRequestHeader = new GetKVConfigRequestHeader();
    pRequestHeader->namespace_ = projectNamespace;
    pRequestHeader->key = key;

    RemotingCommandPtr request =
        RemotingCommand::createRequestCommand(GET_KV_CONFIG_VALUE, pRequestHeader);

    RemotingCommandPtr response = m_pRemotingClient->invokeSync("", request, timeoutMillis);
    if (response)
    {
	    switch (response->getCode())
	    {
	        case SUCCESS_VALUE:
	        {
	            GetKVConfigResponseHeader* ret = (GetKVConfigResponseHeader*)response->getCommandCustomHeader();
	            return ret->value;
	        }
	        default:
	            break;
	    }
	    THROW_MQEXCEPTION(MQClientException, response->getRemark(), response->getCode());
	}

    THROW_MQEXCEPTION(MQClientException, "getKVConfigValue failed", -1);
}

void MQClientAPIImpl::putKVConfigValue(const std::string& projectNamespace,
                                       const std::string& key,
                                       const std::string& value,
                                       int timeoutMillis)
{
    //TODO
}

void MQClientAPIImpl::deleteKVConfigValue(const std::string& projectNamespace,
        const std::string& key,
        int timeoutMillis)
{
    //TODO
}

std::string MQClientAPIImpl::getProjectGroupByIp(const std::string& ip,  int timeoutMillis)
{
    return getKVConfigValue(NamesrvUtil::NAMESPACE_PROJECT_CONFIG, ip, timeoutMillis);
}

std::string MQClientAPIImpl::getKVConfigByValue(const std::string& projectNamespace,
        const std::string& projectGroup,
        int timeoutMillis)
{
    //TODO
    return "";
}

KVTable MQClientAPIImpl::getKVListByNamespace(const std::string& projectNamespace,  int timeoutMillis)
{
    //TODO
    return KVTable();
}

void MQClientAPIImpl::deleteKVConfigByValue(const std::string& projectNamespace,
        const std::string& projectGroup,
        int timeoutMillis)
{
    //TODO
}

SendResult* MQClientAPIImpl::sendMessageSync(const std::string& addr,
        const std::string& brokerName,
        Message& msg,
        int timeoutMillis,
        RemotingCommand* request)
{
    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, request, timeoutMillis);
    return processSendResponse(brokerName, msg.getTopic(), response);
}

void MQClientAPIImpl::sendMessageAsync(const std::string& addr,
                                       const std::string& brokerName,
                                       Message& msg,
                                       int timeoutMillis,
                                       RemotingCommand* request,
                                       SendCallback* pSendCallback)
{
    ProducerInvokeCallback* callback = new ProducerInvokeCallback(pSendCallback, this, msg.getTopic(), brokerName);
    m_pRemotingClient->invokeAsync(addr, request, timeoutMillis, callback);
}

SendResult* MQClientAPIImpl::processSendResponse(const std::string& brokerName,
        const std::string& topic,
        RemotingCommand* pResponse)
{
    if (pResponse == NULL)
    {
        return NULL;
    }

    switch (pResponse->getCode())
    {
        case FLUSH_DISK_TIMEOUT_VALUE:
        case FLUSH_SLAVE_TIMEOUT_VALUE:
        case SLAVE_NOT_AVAILABLE_VALUE:
        {
            // TODO LOG
        }
        case SUCCESS_VALUE:
        {
            SendStatus sendStatus = SEND_OK;
            switch (pResponse->getCode())
            {
                case FLUSH_DISK_TIMEOUT_VALUE:
                    sendStatus = FLUSH_DISK_TIMEOUT;
                    break;
                case FLUSH_SLAVE_TIMEOUT_VALUE:
                    sendStatus = FLUSH_SLAVE_TIMEOUT;
                    break;
                case SLAVE_NOT_AVAILABLE_VALUE:
                    sendStatus = SLAVE_NOT_AVAILABLE;
                    break;
                case SUCCESS_VALUE:
                    sendStatus = SEND_OK;
                    break;
                default:
                    //assert false;
                    break;
            }

            SendMessageResponseHeader* responseHeader = (SendMessageResponseHeader*)pResponse->getCommandCustomHeader();
            MessageQueue messageQueue(topic, brokerName, responseHeader->queueId);
            SendResult* ret = new SendResult(sendStatus, responseHeader->msgId, messageQueue,
                                             responseHeader->queueOffset, m_projectGroupPrefix);

            return ret;
        }
        default:
            break;
    }

    THROW_MQEXCEPTION(MQClientException, pResponse->getRemark(), pResponse->getCode());
}

void MQClientAPIImpl::pullMessageAsync(const std::string& addr,
                                       RemotingCommand* pRequest,
                                       int timeoutMillis,
                                       PullCallback* pPullCallback)
{
    ConsumerInvokeCallback* callback = new ConsumerInvokeCallback(pPullCallback, this);
    m_pRemotingClient->invokeAsync(addr, pRequest, timeoutMillis, callback);
}

PullResult* MQClientAPIImpl::processPullResponse(RemotingCommand* pResponse)
{
    PullStatus pullStatus = NO_NEW_MSG;
    switch (pResponse->getCode())
    {
        case SUCCESS_VALUE:
            pullStatus = FOUND;
            break;
        case PULL_NOT_FOUND_VALUE:
            pullStatus = NO_NEW_MSG;
            break;
        case PULL_RETRY_IMMEDIATELY_VALUE:
            pullStatus = NO_MATCHED_MSG;
            break;
        case PULL_OFFSET_MOVED_VALUE:
            pullStatus = OFFSET_ILLEGAL;
            break;
        default:
            THROW_MQEXCEPTION(MQBrokerException, pResponse->getRemark(), pResponse->getCode());
            break;
    }

    PullMessageResponseHeader* responseHeader = (PullMessageResponseHeader*) pResponse->getCommandCustomHeader();
    std::list<MessageExt*> msgFoundList;
    return new PullResultExt(pullStatus, responseHeader->nextBeginOffset,
                             responseHeader->minOffset, responseHeader->maxOffset, msgFoundList,
                             responseHeader->suggestWhichBrokerId, pResponse->getBody(), pResponse->getBodyLen());
}

PullResult* MQClientAPIImpl::pullMessageSync(const std::string& addr,
        RemotingCommand* pRequest,
        int timeoutMillis)
{
    RemotingCommandPtr response = m_pRemotingClient->invokeSync(addr, pRequest, timeoutMillis);
    PullResult* result = processPullResponse(response);

    response->setBody(NULL, 0, false);
    return result;
}

}