-
Notifications
You must be signed in to change notification settings - Fork 3.6k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
HTTP lookups for c++ client lib #317
Merged
Merged
Changes from all commits
Commits
Show all changes
37 commits
Select commit
Hold shift + click to select a range
f614e3c
Blocking libcurl
2e8eed9
SSwitched to asio
ad97561
Working
c141f8b
Working so far
cb716b3
Not working
e749680
Base working now
5a4183c
Redirect left
14e0b89
Redirection remains
dc5b2a6
Added Removed mentions of my name
1943566
Url extension
980e211
Merge branch 'master' of https://github.com/yahoo/pulsar into httplookup
a6db72d
redirects handled
846ea50
Corrections in redirects
26a9aa3
Fixed socket resetting logic
9c82572
Handled carriage
1e59e49
added more error handling
542725f
added more error handling
7a3c76a
added more error handling
e60aaf2
added more error handling
1897790
Addressed Maurice's comments
8d1978a
using libcurl
12dc686
Removed HTTPWrapper files
b895c65
Removed mentions of my name
8926955
Compilation errors resolved
b8684aa
Formatting issues
1085bd8
Fixed JSON parsing logic since find API is not present in linux JSON …
cabffde
CR Comments and corrected header logic
6c0b24b
Corrected Auth part
5376cd5
Corrected Auth part
455e7f1
changed this to shared_from_this()
e94fe7e
Switching on CURLOPT_NOSIGNAL
40a4d33
Handled Matteo's Code Review Comments
631334e
Merge branch 'master' of https://github.com/yahoo/pulsar into httplookup
764efff
Fixed compilation error
6c7e011
Made curl initialization happen once
18fe553
Rearranged the class
2d0d9fc
Addresses Andrew's CR comments
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,200 @@ | ||
/** | ||
* Copyright 2016 Yahoo Inc. | ||
* | ||
* 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 <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; | ||
|
||
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_); | ||
|
||
// 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); | ||
|
||
// Make get call to server | ||
res = curl_easy_perform(handle); | ||
|
||
// Free header list | ||
curl_slist_free_all(list); | ||
|
||
switch(res) { | ||
case CURLE_OK: | ||
LOG_DEBUG("Response received successfully for url " << completeUrl); | ||
promise.setValue((requestType == PartitionMetaData) ? parsePartitionData(responseData) : parseLookupData(responseData)); | ||
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()); | ||
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(); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Used get instead of find since the json_cpp version in yahoo env doesn't support the find API. |
||
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); | ||
return lookupDataResultPtr; | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,61 @@ | ||
/** | ||
* Copyright 2016 Yahoo Inc. | ||
* | ||
* 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. | ||
*/ | ||
|
||
#ifndef PULSAR_CPP_HTTPLOOKUPSERVICE_H | ||
#define PULSAR_CPP_HTTPLOOKUPSERVICE_H | ||
|
||
#include <lib/LookupService.h> | ||
#include <lib/ClientImpl.h> | ||
#include <lib/Url.h> | ||
#include <json/value.h> | ||
#include <json/reader.h> | ||
#include <boost/bind.hpp> | ||
#include <curl/curl.h> | ||
|
||
namespace pulsar { | ||
class HTTPLookupService : public LookupService, public boost::enable_shared_from_this<HTTPLookupService> { | ||
class CurlInitializer { | ||
public: | ||
CurlInitializer() { | ||
// Once per application - https://curl.haxx.se/mail/lib-2015-11/0052.html | ||
curl_global_init (CURL_GLOBAL_ALL); | ||
} | ||
~CurlInitializer() { | ||
curl_global_cleanup(); | ||
} | ||
}; | ||
static CurlInitializer curlInitializer; | ||
enum RequestType {Lookup, PartitionMetaData}; | ||
typedef Promise<Result, LookupDataResultPtr> LookupPromise; | ||
ExecutorServiceProviderPtr executorProvider_; | ||
std::string adminUrl_; | ||
AuthenticationPtr authenticationPtr_; | ||
int lookupTimeoutInSeconds_; | ||
|
||
static LookupDataResultPtr parsePartitionData(const std::string&); | ||
static LookupDataResultPtr parseLookupData(const std::string&); | ||
void sendHTTPRequest(LookupPromise, const std::string, RequestType); | ||
public: | ||
HTTPLookupService(const std::string&, const ClientConfiguration&, const AuthenticationPtr&); | ||
|
||
Future<Result, LookupDataResultPtr> lookupAsync(const std::string&); | ||
|
||
Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const DestinationNamePtr&); | ||
}; | ||
|
||
} | ||
|
||
#endif //PULSAR_CPP_HTTPLOOKUPSERVICE_H |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can you also update README with the dependency list?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done