pulsar-client-cpp/lib/HTTPLookupService.cc - pulsar - Git at Google

 /**
  * 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 <lib/HTTPLookupService.h>

 #include <curl/curl.h>

 #include <boost/property_tree/json_parser.hpp>
 #include <boost/property_tree/ptree.hpp>
 namespace ptree = boost::property_tree;

 DECLARE_LOG_OBJECT()

 namespace pulsar {

 const static std::string V1_PATH = "/lookup/v2/destination/";
 const static std::string V2_PATH = "/lookup/v2/topic/";

 const static std::string ADMIN_PATH_V1 = "/admin/";
 const static std::string ADMIN_PATH_V2 = "/admin/v2/";

 const static int MAX_HTTP_REDIRECTS = 20;
 const static std::string PARTITION_METHOD_NAME = "partitions";
 const static int NUMBER_OF_LOOKUP_THREADS = 1;

 static inline bool needRedirection(long code) { return (code == 307 || code == 302 || code == 301); }

 HTTPLookupService::CurlInitializer::CurlInitializer() {
     // Once per application - https://curl.haxx.se/mail/lib-2015-11/0052.html
     curl_global_init(CURL_GLOBAL_ALL);
 }
 HTTPLookupService::CurlInitializer::~CurlInitializer() { curl_global_cleanup(); }

 HTTPLookupService::CurlInitializer HTTPLookupService::curlInitializer;

 HTTPLookupService::HTTPLookupService(const std::string &lookupUrl,
                                      const ClientConfiguration &clientConfiguration,
                                      const AuthenticationPtr &authData)
     : executorProvider_(std::make_shared<ExecutorServiceProvider>(NUMBER_OF_LOOKUP_THREADS)),
       authenticationPtr_(authData),
       lookupTimeoutInSeconds_(clientConfiguration.getOperationTimeoutSeconds()),
       tlsTrustCertsFilePath_(clientConfiguration.getTlsTrustCertsFilePath()),
       isUseTls_(clientConfiguration.isUseTls()),
       tlsAllowInsecure_(clientConfiguration.isTlsAllowInsecureConnection()),
       tlsValidateHostname_(clientConfiguration.isValidateHostName()) {
     if (lookupUrl[lookupUrl.length() - 1] == '/') {
         // Remove trailing '/'
         adminUrl_ = lookupUrl.substr(0, lookupUrl.length() - 1);
     } else {
         adminUrl_ = lookupUrl;
     }
 }

 Future<Result, LookupDataResultPtr> HTTPLookupService::lookupAsync(const std::string &topic) {
     LookupPromise promise;
     std::shared_ptr<TopicName> topicName = TopicName::get(topic);
     if (!topicName) {
         LOG_ERROR("Unable to parse topic - " << topic);
         promise.setFailed(ResultInvalidTopicName);
         return promise.getFuture();
     }

     std::stringstream completeUrlStream;
     if (topicName->isV2Topic()) {
         completeUrlStream << adminUrl_ << V2_PATH << topicName->getDomain() << "/" << topicName->getProperty()
                           << '/' << topicName->getNamespacePortion() << '/'
                           << topicName->getEncodedLocalName();
     } else {
         completeUrlStream << adminUrl_ << V1_PATH << topicName->getDomain() << "/" << topicName->getProperty()
                           << '/' << topicName->getCluster() << '/' << topicName->getNamespacePortion() << '/'
                           << topicName->getEncodedLocalName();
     }

     executorProvider_->get()->postWork(std::bind(&HTTPLookupService::handleLookupHTTPRequest,
                                                  shared_from_this(), promise, completeUrlStream.str(),
                                                  Lookup));
     return promise.getFuture();
 }

 Future<Result, LookupDataResultPtr> HTTPLookupService::getPartitionMetadataAsync(
     const TopicNamePtr &topicName) {
     LookupPromise promise;
     std::stringstream completeUrlStream;

     if (topicName->isV2Topic()) {
         completeUrlStream << adminUrl_ << ADMIN_PATH_V2 << topicName->getDomain() << '/'
                           << topicName->getProperty() << '/' << topicName->getNamespacePortion() << '/'
                           << topicName->getEncodedLocalName() << '/' << PARTITION_METHOD_NAME;
     } else {
         completeUrlStream << adminUrl_ << ADMIN_PATH_V1 << topicName->getDomain() << '/'
                           << topicName->getProperty() << '/' << topicName->getCluster() << '/'
                           << topicName->getNamespacePortion() << '/' << topicName->getEncodedLocalName()
                           << '/' << PARTITION_METHOD_NAME;
     }

     completeUrlStream << "?checkAllowAutoCreation=true";
     executorProvider_->get()->postWork(std::bind(&HTTPLookupService::handleLookupHTTPRequest,
                                                  shared_from_this(), promise, completeUrlStream.str(),
                                                  PartitionMetaData));
     return promise.getFuture();
 }

 Future<Result, NamespaceTopicsPtr> HTTPLookupService::getTopicsOfNamespaceAsync(
     const NamespaceNamePtr &nsName) {
     NamespaceTopicsPromise promise;
     std::stringstream completeUrlStream;

     if (nsName->isV2()) {
         completeUrlStream << adminUrl_ << ADMIN_PATH_V2 << "namespaces" << '/' << nsName->toString() << '/'
                           << "topics";
     } else {
         completeUrlStream << adminUrl_ << ADMIN_PATH_V1 << "namespaces" << '/' << nsName->toString() << '/'
                           << "destinations";
     }

     executorProvider_->get()->postWork(std::bind(&HTTPLookupService::handleNamespaceTopicsHTTPRequest,
                                                  shared_from_this(), promise, completeUrlStream.str()));
     return promise.getFuture();
 }

 static size_t curlWriteCallback(void *contents, size_t size, size_t nmemb, void *responseDataPtr) {
     ((std::string *)responseDataPtr)->append((char *)contents, size * nmemb);
     return size * nmemb;
 }

 void HTTPLookupService::handleNamespaceTopicsHTTPRequest(NamespaceTopicsPromise promise,
                                                          const std::string completeUrl) {
     std::string responseData;
     Result result = sendHTTPRequest(completeUrl, responseData);

     if (result != ResultOk) {
         promise.setFailed(result);
     } else {
         promise.setValue(parseNamespaceTopicsData(responseData));
     }
 }

 Result HTTPLookupService::sendHTTPRequest(std::string completeUrl, std::string &responseData) {
     uint16_t reqCount = 0;
     Result retResult = ResultOk;
     while (++reqCount <= MAX_HTTP_REDIRECTS) {
         CURL *handle;
         CURLcode res;
         std::string version = std::string("Pulsar-CPP-v") + _PULSAR_VERSION_INTERNAL_;
         handle = curl_easy_init();

         if (!handle) {
             LOG_ERROR("Unable to curl_easy_init for url " << completeUrl);
             // No curl_easy_cleanup required since handle not initialized
             return ResultLookupError;
         }
         // set URL
         curl_easy_setopt(handle, CURLOPT_URL, completeUrl.c_str());

         // Write callback
         curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, curlWriteCallback);
         curl_easy_setopt(handle, CURLOPT_WRITEDATA, &responseData);

         // New connection is made for each call
         curl_easy_setopt(handle, CURLOPT_FRESH_CONNECT, 1L);
         curl_easy_setopt(handle, CURLOPT_FORBID_REUSE, 1L);

         // Skipping signal handling - results in timeouts not honored during the DNS lookup
         curl_easy_setopt(handle, CURLOPT_NOSIGNAL, 1L);

         // Timer
         curl_easy_setopt(handle, CURLOPT_TIMEOUT, lookupTimeoutInSeconds_);

         // Set User Agent
         curl_easy_setopt(handle, CURLOPT_USERAGENT, version.c_str());

         // Fail if HTTP return code >=400
         curl_easy_setopt(handle, CURLOPT_FAILONERROR, 1L);

         // Authorization data
         AuthenticationDataPtr authDataContent;
         Result authResult = authenticationPtr_->getAuthData(authDataContent);
         if (authResult != ResultOk) {
             LOG_ERROR("Failed to getAuthData: " << authResult);
             curl_easy_cleanup(handle);
             return authResult;
         }
         struct curl_slist *list = NULL;
         if (authDataContent->hasDataForHttp()) {
             list = curl_slist_append(list, authDataContent->getHttpHeaders().c_str());
         }
         curl_easy_setopt(handle, CURLOPT_HTTPHEADER, list);

         // TLS
         if (isUseTls_) {
             if (curl_easy_setopt(handle, CURLOPT_SSLENGINE, NULL) != CURLE_OK) {
                 LOG_ERROR("Unable to load SSL engine for url " << completeUrl);
                 curl_easy_cleanup(handle);
                 return ResultConnectError;
             }
             if (curl_easy_setopt(handle, CURLOPT_SSLENGINE_DEFAULT, 1L) != CURLE_OK) {
                 LOG_ERROR("Unable to load SSL engine as default, for url " << completeUrl);
                 curl_easy_cleanup(handle);
                 return ResultConnectError;
             }
             curl_easy_setopt(handle, CURLOPT_SSLCERTTYPE, "PEM");

             if (tlsAllowInsecure_) {
                 curl_easy_setopt(handle, CURLOPT_SSL_VERIFYPEER, 0L);
             } else {
                 curl_easy_setopt(handle, CURLOPT_SSL_VERIFYPEER, 1L);
             }

             if (!tlsTrustCertsFilePath_.empty()) {
                 curl_easy_setopt(handle, CURLOPT_CAINFO, tlsTrustCertsFilePath_.c_str());
             }

             curl_easy_setopt(handle, CURLOPT_SSL_VERIFYHOST, tlsValidateHostname_ ? 1L : 0L);

             if (authDataContent->hasDataForTls()) {
                 curl_easy_setopt(handle, CURLOPT_SSLCERT, authDataContent->getTlsCertificates().c_str());
                 curl_easy_setopt(handle, CURLOPT_SSLKEY, authDataContent->getTlsPrivateKey().c_str());
             }
         }

         LOG_INFO("Curl [" << reqCount << "] Lookup Request sent for " << completeUrl);

         // Make get call to server
         res = curl_easy_perform(handle);

         long response_code = -1;
         curl_easy_getinfo(handle, CURLINFO_RESPONSE_CODE, &response_code);
         LOG_INFO("Response received for url " << completeUrl << " response_code " << response_code
                                               << " curl res " << res);

         // Free header list
         curl_slist_free_all(list);

         switch (res) {
             case CURLE_OK:
                 long response_code;
                 curl_easy_getinfo(handle, CURLINFO_RESPONSE_CODE, &response_code);
                 LOG_INFO("Response received for url " << completeUrl << " code " << response_code);
                 if (response_code == 200) {
                     retResult = ResultOk;
                 } else if (needRedirection(response_code)) {
                     char *url = NULL;
                     curl_easy_getinfo(handle, CURLINFO_REDIRECT_URL, &url);
                     LOG_INFO("Response from url " << completeUrl << " to new url " << url);
                     completeUrl = url;
                     retResult = ResultLookupError;
                 } else {
                     retResult = ResultLookupError;
                 }
                 break;
             case CURLE_COULDNT_CONNECT:
             case CURLE_COULDNT_RESOLVE_PROXY:
             case CURLE_COULDNT_RESOLVE_HOST:
             case CURLE_HTTP_RETURNED_ERROR:
                 LOG_ERROR("Response failed for url " << completeUrl << ". Error Code " << res);
                 retResult = ResultConnectError;
                 break;
             case CURLE_READ_ERROR:
                 LOG_ERROR("Response failed for url " << completeUrl << ". Error Code " << res);
                 retResult = ResultReadError;
                 break;
             case CURLE_OPERATION_TIMEDOUT:
                 LOG_ERROR("Response failed for url " << completeUrl << ". Error Code " << res);
                 retResult = ResultTimeout;
                 break;
             default:
                 LOG_ERROR("Response failed for url " << completeUrl << ". Error Code " << res);
                 retResult = ResultLookupError;
                 break;
         }
         curl_easy_cleanup(handle);
         if (!needRedirection(response_code)) {
             break;
         }
     }

     return retResult;
 }

 LookupDataResultPtr HTTPLookupService::parsePartitionData(const std::string &json) {
     ptree::ptree root;
     std::stringstream stream;
     stream << json;
     try {
         ptree::read_json(stream, root);
     } catch (ptree::json_parser_error &e) {
         LOG_ERROR("Failed to parse json of Partition Metadata: " << e.what() << "\nInput Json = " << json);
         return LookupDataResultPtr();
     }

     LookupDataResultPtr lookupDataResultPtr = std::make_shared<LookupDataResult>();
     lookupDataResultPtr->setPartitions(root.get<int>("partitions", 0));
     LOG_INFO("parsePartitionData = " << *lookupDataResultPtr);
     return lookupDataResultPtr;
 }

 LookupDataResultPtr HTTPLookupService::parseLookupData(const std::string &json) {
     ptree::ptree root;
     std::stringstream stream;
     stream << json;
     try {
         ptree::read_json(stream, root);
     } catch (ptree::json_parser_error &e) {
         LOG_ERROR("Failed to parse json : " << e.what() << "\nInput Json = " << json);
         return LookupDataResultPtr();
     }

     const std::string defaultNotFoundString = "Url Not found";
     const std::string brokerUrl = root.get<std::string>("brokerUrl", defaultNotFoundString);
     if (brokerUrl == defaultNotFoundString) {
         LOG_ERROR("malformed json! - brokerUrl not present" << json);
         return LookupDataResultPtr();
     }

     std::string brokerUrlTls = root.get<std::string>("brokerUrlTls", defaultNotFoundString);
     if (brokerUrlTls == defaultNotFoundString) {
         brokerUrlTls = root.get<std::string>("brokerUrlSsl", defaultNotFoundString);
         if (brokerUrlTls == defaultNotFoundString) {
             LOG_ERROR("malformed json! - brokerUrlTls not present" << json);
             return LookupDataResultPtr();
         }
     }

     LookupDataResultPtr lookupDataResultPtr = std::make_shared<LookupDataResult>();
     lookupDataResultPtr->setBrokerUrl(brokerUrl);
     lookupDataResultPtr->setBrokerUrlTls(brokerUrlTls);

     LOG_INFO("parseLookupData = " << *lookupDataResultPtr);
     return lookupDataResultPtr;
 }

 NamespaceTopicsPtr HTTPLookupService::parseNamespaceTopicsData(const std::string &json) {
     LOG_DEBUG("GetNamespaceTopics json = " << json);
     ptree::ptree root;
     std::stringstream stream;
     stream << json;
     try {
         ptree::read_json(stream, root);
     } catch (ptree::json_parser_error &e) {
         LOG_ERROR("Failed to parse json of Topics of Namespace: " << e.what() << "\nInput Json = " << json);
         return NamespaceTopicsPtr();
     }

     // passed in json is like: ["topic1", "topic2"...]
     // root will be an array of topics
     std::set<std::string> topicSet;
     // get all topics
     for (const auto &item : root) {
         // remove partition part
         const std::string topicName = item.second.get_value<std::string>();
         int pos = topicName.find("-partition-");
         std::string filteredName = topicName.substr(0, pos);

         // filter duped topic name
         if (topicSet.find(filteredName) == topicSet.end()) {
             topicSet.insert(filteredName);
         }
     }

     NamespaceTopicsPtr topicsResultPtr =
         std::make_shared<std::vector<std::string>>(topicSet.begin(), topicSet.end());

     return topicsResultPtr;
 }

 void HTTPLookupService::handleLookupHTTPRequest(LookupPromise promise, const std::string completeUrl,
                                                 RequestType requestType) {
     std::string responseData;
     Result result = sendHTTPRequest(completeUrl, responseData);

     if (result != ResultOk) {
         promise.setFailed(result);
     } else {
         promise.setValue((requestType == PartitionMetaData) ? parsePartitionData(responseData)
                                                             : parseLookupData(responseData));
     }
 }

 }  // namespace pulsar
	/**
	* 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 <lib/HTTPLookupService.h>

	#include <curl/curl.h>

	#include <boost/property_tree/json_parser.hpp>
	#include <boost/property_tree/ptree.hpp>
	namespace ptree = boost::property_tree;

	DECLARE_LOG_OBJECT()

	namespace pulsar {

	const static std::string V1_PATH = "/lookup/v2/destination/";
	const static std::string V2_PATH = "/lookup/v2/topic/";

	const static std::string ADMIN_PATH_V1 = "/admin/";
	const static std::string ADMIN_PATH_V2 = "/admin/v2/";

	const static int MAX_HTTP_REDIRECTS = 20;
	const static std::string PARTITION_METHOD_NAME = "partitions";
	const static int NUMBER_OF_LOOKUP_THREADS = 1;

	static inline bool needRedirection(long code) { return (code == 307 \|\| code == 302 \|\| code == 301); }

	HTTPLookupService::CurlInitializer::CurlInitializer() {
	// Once per application - https://curl.haxx.se/mail/lib-2015-11/0052.html
	curl_global_init(CURL_GLOBAL_ALL);
	}
	HTTPLookupService::CurlInitializer::~CurlInitializer() { curl_global_cleanup(); }

	HTTPLookupService::CurlInitializer HTTPLookupService::curlInitializer;

	HTTPLookupService::HTTPLookupService(const std::string &lookupUrl,
	const ClientConfiguration &clientConfiguration,
	const AuthenticationPtr &authData)
	: executorProvider_(std::make_shared<ExecutorServiceProvider>(NUMBER_OF_LOOKUP_THREADS)),
	authenticationPtr_(authData),
	lookupTimeoutInSeconds_(clientConfiguration.getOperationTimeoutSeconds()),
	tlsTrustCertsFilePath_(clientConfiguration.getTlsTrustCertsFilePath()),
	isUseTls_(clientConfiguration.isUseTls()),
	tlsAllowInsecure_(clientConfiguration.isTlsAllowInsecureConnection()),
	tlsValidateHostname_(clientConfiguration.isValidateHostName()) {
	if (lookupUrl[lookupUrl.length() - 1] == '/') {
	// Remove trailing '/'
	adminUrl_ = lookupUrl.substr(0, lookupUrl.length() - 1);
	} else {
	adminUrl_ = lookupUrl;
	}
	}

	Future<Result, LookupDataResultPtr> HTTPLookupService::lookupAsync(const std::string &topic) {
	LookupPromise promise;
	std::shared_ptr<TopicName> topicName = TopicName::get(topic);
	if (!topicName) {
	LOG_ERROR("Unable to parse topic - " << topic);
	promise.setFailed(ResultInvalidTopicName);
	return promise.getFuture();
	}

	std::stringstream completeUrlStream;
	if (topicName->isV2Topic()) {
	completeUrlStream << adminUrl_ << V2_PATH << topicName->getDomain() << "/" << topicName->getProperty()
	<< '/' << topicName->getNamespacePortion() << '/'
	<< topicName->getEncodedLocalName();
	} else {
	completeUrlStream << adminUrl_ << V1_PATH << topicName->getDomain() << "/" << topicName->getProperty()
	<< '/' << topicName->getCluster() << '/' << topicName->getNamespacePortion() << '/'
	<< topicName->getEncodedLocalName();
	}

	executorProvider_->get()->postWork(std::bind(&HTTPLookupService::handleLookupHTTPRequest,
	shared_from_this(), promise, completeUrlStream.str(),
	Lookup));
	return promise.getFuture();
	}

	Future<Result, LookupDataResultPtr> HTTPLookupService::getPartitionMetadataAsync(
	const TopicNamePtr &topicName) {
	LookupPromise promise;
	std::stringstream completeUrlStream;

	if (topicName->isV2Topic()) {
	completeUrlStream << adminUrl_ << ADMIN_PATH_V2 << topicName->getDomain() << '/'
	<< topicName->getProperty() << '/' << topicName->getNamespacePortion() << '/'
	<< topicName->getEncodedLocalName() << '/' << PARTITION_METHOD_NAME;
	} else {
	completeUrlStream << adminUrl_ << ADMIN_PATH_V1 << topicName->getDomain() << '/'
	<< topicName->getProperty() << '/' << topicName->getCluster() << '/'
	<< topicName->getNamespacePortion() << '/' << topicName->getEncodedLocalName()
	<< '/' << PARTITION_METHOD_NAME;
	}

	completeUrlStream << "?checkAllowAutoCreation=true";
	executorProvider_->get()->postWork(std::bind(&HTTPLookupService::handleLookupHTTPRequest,
	shared_from_this(), promise, completeUrlStream.str(),
	PartitionMetaData));
	return promise.getFuture();
	}

	Future<Result, NamespaceTopicsPtr> HTTPLookupService::getTopicsOfNamespaceAsync(
	const NamespaceNamePtr &nsName) {
	NamespaceTopicsPromise promise;
	std::stringstream completeUrlStream;

	if (nsName->isV2()) {
	completeUrlStream << adminUrl_ << ADMIN_PATH_V2 << "namespaces" << '/' << nsName->toString() << '/'
	<< "topics";
	} else {
	completeUrlStream << adminUrl_ << ADMIN_PATH_V1 << "namespaces" << '/' << nsName->toString() << '/'
	<< "destinations";
	}

	executorProvider_->get()->postWork(std::bind(&HTTPLookupService::handleNamespaceTopicsHTTPRequest,
	shared_from_this(), promise, completeUrlStream.str()));
	return promise.getFuture();
	}

	static size_t curlWriteCallback(void contents, size_t size, size_t nmemb, void responseDataPtr) {
	((std::string )responseDataPtr)->append((char )contents, size * nmemb);
	return size * nmemb;
	}

	void HTTPLookupService::handleNamespaceTopicsHTTPRequest(NamespaceTopicsPromise promise,
	const std::string completeUrl) {
	std::string responseData;
	Result result = sendHTTPRequest(completeUrl, responseData);

	if (result != ResultOk) {
	promise.setFailed(result);
	} else {
	promise.setValue(parseNamespaceTopicsData(responseData));
	}
	}

	Result HTTPLookupService::sendHTTPRequest(std::string completeUrl, std::string &responseData) {
	uint16_t reqCount = 0;
	Result retResult = ResultOk;
	while (++reqCount <= MAX_HTTP_REDIRECTS) {
	CURL *handle;
	CURLcode res;
	std::string version = std::string("Pulsar-CPP-v") + _PULSAR_VERSION_INTERNAL_;
	handle = curl_easy_init();

	if (!handle) {
	LOG_ERROR("Unable to curl_easy_init for url " << completeUrl);
	// No curl_easy_cleanup required since handle not initialized
	return ResultLookupError;
	}
	// set URL
	curl_easy_setopt(handle, CURLOPT_URL, completeUrl.c_str());

	// Write callback
	curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, curlWriteCallback);
	curl_easy_setopt(handle, CURLOPT_WRITEDATA, &responseData);

	// New connection is made for each call
	curl_easy_setopt(handle, CURLOPT_FRESH_CONNECT, 1L);
	curl_easy_setopt(handle, CURLOPT_FORBID_REUSE, 1L);

	// Skipping signal handling - results in timeouts not honored during the DNS lookup
	curl_easy_setopt(handle, CURLOPT_NOSIGNAL, 1L);

	// Timer
	curl_easy_setopt(handle, CURLOPT_TIMEOUT, lookupTimeoutInSeconds_);

	// Set User Agent
	curl_easy_setopt(handle, CURLOPT_USERAGENT, version.c_str());

	// Fail if HTTP return code >=400
	curl_easy_setopt(handle, CURLOPT_FAILONERROR, 1L);

	// Authorization data
	AuthenticationDataPtr authDataContent;
	Result authResult = authenticationPtr_->getAuthData(authDataContent);
	if (authResult != ResultOk) {
	LOG_ERROR("Failed to getAuthData: " << authResult);
	curl_easy_cleanup(handle);
	return authResult;
	}
	struct curl_slist *list = NULL;
	if (authDataContent->hasDataForHttp()) {
	list = curl_slist_append(list, authDataContent->getHttpHeaders().c_str());
	}
	curl_easy_setopt(handle, CURLOPT_HTTPHEADER, list);

	// TLS
	if (isUseTls_) {
	if (curl_easy_setopt(handle, CURLOPT_SSLENGINE, NULL) != CURLE_OK) {
	LOG_ERROR("Unable to load SSL engine for url " << completeUrl);
	curl_easy_cleanup(handle);
	return ResultConnectError;
	}
	if (curl_easy_setopt(handle, CURLOPT_SSLENGINE_DEFAULT, 1L) != CURLE_OK) {
	LOG_ERROR("Unable to load SSL engine as default, for url " << completeUrl);
	curl_easy_cleanup(handle);
	return ResultConnectError;
	}
	curl_easy_setopt(handle, CURLOPT_SSLCERTTYPE, "PEM");

	if (tlsAllowInsecure_) {
	curl_easy_setopt(handle, CURLOPT_SSL_VERIFYPEER, 0L);
	} else {
	curl_easy_setopt(handle, CURLOPT_SSL_VERIFYPEER, 1L);
	}

	if (!tlsTrustCertsFilePath_.empty()) {
	curl_easy_setopt(handle, CURLOPT_CAINFO, tlsTrustCertsFilePath_.c_str());
	}

	curl_easy_setopt(handle, CURLOPT_SSL_VERIFYHOST, tlsValidateHostname_ ? 1L : 0L);

	if (authDataContent->hasDataForTls()) {
	curl_easy_setopt(handle, CURLOPT_SSLCERT, authDataContent->getTlsCertificates().c_str());
	curl_easy_setopt(handle, CURLOPT_SSLKEY, authDataContent->getTlsPrivateKey().c_str());
	}
	}

	LOG_INFO("Curl [" << reqCount << "] Lookup Request sent for " << completeUrl);

	// Make get call to server
	res = curl_easy_perform(handle);

	long response_code = -1;
	curl_easy_getinfo(handle, CURLINFO_RESPONSE_CODE, &response_code);
	LOG_INFO("Response received for url " << completeUrl << " response_code " << response_code
	<< " curl res " << res);

	// Free header list
	curl_slist_free_all(list);

	switch (res) {
	case CURLE_OK:
	long response_code;
	curl_easy_getinfo(handle, CURLINFO_RESPONSE_CODE, &response_code);
	LOG_INFO("Response received for url " << completeUrl << " code " << response_code);
	if (response_code == 200) {
	retResult = ResultOk;
	} else if (needRedirection(response_code)) {
	char *url = NULL;
	curl_easy_getinfo(handle, CURLINFO_REDIRECT_URL, &url);
	LOG_INFO("Response from url " << completeUrl << " to new url " << url);
	completeUrl = url;
	retResult = ResultLookupError;
	} else {
	retResult = ResultLookupError;
	}
	break;
	case CURLE_COULDNT_CONNECT:
	case CURLE_COULDNT_RESOLVE_PROXY:
	case CURLE_COULDNT_RESOLVE_HOST:
	case CURLE_HTTP_RETURNED_ERROR:
	LOG_ERROR("Response failed for url " << completeUrl << ". Error Code " << res);
	retResult = ResultConnectError;
	break;
	case CURLE_READ_ERROR:
	LOG_ERROR("Response failed for url " << completeUrl << ". Error Code " << res);
	retResult = ResultReadError;
	break;
	case CURLE_OPERATION_TIMEDOUT:
	LOG_ERROR("Response failed for url " << completeUrl << ". Error Code " << res);
	retResult = ResultTimeout;
	break;
	default:
	LOG_ERROR("Response failed for url " << completeUrl << ". Error Code " << res);
	retResult = ResultLookupError;
	break;
	}
	curl_easy_cleanup(handle);
	if (!needRedirection(response_code)) {
	break;
	}
	}

	return retResult;
	}

	LookupDataResultPtr HTTPLookupService::parsePartitionData(const std::string &json) {
	ptree::ptree root;
	std::stringstream stream;
	stream << json;
	try {
	ptree::read_json(stream, root);
	} catch (ptree::json_parser_error &e) {
	LOG_ERROR("Failed to parse json of Partition Metadata: " << e.what() << "\nInput Json = " << json);
	return LookupDataResultPtr();
	}

	LookupDataResultPtr lookupDataResultPtr = std::make_shared<LookupDataResult>();
	lookupDataResultPtr->setPartitions(root.get<int>("partitions", 0));
	LOG_INFO("parsePartitionData = " << *lookupDataResultPtr);
	return lookupDataResultPtr;
	}

	LookupDataResultPtr HTTPLookupService::parseLookupData(const std::string &json) {
	ptree::ptree root;
	std::stringstream stream;
	stream << json;
	try {
	ptree::read_json(stream, root);
	} catch (ptree::json_parser_error &e) {
	LOG_ERROR("Failed to parse json : " << e.what() << "\nInput Json = " << json);
	return LookupDataResultPtr();
	}

	const std::string defaultNotFoundString = "Url Not found";
	const std::string brokerUrl = root.get<std::string>("brokerUrl", defaultNotFoundString);
	if (brokerUrl == defaultNotFoundString) {
	LOG_ERROR("malformed json! - brokerUrl not present" << json);
	return LookupDataResultPtr();
	}

	std::string brokerUrlTls = root.get<std::string>("brokerUrlTls", defaultNotFoundString);
	if (brokerUrlTls == defaultNotFoundString) {
	brokerUrlTls = root.get<std::string>("brokerUrlSsl", defaultNotFoundString);
	if (brokerUrlTls == defaultNotFoundString) {
	LOG_ERROR("malformed json! - brokerUrlTls not present" << json);
	return LookupDataResultPtr();
	}
	}

	LookupDataResultPtr lookupDataResultPtr = std::make_shared<LookupDataResult>();
	lookupDataResultPtr->setBrokerUrl(brokerUrl);
	lookupDataResultPtr->setBrokerUrlTls(brokerUrlTls);

	LOG_INFO("parseLookupData = " << *lookupDataResultPtr);
	return lookupDataResultPtr;
	}

	NamespaceTopicsPtr HTTPLookupService::parseNamespaceTopicsData(const std::string &json) {
	LOG_DEBUG("GetNamespaceTopics json = " << json);
	ptree::ptree root;
	std::stringstream stream;
	stream << json;
	try {
	ptree::read_json(stream, root);
	} catch (ptree::json_parser_error &e) {
	LOG_ERROR("Failed to parse json of Topics of Namespace: " << e.what() << "\nInput Json = " << json);
	return NamespaceTopicsPtr();
	}

	// passed in json is like: ["topic1", "topic2"...]
	// root will be an array of topics
	std::set<std::string> topicSet;
	// get all topics
	for (const auto &item : root) {
	// remove partition part
	const std::string topicName = item.second.get_value<std::string>();
	int pos = topicName.find("-partition-");
	std::string filteredName = topicName.substr(0, pos);

	// filter duped topic name
	if (topicSet.find(filteredName) == topicSet.end()) {
	topicSet.insert(filteredName);
	}
	}

	NamespaceTopicsPtr topicsResultPtr =
	std::make_shared<std::vector<std::string>>(topicSet.begin(), topicSet.end());

	return topicsResultPtr;
	}

	void HTTPLookupService::handleLookupHTTPRequest(LookupPromise promise, const std::string completeUrl,
	RequestType requestType) {
	std::string responseData;
	Result result = sendHTTPRequest(completeUrl, responseData);

	if (result != ResultOk) {
	promise.setFailed(result);
	} else {
	promise.setValue((requestType == PartitionMetaData) ? parsePartitionData(responseData)
	: parseLookupData(responseData));
	}
	}

	} // namespace pulsar