/*

* 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.

*/

#include "MQClient.h"

#include "Logging.h"

#include "MQClientFactory.h"

#include "MQClientManager.h"

#include "TopicPublishInfo.h"

#include "UtilAll.h"

namespace rocketmq {

#define ROCKETMQCPP_VERSION "1.0.1"

#define BUILD_DATE "03-14-2018"

// display version: strings bin/librocketmq.so |grep VERSION

const char *rocketmq_build_time =

"VERSION: " ROCKETMQCPP_VERSION ", BUILD DATE: " BUILD_DATE " ";

//<!************************************************************************

MQClient::MQClient() {

string NAMESRV_ADDR_ENV = "NAMESRV_ADDR";

if (const char *addr = getenv(NAMESRV_ADDR_ENV.c_str()))

m_namesrvAddr = addr;

else

m_namesrvAddr = "";

m_instanceName = "DEFAULT";

m_clientFactory = NULL;

m_serviceState = CREATE_JUST;

m_pullThreadNum = boost::thread::hardware_concurrency();

m_tcpConnectTimeout = 3000; // 3s

m_tcpTransportTryLockTimeout = 3; // 3s

m_unitName = "";

}

MQClient::~MQClient() {}

string MQClient::getMQClientId() const {

string clientIP = UtilAll::getLocalAddress();

string processId = UtilAll::to_string(getpid());

return processId + "-" + clientIP + "@" + m_instanceName;

}

//<!groupName;

const string &MQClient::getGroupName() const { return m_GroupName; }

void MQClient::setGroupName(const string &groupname) {

m_GroupName = groupname;

}

const string &MQClient::getNamesrvAddr() const { return m_namesrvAddr; }

void MQClient::setNamesrvAddr(const string &namesrvAddr) {

m_namesrvAddr = namesrvAddr;

}

const string &MQClient::getNamesrvDomain() const { return m_namesrvDomain; }

void MQClient::setNamesrvDomain(const string &namesrvDomain) {

m_namesrvDomain = namesrvDomain;

}

const string &MQClient::getInstanceName() const { return m_instanceName; }

void MQClient::setInstanceName(const string &instanceName) {

m_instanceName = instanceName;

}

void MQClient::createTopic(const string &key, const string &newTopic,

int queueNum) {

try {

getFactory()->createTopic(key, newTopic, queueNum, m_SessionCredentials);

} catch (MQException &e) {

LOG_ERROR(e.what());

}

}

int64 MQClient::earliestMsgStoreTime(const MQMessageQueue &mq) {

return getFactory()->earliestMsgStoreTime(mq, m_SessionCredentials);

}

QueryResult MQClient::queryMessage(const string &topic, const string &key,

int maxNum, int64 begin, int64 end) {

return getFactory()->queryMessage(topic, key, maxNum, begin, end,

m_SessionCredentials);

}

int64 MQClient::minOffset(const MQMessageQueue &mq) {

return getFactory()->minOffset(mq, m_SessionCredentials);

}

int64 MQClient::maxOffset(const MQMessageQueue &mq) {

return getFactory()->maxOffset(mq, m_SessionCredentials);

}

int64 MQClient::searchOffset(const MQMessageQueue &mq, uint64_t timestamp) {

return getFactory()->searchOffset(mq, timestamp, m_SessionCredentials);

}

MQMessageExt *MQClient::viewMessage(const string &msgId) {

return getFactory()->viewMessage(msgId, m_SessionCredentials);

}

vector<MQMessageQueue> MQClient::getTopicMessageQueueInfo(const string &topic) {

boost::weak_ptr<TopicPublishInfo> weak_topicPublishInfo(

getFactory()->tryToFindTopicPublishInfo(topic, m_SessionCredentials));

boost::shared_ptr<TopicPublishInfo> topicPublishInfo(

weak_topicPublishInfo.lock());

if (topicPublishInfo) {

return topicPublishInfo->getMessageQueueList();

}

THROW_MQEXCEPTION(

MQClientException,

"could not find MessageQueue Info of topic: [" + topic + "].", -1);

}

void MQClient::start() {

if (getFactory() == NULL) {

m_clientFactory = MQClientManager::getInstance()->getMQClientFactory(

getMQClientId(), m_pullThreadNum, m_tcpConnectTimeout,

m_tcpTransportTryLockTimeout, m_unitName);

}

LOG_INFO(

"MQClient "

"start,groupname:%s,clientID:%s,instanceName:%s,nameserveraddr:%s",

getGroupName().c_str(), getMQClientId().c_str(),

getInstanceName().c_str(), getNamesrvAddr().c_str());

}

void MQClient::shutdown() { m_clientFactory = NULL; }

MQClientFactory *MQClient::getFactory() const { return m_clientFactory; }

bool MQClient::isServiceStateOk() { return m_serviceState == RUNNING; }

void MQClient::setLogLevel(elogLevel inputLevel) {

ALOG_ADAPTER->setLogLevel(inputLevel);

}

elogLevel MQClient::getLogLevel() { return ALOG_ADAPTER->getLogLevel(); }

void MQClient::setLogFileSizeAndNum(int fileNum, long perFileSize) {

ALOG_ADAPTER->setLogFileNumAndSize(fileNum, perFileSize);

}

void MQClient::setTcpTransportPullThreadNum(int num) {

if (num > m_pullThreadNum) {

m_pullThreadNum = num;

}

}

const int MQClient::getTcpTransportPullThreadNum() const {

return m_pullThreadNum;

}

void MQClient::setTcpTransportConnectTimeout(uint64_t timeout) {

m_tcpConnectTimeout = timeout;

}

const uint64_t MQClient::getTcpTransportConnectTimeout() const {

return m_tcpConnectTimeout;

}

void MQClient::setTcpTransportTryLockTimeout(uint64_t timeout) {

if (timeout < 1000) {

timeout = 1000;

}

m_tcpTransportTryLockTimeout = timeout / 1000;

}

const uint64_t MQClient::getTcpTransportTryLockTimeout() const {

return m_tcpTransportTryLockTimeout;

}

void MQClient::setUnitName(string unitName) { m_unitName = unitName; }

const string &MQClient::getUnitName() { return m_unitName; }

void MQClient::setSessionCredentials(const string &input_accessKey,

const string &input_secretKey,

const string &input_onsChannel) {

m_SessionCredentials.setAccessKey(input_accessKey);

m_SessionCredentials.setSecretKey(input_secretKey);

m_SessionCredentials.setAuthChannel(input_onsChannel);

}

const SessionCredentials &MQClient::getSessionCredentials() const {

return m_SessionCredentials;

}

//<!************************************************************************

} //<!end namespace;