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>

 DECLARE_LOG_OBJECT()

 namespace pulsar {
     const static std::string V2_PATH = "/lookup/v2/destination/";
     const static std::string PARTITION_PATH = "/admin/persistent/";
     const static int MAX_HTTP_REDIRECTS = 20;
     const static std::string PARTITION_METHOD_NAME = "partitions";
     const static int NUMBER_OF_LOOKUP_THREADS = 1;

     HTTPLookupService::CurlInitializer HTTPLookupService::curlInitializer;

     HTTPLookupService::HTTPLookupService(const std::string &lookupUrl,
             const ClientConfiguration &clientConfiguration,
             const AuthenticationPtr &authData)
     : executorProvider_(boost::make_shared<ExecutorServiceProvider>(NUMBER_OF_LOOKUP_THREADS)),
     authenticationPtr_(authData),
     lookupTimeoutInSeconds_(clientConfiguration.getOperationTimeoutSeconds()) {
         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 &destinationName) {
         LookupPromise promise;
         boost::shared_ptr<DestinationName> dn = DestinationName::get(destinationName);
         if (!dn) {
             LOG_ERROR("Unable to parse destination - " << destinationName);
             promise.setFailed(ResultInvalidTopicName);
             return promise.getFuture();
         }

         std::stringstream completeUrlStream;
         completeUrlStream << adminUrl_ << V2_PATH << "persistent/" << dn->getProperty() << '/' << dn->getCluster()
         << '/' << dn->getNamespacePortion() << '/' << dn->getEncodedLocalName();
         executorProvider_->get()->postWork(boost::bind(&HTTPLookupService::sendHTTPRequest, shared_from_this(), promise, completeUrlStream.str(), Lookup));
         return promise.getFuture();
     }

     Future<Result, LookupDataResultPtr> HTTPLookupService::getPartitionMetadataAsync(const DestinationNamePtr &dn) {
         LookupPromise promise;
         std::stringstream completeUrlStream;
         completeUrlStream << adminUrl_ << PARTITION_PATH << dn->getProperty() << '/' << dn->getCluster()
         << '/' << dn->getNamespacePortion() << '/' << dn->getEncodedLocalName() << '/'
         << PARTITION_METHOD_NAME;
         executorProvider_->get()->postWork(boost::bind(&HTTPLookupService::sendHTTPRequest, shared_from_this(), promise, completeUrlStream.str(), PartitionMetaData));
         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::sendHTTPRequest(LookupPromise promise, const std::string completeUrl,
             RequestType requestType) {
         CURL *handle;
         CURLcode res;
         std::string responseData;
         std::string version = std::string("Pulsar-CPP-v") + _PULSAR_VERSION_;
         handle = curl_easy_init();

         if(!handle) {
             LOG_ERROR("Unable to curl_easy_init for url " << completeUrl);
             promise.setFailed(ResultLookupError);
             // No curl_easy_cleanup required since handle not initialized
             return;
         }
         // 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());

         // Redirects
         curl_easy_setopt(handle, CURLOPT_FOLLOWLOCATION, 1L);
         curl_easy_setopt(handle, CURLOPT_MAXREDIRS, MAX_HTTP_REDIRECTS);

         // 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("All Authentication methods should have AuthenticationData and return true on getAuthData for url " << completeUrl);
             promise.setFailed(authResult);
             curl_easy_cleanup(handle);
             return;
         }
         struct curl_slist *list = NULL;
         if (authDataContent->hasDataForHttp()) {
             list = curl_slist_append(list, authDataContent->getHttpHeaders().c_str());
         }
         curl_easy_setopt(handle, CURLOPT_HTTPHEADER, list);

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

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

         // 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) {
                     promise.setValue((requestType == PartitionMetaData) ? parsePartitionData(responseData) : parseLookupData(responseData));
                 } else {
                     promise.setFailed(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);
                 promise.setFailed(ResultConnectError);
                 break;
             case CURLE_READ_ERROR:
                 LOG_ERROR("Response failed for url "<<completeUrl << ". Error Code "<<res);
                 promise.setFailed(ResultReadError);
                 break;
             case CURLE_OPERATION_TIMEDOUT:
                 LOG_ERROR("Response failed for url "<<completeUrl << ". Error Code "<<res);
                 promise.setFailed(ResultTimeout);
                 break;
             default:
                 LOG_ERROR("Response failed for url "<<completeUrl << ". Error Code "<<res);
                 promise.setFailed(ResultLookupError);
                 break;
         }
         curl_easy_cleanup(handle);
     }

     LookupDataResultPtr HTTPLookupService::parsePartitionData(const std::string &json) {
         Json::Value root;
         Json::Reader reader;
         if (!reader.parse(json, root, false)) {
             LOG_ERROR("Failed to parse json of Partition Metadata: " << reader.getFormatedErrorMessages()
                     << "\nInput Json = " << json);
             return LookupDataResultPtr();
         }
         LookupDataResultPtr lookupDataResultPtr = boost::make_shared<LookupDataResult>();
         lookupDataResultPtr->setPartitions(root.get("partitions", 0).asInt());
         LOG_INFO("parsePartitionData = "<<*lookupDataResultPtr);
         return lookupDataResultPtr;
     }

     LookupDataResultPtr HTTPLookupService::parseLookupData(const std::string &json) {
         Json::Value root;
         Json::Reader reader;
         if (!reader.parse(json, root, false)) {
             LOG_ERROR("Failed to parse json : " << reader.getFormatedErrorMessages()
                     << "\nInput Json = " << json);
             return LookupDataResultPtr();
         }
         const std::string defaultNotFoundString = "Url Not found";
         const std::string brokerUrl = root.get("brokerUrl", defaultNotFoundString).asString();
         if (brokerUrl == defaultNotFoundString) {
             LOG_ERROR("malformed json! - brokerUrl not present" << json);
             return LookupDataResultPtr();
         }

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

         LookupDataResultPtr lookupDataResultPtr = boost::make_shared<LookupDataResult>();
         lookupDataResultPtr->setBrokerUrl(brokerUrl);
         lookupDataResultPtr->setBrokerUrlSsl(brokerUrlSsl);
         LOG_INFO("parseLookupData = "<<*lookupDataResultPtr);
         return lookupDataResultPtr;
     }
 }
	/**
	* 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>

	DECLARE_LOG_OBJECT()

	namespace pulsar {
	const static std::string V2_PATH = "/lookup/v2/destination/";
	const static std::string PARTITION_PATH = "/admin/persistent/";
	const static int MAX_HTTP_REDIRECTS = 20;
	const static std::string PARTITION_METHOD_NAME = "partitions";
	const static int NUMBER_OF_LOOKUP_THREADS = 1;

	HTTPLookupService::CurlInitializer HTTPLookupService::curlInitializer;

	HTTPLookupService::HTTPLookupService(const std::string &lookupUrl,
	const ClientConfiguration &clientConfiguration,
	const AuthenticationPtr &authData)
	: executorProvider_(boost::make_shared<ExecutorServiceProvider>(NUMBER_OF_LOOKUP_THREADS)),
	authenticationPtr_(authData),
	lookupTimeoutInSeconds_(clientConfiguration.getOperationTimeoutSeconds()) {
	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 &destinationName) {
	LookupPromise promise;
	boost::shared_ptr<DestinationName> dn = DestinationName::get(destinationName);
	if (!dn) {
	LOG_ERROR("Unable to parse destination - " << destinationName);
	promise.setFailed(ResultInvalidTopicName);
	return promise.getFuture();
	}

	std::stringstream completeUrlStream;
	completeUrlStream << adminUrl_ << V2_PATH << "persistent/" << dn->getProperty() << '/' << dn->getCluster()
	<< '/' << dn->getNamespacePortion() << '/' << dn->getEncodedLocalName();
	executorProvider_->get()->postWork(boost::bind(&HTTPLookupService::sendHTTPRequest, shared_from_this(), promise, completeUrlStream.str(), Lookup));
	return promise.getFuture();
	}

	Future<Result, LookupDataResultPtr> HTTPLookupService::getPartitionMetadataAsync(const DestinationNamePtr &dn) {
	LookupPromise promise;
	std::stringstream completeUrlStream;
	completeUrlStream << adminUrl_ << PARTITION_PATH << dn->getProperty() << '/' << dn->getCluster()
	<< '/' << dn->getNamespacePortion() << '/' << dn->getEncodedLocalName() << '/'
	<< PARTITION_METHOD_NAME;
	executorProvider_->get()->postWork(boost::bind(&HTTPLookupService::sendHTTPRequest, shared_from_this(), promise, completeUrlStream.str(), PartitionMetaData));
	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::sendHTTPRequest(LookupPromise promise, const std::string completeUrl,
	RequestType requestType) {
	CURL *handle;
	CURLcode res;
	std::string responseData;
	std::string version = std::string("Pulsar-CPP-v") + _PULSAR_VERSION_;
	handle = curl_easy_init();

	if(!handle) {
	LOG_ERROR("Unable to curl_easy_init for url " << completeUrl);
	promise.setFailed(ResultLookupError);
	// No curl_easy_cleanup required since handle not initialized
	return;
	}
	// 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());

	// Redirects
	curl_easy_setopt(handle, CURLOPT_FOLLOWLOCATION, 1L);
	curl_easy_setopt(handle, CURLOPT_MAXREDIRS, MAX_HTTP_REDIRECTS);

	// 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("All Authentication methods should have AuthenticationData and return true on getAuthData for url " << completeUrl);
	promise.setFailed(authResult);
	curl_easy_cleanup(handle);
	return;
	}
	struct curl_slist *list = NULL;
	if (authDataContent->hasDataForHttp()) {
	list = curl_slist_append(list, authDataContent->getHttpHeaders().c_str());
	}
	curl_easy_setopt(handle, CURLOPT_HTTPHEADER, list);

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

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

	// 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) {
	promise.setValue((requestType == PartitionMetaData) ? parsePartitionData(responseData) : parseLookupData(responseData));
	} else {
	promise.setFailed(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);
	promise.setFailed(ResultConnectError);
	break;
	case CURLE_READ_ERROR:
	LOG_ERROR("Response failed for url "<<completeUrl << ". Error Code "<<res);
	promise.setFailed(ResultReadError);
	break;
	case CURLE_OPERATION_TIMEDOUT:
	LOG_ERROR("Response failed for url "<<completeUrl << ". Error Code "<<res);
	promise.setFailed(ResultTimeout);
	break;
	default:
	LOG_ERROR("Response failed for url "<<completeUrl << ". Error Code "<<res);
	promise.setFailed(ResultLookupError);
	break;
	}
	curl_easy_cleanup(handle);
	}

	LookupDataResultPtr HTTPLookupService::parsePartitionData(const std::string &json) {
	Json::Value root;
	Json::Reader reader;
	if (!reader.parse(json, root, false)) {
	LOG_ERROR("Failed to parse json of Partition Metadata: " << reader.getFormatedErrorMessages()
	<< "\nInput Json = " << json);
	return LookupDataResultPtr();
	}
	LookupDataResultPtr lookupDataResultPtr = boost::make_shared<LookupDataResult>();
	lookupDataResultPtr->setPartitions(root.get("partitions", 0).asInt());
	LOG_INFO("parsePartitionData = "<<*lookupDataResultPtr);
	return lookupDataResultPtr;
	}

	LookupDataResultPtr HTTPLookupService::parseLookupData(const std::string &json) {
	Json::Value root;
	Json::Reader reader;
	if (!reader.parse(json, root, false)) {
	LOG_ERROR("Failed to parse json : " << reader.getFormatedErrorMessages()
	<< "\nInput Json = " << json);
	return LookupDataResultPtr();
	}
	const std::string defaultNotFoundString = "Url Not found";
	const std::string brokerUrl = root.get("brokerUrl", defaultNotFoundString).asString();
	if (brokerUrl == defaultNotFoundString) {
	LOG_ERROR("malformed json! - brokerUrl not present" << json);
	return LookupDataResultPtr();
	}

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

	LookupDataResultPtr lookupDataResultPtr = boost::make_shared<LookupDataResult>();
	lookupDataResultPtr->setBrokerUrl(brokerUrl);
	lookupDataResultPtr->setBrokerUrlSsl(brokerUrlSsl);
	LOG_INFO("parseLookupData = "<<*lookupDataResultPtr);
	return lookupDataResultPtr;
	}
	}