kapitan/IXWebSocket
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Feature/http async (#90)

Browse Source 
* unittest working / uses shared_ptr for a bunch of things 🗿

* fix command line tools

* fix ws + add doc

* add more logging
This commit is contained in:
Benjamin Sergeant 2019-06-05 17:04:24 -07:00 committed by GitHub
parent 012193c74e
commit 78b3d7ff2d
14 changed files with 490 additions and 223 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
README.md
View File

@ -129,33 +129,33 @@ Here is what the HTTP client API looks like. Note that HTTP client support is ve
// Preparation // Preparation
// //
HttpClient httpClient; HttpClient httpClient;
HttpRequestArgs args; HttpRequestArgsPtr args = httpClient.createRequest();
// Custom headers can be set // Custom headers can be set
WebSocketHttpHeaders headers; WebSocketHttpHeaders headers;
headers["Foo"] = "bar"; headers["Foo"] = "bar";
args.extraHeaders = headers; args->extraHeaders = headers;
// Timeout options // Timeout options
args.connectTimeout = connectTimeout; args->connectTimeout = connectTimeout;
args.transferTimeout = transferTimeout; args->transferTimeout = transferTimeout;
// Redirect options // Redirect options
args.followRedirects = followRedirects; args->followRedirects = followRedirects;
args.maxRedirects = maxRedirects; args->maxRedirects = maxRedirects;
// Misc // Misc
args.compress = compress; // Enable gzip compression args->compress = compress; // Enable gzip compression
args.verbose = verbose; args->verbose = verbose;
args.logger = [](const std::string& msg) args->logger = [](const std::string& msg)
{ {
std::cout << msg; std::cout << msg;
}; };
// //
// Request // Synchronous Request
// //
HttpResponse out; HttpResponsePtr out;
std::string url = "https://www.google.com"; std::string url = "https://www.google.com";
// HEAD request // HEAD request
@ -175,13 +175,30 @@ out = httpClient.post(url, std::string("foo=bar"), args);
// //
// Result // Result
// //
auto statusCode = std::get<0>(out); auto errorCode = response->errorCode; // Can be HttpErrorCode::Ok, HttpErrorCode::UrlMalformed, etc...
auto errorCode = std::get<1>(out); auto errorCode = response->errorCode; // 200, 404, etc...
auto responseHeaders = std::get<2>(out); auto responseHeaders = response->headers; // All the headers in a special case-insensitive unordered_map of (string, string)
auto payload = std::get<3>(out); auto payload = response->payload; // All the bytes from the response as an std::string
auto errorMsg = std::get<4>(out); auto errorMsg = response->errorMsg; // Descriptive error message in case of failure
auto uploadSize = std::get<5>(out); auto uploadSize = response->uploadSize; // Byte count of uploaded data
auto downloadSize = std::get<6>(out); auto downloadSize = response->downloadSize; // Byte count of downloaded data
//
// Asynchronous Request
//
bool async = true;
HttpClient httpClient(async);
auto args = httpClient.createRequest(url, HttpClient::kGet);
// Push the request to a queue,
bool ok = httpClient.performRequest(args, [](const HttpResponsePtr& response)
{
// This callback execute in a background thread. Make sure you uses appropriate protection such as mutex
auto statusCode = response->statusCode; // acess results
}
);
// ok will be false if your httpClient is not async
``` ```
## Build ## Build

23
docker/Dockerfile.ubuntu_bionic Normal file
View File

@ -0,0 +1,23 @@
# Build time
FROM ubuntu:bionic as build
ENV DEBIAN_FRONTEND noninteractive
RUN apt-get update
RUN apt-get -y install wget
RUN mkdir -p /tmp/cmake
WORKDIR /tmp/cmake
RUN wget https://github.com/Kitware/CMake/releases/download/v3.14.0/cmake-3.14.0-Linux-x86_64.tar.gz
RUN tar zxf cmake-3.14.0-Linux-x86_64.tar.gz
RUN apt-get -y install g++
RUN apt-get -y install libssl-dev
RUN apt-get -y install libz-dev
RUN apt-get -y install make
RUN apt-get -y install python
COPY . .
ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-Linux-x86_64/bin
ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
RUN ["make", "ws"]

166
ixwebsocket/IXHttpClient.cpp
View File

@ -24,21 +24,89 @@ namespace ix
const std::string HttpClient::kDel = "DEL"; const std::string HttpClient::kDel = "DEL";
const std::string HttpClient::kPut = "PUT"; const std::string HttpClient::kPut = "PUT";
HttpClient::HttpClient() HttpClient::HttpClient(bool async) : _async(async), _stop(false)
{ {
if (!_async) return;
_thread = std::thread(&HttpClient::run, this);
} }
HttpClient::~HttpClient() HttpClient::~HttpClient()
{ {
if (!_thread.joinable()) return;
_stop = true;
_condition.notify_one();
_thread.join();
} }
HttpResponse HttpClient::request( HttpRequestArgsPtr HttpClient::createRequest(const std::string& url,
const std::string& verb)
{
auto request = std::make_shared<HttpRequestArgs>();
request->url = url;
request->verb = verb;
return request;
}
bool HttpClient::performRequest(HttpRequestArgsPtr args,
const OnResponseCallback& onResponseCallback)
{
if (!_async) return false;
// Enqueue the task
{
// acquire lock
std::unique_lock<std::mutex> lock(_queueMutex);
// add the task
_queue.push(std::make_pair(args, onResponseCallback));
} // release lock
// wake up one thread
_condition.notify_one();
return true;
}
void HttpClient::run()
{
while (true)
{
HttpRequestArgsPtr args;
OnResponseCallback onResponseCallback;
{
std::unique_lock<std::mutex> lock(_queueMutex);
while (!_stop && _queue.empty())
{
_condition.wait(lock);
}
if (_stop) return;
auto p = _queue.front();
_queue.pop();
args = p.first;
onResponseCallback = p.second;
}
if (_stop) return;
HttpResponsePtr response = request(args->url, args->verb, args->body, args);
onResponseCallback(response);
if (_stop) return;
}
}
HttpResponsePtr HttpClient::request(
const std::string& url, const std::string& url,
const std::string& verb, const std::string& verb,
const std::string& body, const std::string& body,
const HttpRequestArgs& args, HttpRequestArgsPtr args,
int redirects) int redirects)
{ {
uint64_t uploadSize = 0; uint64_t uploadSize = 0;
@ -54,7 +122,7 @@ namespace ix
{ {
std::stringstream ss; std::stringstream ss;
ss << "Cannot parse url: " << url; ss << "Cannot parse url: " << url;
return HttpResponse(code, HttpErrorCode::UrlMalformed, return std::make_shared<HttpResponse>(code, HttpErrorCode::UrlMalformed,
headers, payload, ss.str(), headers, payload, ss.str(),
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -65,7 +133,7 @@ namespace ix
if (!_socket) if (!_socket)
{ {
return HttpResponse(code, HttpErrorCode::CannotCreateSocket, return std::make_shared<HttpResponse>(code, HttpErrorCode::CannotCreateSocket,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -75,13 +143,13 @@ namespace ix
ss << verb << " " << path << " HTTP/1.1\r\n"; ss << verb << " " << path << " HTTP/1.1\r\n";
ss << "Host: " << host << "\r\n"; ss << "Host: " << host << "\r\n";
if (args.compress) if (args->compress)
{ {
ss << "Accept-Encoding: gzip" << "\r\n"; ss << "Accept-Encoding: gzip" << "\r\n";
} }
// Append extra headers // Append extra headers
for (auto&& it : args.extraHeaders) for (auto&& it : args->extraHeaders)
{ {
ss << it.first << ": " << it.second << "\r\n"; ss << it.first << ": " << it.second << "\r\n";
} }
@ -103,7 +171,7 @@ namespace ix
ss << "Content-Length: " << body.size() << "\r\n"; ss << "Content-Length: " << body.size() << "\r\n";
// Set default Content-Type if unspecified // Set default Content-Type if unspecified
if (args.extraHeaders.find("Content-Type") == args.extraHeaders.end()) if (args->extraHeaders.find("Content-Type") == args->extraHeaders.end())
{ {
ss << "Content-Type: application/x-www-form-urlencoded" << "\r\n"; ss << "Content-Type: application/x-www-form-urlencoded" << "\r\n";
} }
@ -121,23 +189,23 @@ namespace ix
// Make a cancellation object dealing with connection timeout // Make a cancellation object dealing with connection timeout
auto isCancellationRequested = auto isCancellationRequested =
makeCancellationRequestWithTimeout(args.connectTimeout, requestInitCancellation); makeCancellationRequestWithTimeout(args->connectTimeout, requestInitCancellation);
bool success = _socket->connect(host, port, errMsg, isCancellationRequested); bool success = _socket->connect(host, port, errMsg, isCancellationRequested);
if (!success) if (!success)
{ {
std::stringstream ss; std::stringstream ss;
ss << "Cannot connect to url: " << url << " / error : " << errMsg; ss << "Cannot connect to url: " << url << " / error : " << errMsg;
return HttpResponse(code, HttpErrorCode::CannotConnect, return std::make_shared<HttpResponse>(code, HttpErrorCode::CannotConnect,
headers, payload, ss.str(), headers, payload, ss.str(),
uploadSize, downloadSize); uploadSize, downloadSize);
} }
// Make a new cancellation object dealing with transfer timeout // Make a new cancellation object dealing with transfer timeout
isCancellationRequested = isCancellationRequested =
makeCancellationRequestWithTimeout(args.transferTimeout, requestInitCancellation); makeCancellationRequestWithTimeout(args->transferTimeout, requestInitCancellation);
if (args.verbose) if (args->verbose)
{ {
std::stringstream ss; std::stringstream ss;
ss << "Sending " << verb << " request " ss << "Sending " << verb << " request "
@ -154,7 +222,7 @@ namespace ix
if (!_socket->writeBytes(req, isCancellationRequested)) if (!_socket->writeBytes(req, isCancellationRequested))
{ {
std::string errorMsg("Cannot send request"); std::string errorMsg("Cannot send request");
return HttpResponse(code, HttpErrorCode::SendError, return std::make_shared<HttpResponse>(code, HttpErrorCode::SendError,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -168,12 +236,12 @@ namespace ix
if (!lineValid) if (!lineValid)
{ {
std::string errorMsg("Cannot retrieve status line"); std::string errorMsg("Cannot retrieve status line");
return HttpResponse(code, HttpErrorCode::CannotReadStatusLine, return std::make_shared<HttpResponse>(code, HttpErrorCode::CannotReadStatusLine,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
if (args.verbose) if (args->verbose)
{ {
std::stringstream ss; std::stringstream ss;
ss << "Status line " << line; ss << "Status line " << line;
@ -183,7 +251,7 @@ namespace ix
if (sscanf(line.c_str(), "HTTP/1.1 %d", &code) != 1) if (sscanf(line.c_str(), "HTTP/1.1 %d", &code) != 1)
{ {
std::string errorMsg("Cannot parse response code from status line"); std::string errorMsg("Cannot parse response code from status line");
return HttpResponse(code, HttpErrorCode::MissingStatus, return std::make_shared<HttpResponse>(code, HttpErrorCode::MissingStatus,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -195,27 +263,27 @@ namespace ix
if (!headersValid) if (!headersValid)
{ {
std::string errorMsg("Cannot parse http headers"); std::string errorMsg("Cannot parse http headers");
return HttpResponse(code, HttpErrorCode::HeaderParsingError, return std::make_shared<HttpResponse>(code, HttpErrorCode::HeaderParsingError,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
// Redirect ? // Redirect ?
if ((code >= 301 && code <= 308) && args.followRedirects) if ((code >= 301 && code <= 308) && args->followRedirects)
{ {
if (headers.find("Location") == headers.end()) if (headers.find("Location") == headers.end())
{ {
std::string errorMsg("Missing location header for redirect"); std::string errorMsg("Missing location header for redirect");
return HttpResponse(code, HttpErrorCode::MissingLocation, return std::make_shared<HttpResponse>(code, HttpErrorCode::MissingLocation,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
if (redirects >= args.maxRedirects) if (redirects >= args->maxRedirects)
{ {
std::stringstream ss; std::stringstream ss;
ss << "Too many redirects: " << redirects; ss << "Too many redirects: " << redirects;
return HttpResponse(code, HttpErrorCode::TooManyRedirects, return std::make_shared<HttpResponse>(code, HttpErrorCode::TooManyRedirects,
headers, payload, ss.str(), headers, payload, ss.str(),
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -227,7 +295,7 @@ namespace ix
if (verb == "HEAD") if (verb == "HEAD")
{ {
return HttpResponse(code, HttpErrorCode::Ok, return std::make_shared<HttpResponse>(code, HttpErrorCode::Ok,
headers, payload, std::string(), headers, payload, std::string(),
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -243,12 +311,12 @@ namespace ix
payload.reserve(contentLength); payload.reserve(contentLength);
auto chunkResult = _socket->readBytes(contentLength, auto chunkResult = _socket->readBytes(contentLength,
args.onProgressCallback, args->onProgressCallback,
isCancellationRequested); isCancellationRequested);
if (!chunkResult.first) if (!chunkResult.first)
{ {
errorMsg = "Cannot read chunk"; errorMsg = "Cannot read chunk";
return HttpResponse(code, HttpErrorCode::ChunkReadError, return std::make_shared<HttpResponse>(code, HttpErrorCode::ChunkReadError,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -266,7 +334,7 @@ namespace ix
if (!lineResult.first) if (!lineResult.first)
{ {
return HttpResponse(code, HttpErrorCode::ChunkReadError, return std::make_shared<HttpResponse>(code, HttpErrorCode::ChunkReadError,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -276,7 +344,7 @@ namespace ix
ss << std::hex << line; ss << std::hex << line;
ss >> chunkSize; ss >> chunkSize;
if (args.verbose) if (args->verbose)
{ {
std::stringstream oss; std::stringstream oss;
oss << "Reading " << chunkSize << " bytes" oss << "Reading " << chunkSize << " bytes"
@ -288,12 +356,12 @@ namespace ix
// Read a chunk // Read a chunk
auto chunkResult = _socket->readBytes((size_t) chunkSize, auto chunkResult = _socket->readBytes((size_t) chunkSize,
args.onProgressCallback, args->onProgressCallback,
isCancellationRequested); isCancellationRequested);
if (!chunkResult.first) if (!chunkResult.first)
{ {
errorMsg = "Cannot read chunk"; errorMsg = "Cannot read chunk";
return HttpResponse(code, HttpErrorCode::ChunkReadError, return std::make_shared<HttpResponse>(code, HttpErrorCode::ChunkReadError,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -304,7 +372,7 @@ namespace ix
if (!lineResult.first) if (!lineResult.first)
{ {
return HttpResponse(code, HttpErrorCode::ChunkReadError, return std::make_shared<HttpResponse>(code, HttpErrorCode::ChunkReadError,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -319,7 +387,7 @@ namespace ix
else else
{ {
std::string errorMsg("Cannot read http body"); std::string errorMsg("Cannot read http body");
return HttpResponse(code, HttpErrorCode::CannotReadBody, return std::make_shared<HttpResponse>(code, HttpErrorCode::CannotReadBody,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
@ -333,60 +401,60 @@ namespace ix
if (!gzipInflate(payload, decompressedPayload)) if (!gzipInflate(payload, decompressedPayload))
{ {
std::string errorMsg("Error decompressing payload"); std::string errorMsg("Error decompressing payload");
return HttpResponse(code, HttpErrorCode::Gzip, return std::make_shared<HttpResponse>(code, HttpErrorCode::Gzip,
headers, payload, errorMsg, headers, payload, errorMsg,
uploadSize, downloadSize); uploadSize, downloadSize);
} }
payload = decompressedPayload; payload = decompressedPayload;
} }
return HttpResponse(code, HttpErrorCode::Ok, return std::make_shared<HttpResponse>(code, HttpErrorCode::Ok,
headers, payload, std::string(), headers, payload, std::string(),
uploadSize, downloadSize); uploadSize, downloadSize);
} }
HttpResponse HttpClient::get(const std::string& url, HttpResponsePtr HttpClient::get(const std::string& url,
const HttpRequestArgs& args) HttpRequestArgsPtr args)
{ {
return request(url, kGet, std::string(), args); return request(url, kGet, std::string(), args);
} }
HttpResponse HttpClient::head(const std::string& url, HttpResponsePtr HttpClient::head(const std::string& url,
const HttpRequestArgs& args) HttpRequestArgsPtr args)
{ {
return request(url, kHead, std::string(), args); return request(url, kHead, std::string(), args);
} }
HttpResponse HttpClient::del(const std::string& url, HttpResponsePtr HttpClient::del(const std::string& url,
const HttpRequestArgs& args) HttpRequestArgsPtr args)
{ {
return request(url, kDel, std::string(), args); return request(url, kDel, std::string(), args);
} }
HttpResponse HttpClient::post(const std::string& url, HttpResponsePtr HttpClient::post(const std::string& url,
const HttpParameters& httpParameters, const HttpParameters& httpParameters,
const HttpRequestArgs& args) HttpRequestArgsPtr args)
{ {
return request(url, kPost, serializeHttpParameters(httpParameters), args); return request(url, kPost, serializeHttpParameters(httpParameters), args);
} }
HttpResponse HttpClient::post(const std::string& url, HttpResponsePtr HttpClient::post(const std::string& url,
const std::string& body, const std::string& body,
const HttpRequestArgs& args) HttpRequestArgsPtr args)
{ {
return request(url, kPost, body, args); return request(url, kPost, body, args);
} }
HttpResponse HttpClient::put(const std::string& url, HttpResponsePtr HttpClient::put(const std::string& url,
const HttpParameters& httpParameters, const HttpParameters& httpParameters,
const HttpRequestArgs& args) HttpRequestArgsPtr args)
{ {
return request(url, kPut, serializeHttpParameters(httpParameters), args); return request(url, kPut, serializeHttpParameters(httpParameters), args);
} }
HttpResponse HttpClient::put(const std::string& url, HttpResponsePtr HttpClient::put(const std::string& url,
const std::string& body, const std::string& body,
const HttpRequestArgs& args) const HttpRequestArgsPtr args)
{ {
return request(url, kPut, body, args); return request(url, kPut, body, args);
} }
@ -488,11 +556,11 @@ namespace ix
} }
void HttpClient::log(const std::string& msg, void HttpClient::log(const std::string& msg,
const HttpRequestArgs& args) HttpRequestArgsPtr args)
{ {
if (args.logger) if (args->logger)
{ {
args.logger(msg); args->logger(msg);
} }
} }
} }

72
ixwebsocket/IXHttpClient.h
View File

@ -10,11 +10,13 @@
#include "IXWebSocketHttpHeaders.h" #include "IXWebSocketHttpHeaders.h"
#include <algorithm> #include <algorithm>
#include <atomic> #include <atomic>
#include <condition_variable>
#include <functional> #include <functional>
#include <map> #include <map>
#include <memory> #include <memory>
#include <mutex> #include <mutex>
#include <tuple> #include <queue>
#include <thread>
namespace ix namespace ix
{ {
@ -34,7 +36,8 @@ namespace ix
MissingLocation = 11, MissingLocation = 11,
TooManyRedirects = 12, TooManyRedirects = 12,
ChunkReadError = 13, ChunkReadError = 13,
CannotReadBody = 14 CannotReadBody = 14,
Invalid = 100
}; };
struct HttpResponse struct HttpResponse
@ -66,12 +69,15 @@ namespace ix
} }
}; };
using HttpResponsePtr = std::shared_ptr<HttpResponse>;
using HttpParameters = std::map<std::string, std::string>; using HttpParameters = std::map<std::string, std::string>;
using Logger = std::function<void(const std::string&)>; using Logger = std::function<void(const std::string&)>;
using OnResponseCallback = std::function<void(const HttpResponsePtr&)>;
struct HttpRequestArgs struct HttpRequestArgs
{ {
std::string url; std::string url;
std::string verb;
WebSocketHttpHeaders extraHeaders; WebSocketHttpHeaders extraHeaders;
std::string body; std::string body;
int connectTimeout; int connectTimeout;
@ -84,51 +90,71 @@ namespace ix
OnProgressCallback onProgressCallback; OnProgressCallback onProgressCallback;
}; };
using HttpRequestArgsPtr = std::shared_ptr<HttpRequestArgs>;
class HttpClient class HttpClient
{ {
public: public:
HttpClient(); HttpClient(bool async = false);
~HttpClient(); ~HttpClient();
HttpResponse get(const std::string& url, const HttpRequestArgs& args); HttpResponsePtr get(const std::string& url, HttpRequestArgsPtr args);
HttpResponse head(const std::string& url, const HttpRequestArgs& args); HttpResponsePtr head(const std::string& url, HttpRequestArgsPtr args);
HttpResponse del(const std::string& url, const HttpRequestArgs& args); HttpResponsePtr del(const std::string& url, HttpRequestArgsPtr args);
HttpResponse post(const std::string& url, HttpResponsePtr post(const std::string& url,
const HttpParameters& httpParameters, const HttpParameters& httpParameters,
const HttpRequestArgs& args); HttpRequestArgsPtr args);
HttpResponse post(const std::string& url, HttpResponsePtr post(const std::string& url,
const std::string& body, const std::string& body,
const HttpRequestArgs& args); HttpRequestArgsPtr args);
HttpResponse put(const std::string& url, HttpResponsePtr put(const std::string& url,
const HttpParameters& httpParameters, const HttpParameters& httpParameters,
const HttpRequestArgs& args); HttpRequestArgsPtr args);
HttpResponse put(const std::string& url, HttpResponsePtr put(const std::string& url,
const std::string& body, const std::string& body,
const HttpRequestArgs& args); HttpRequestArgsPtr args);
HttpResponse request(const std::string& url, HttpResponsePtr request(const std::string& url,
const std::string& verb, const std::string& verb,
const std::string& body, const std::string& body,
const HttpRequestArgs& args, HttpRequestArgsPtr args,
int redirects = 0); int redirects = 0);
// Async API
HttpRequestArgsPtr createRequest(const std::string& url = std::string(),
const std::string& verb = HttpClient::kGet);
bool performRequest(HttpRequestArgsPtr request,
const OnResponseCallback& onResponseCallback);
std::string serializeHttpParameters(const HttpParameters& httpParameters); std::string serializeHttpParameters(const HttpParameters& httpParameters);
std::string urlEncode(const std::string& value); std::string urlEncode(const std::string& value);
private:
void log(const std::string& msg, const HttpRequestArgs& args);
bool gzipInflate(const std::string& in, std::string& out);
std::shared_ptr<Socket> _socket;
const static std::string kPost; const static std::string kPost;
const static std::string kGet; const static std::string kGet;
const static std::string kHead; const static std::string kHead;
const static std::string kDel; const static std::string kDel;
const static std::string kPut; const static std::string kPut;
private:
void log(const std::string& msg, HttpRequestArgsPtr args);
bool gzipInflate(const std::string& in, std::string& out);
// Async API background thread runner
void run();
// Async API
bool _async;
std::queue<std::pair<HttpRequestArgsPtr, OnResponseCallback>> _queue;
mutable std::mutex _queueMutex;
std::condition_variable _condition;
std::atomic<bool> _stop;
std::thread _thread;
std::shared_ptr<Socket> _socket;
}; };
} // namespace ix } // namespace ix

218
test/IXHttpClientTest.cpp
View File

@ -15,85 +15,219 @@ TEST_CASE("http client", "[http]")
{ {
SECTION("Connect to a remote HTTP server") SECTION("Connect to a remote HTTP server")
{ {
std::string url("http://httpbin.org/"); HttpClient httpClient;
WebSocketHttpHeaders headers; WebSocketHttpHeaders headers;
headers["User-Agent"] = "ixwebsocket";
HttpRequestArgs args; std::string url("http://httpbin.org/");
args.extraHeaders = headers; auto args = httpClient.createRequest(url);
args.connectTimeout = 60;
args.transferTimeout = 60; args->extraHeaders = headers;
args.followRedirects = true; args->connectTimeout = 60;
args.maxRedirects = 10; args->transferTimeout = 60;
args.verbose = true; args->followRedirects = true;
args.compress = true; args->maxRedirects = 10;
args.logger = [](const std::string& msg) args->verbose = true;
args->compress = true;
args->logger = [](const std::string& msg)
{ {
std::cout << msg; std::cout << msg;
}; };
args.onProgressCallback = [](int current, int total) -> bool args->onProgressCallback = [](int current, int total) -> bool
{ {
std::cerr << "\r" << "Downloaded " std::cerr << "\r" << "Downloaded "
<< current << " bytes out of " << total; << current << " bytes out of " << total;
return true; return true;
}; };
HttpClient httpClient; auto response = httpClient.get(url, args);
HttpResponse response = httpClient.get(url, args);
for (auto it : response.headers) for (auto it : response->headers)
{ {
std::cerr << it.first << ": " << it.second << std::endl; std::cerr << it.first << ": " << it.second << std::endl;
} }
std::cerr << "Upload size: " << response.uploadSize << std::endl; std::cerr << "Upload size: " << response->uploadSize << std::endl;
std::cerr << "Download size: " << response.downloadSize << std::endl; std::cerr << "Download size: " << response->downloadSize << std::endl;
std::cerr << "Status: " << response.statusCode << std::endl; std::cerr << "Status: " << response->statusCode << std::endl;
std::cerr << "Error message: " << response->errorMsg << std::endl;
REQUIRE(response.errorCode == HttpErrorCode::Ok); REQUIRE(response->errorCode == HttpErrorCode::Ok);
REQUIRE(response->statusCode == 200);
} }
#if defined(IXWEBSOCKET_USE_TLS)
SECTION("Connect to a remote HTTPS server") SECTION("Connect to a remote HTTPS server")
{ {
std::string url("https://httpbin.org/"); HttpClient httpClient;
WebSocketHttpHeaders headers; WebSocketHttpHeaders headers;
headers["User-Agent"] = "ixwebsocket";
HttpRequestArgs args; std::string url("https://httpbin.org/");
args.extraHeaders = headers; auto args = httpClient.createRequest(url);
args.connectTimeout = 60;
args.transferTimeout = 60; args->extraHeaders = headers;
args.followRedirects = true; args->connectTimeout = 60;
args.maxRedirects = 10; args->transferTimeout = 60;
args.verbose = true; args->followRedirects = true;
args.compress = true; args->maxRedirects = 10;
args.logger = [](const std::string& msg) args->verbose = true;
args->compress = true;
args->logger = [](const std::string& msg)
{ {
std::cout << msg; std::cout << msg;
}; };
args.onProgressCallback = [](int current, int total) -> bool args->onProgressCallback = [](int current, int total) -> bool
{ {
std::cerr << "\r" << "Downloaded " std::cerr << "\r" << "Downloaded "
<< current << " bytes out of " << total; << current << " bytes out of " << total;
return true; return true;
}; };
HttpClient httpClient; auto response = httpClient.get(url, args);
HttpResponse response = httpClient.get(url, args);
for (auto it : response.headers) for (auto it : response->headers)
{ {
std::cerr << it.first << ": " << it.second << std::endl; std::cerr << it.first << ": " << it.second << std::endl;
} }
std::cerr << "Upload size: " << response.uploadSize << std::endl; std::cerr << "Upload size: " << response->uploadSize << std::endl;
std::cerr << "Download size: " << response.downloadSize << std::endl; std::cerr << "Download size: " << response->downloadSize << std::endl;
std::cerr << "Status: " << response.statusCode << std::endl; std::cerr << "Status: " << response->statusCode << std::endl;
std::cerr << "Error message: " << response->errorMsg << std::endl;
REQUIRE(response.errorCode == HttpErrorCode::Ok); REQUIRE(response->errorCode == HttpErrorCode::Ok);
REQUIRE(response->statusCode == 200);
}
SECTION("Async API, one call")
{
bool async = true;
HttpClient httpClient(async);
WebSocketHttpHeaders headers;
std::string url("https://httpbin.org/");
auto args = httpClient.createRequest(url);
args->extraHeaders = headers;
args->connectTimeout = 60;
args->transferTimeout = 60;
args->followRedirects = true;
args->maxRedirects = 10;
args->verbose = true;
args->compress = true;
args->logger = [](const std::string& msg)
{
std::cout << msg;
};
args->onProgressCallback = [](int current, int total) -> bool
{
std::cerr << "\r" << "Downloaded "
<< current << " bytes out of " << total;
return true;
};
std::atomic<bool> requestCompleted(false);
std::atomic<int> statusCode(0);
httpClient.performRequest(args, [&requestCompleted, &statusCode]
(const HttpResponsePtr& response)
{
std::cerr << "Upload size: " << response->uploadSize << std::endl;
std::cerr << "Download size: " << response->downloadSize << std::endl;
std::cerr << "Status: " << response->statusCode << std::endl;
std::cerr << "Error message: " << response->errorMsg << std::endl;
// In case of failure, print response->errorMsg
statusCode = response->statusCode;
requestCompleted = true;
}
);
int wait = 0;
while (wait < 5000)
{
if (requestCompleted) break;
std::chrono::duration<double, std::milli> duration(10);
std::this_thread::sleep_for(duration);
wait += 10;
}
std::cerr << "Done" << std::endl;
REQUIRE(statusCode == 200);
}
SECTION("Async API, multiple calls")
{
bool async = true;
HttpClient httpClient(async);
WebSocketHttpHeaders headers;
std::string url("http://httpbin.org/");
auto args = httpClient.createRequest(url);
args->extraHeaders = headers;
args->connectTimeout = 60;
args->transferTimeout = 60;
args->followRedirects = true;
args->maxRedirects = 10;
args->verbose = true;
args->compress = true;
args->logger = [](const std::string& msg)
{
std::cout << msg;
};
args->onProgressCallback = [](int current, int total) -> bool
{
std::cerr << "\r" << "Downloaded "
<< current << " bytes out of " << total;
return true;
};
std::atomic<bool> requestCompleted(false);
std::atomic<int> statusCode0(0);
std::atomic<int> statusCode1(0);
std::atomic<int> statusCode2(0);
for (int i = 0; i < 3; ++i)
{
httpClient.performRequest(args, [i, &requestCompleted, &statusCode0, &statusCode1, &statusCode2]
(const HttpResponsePtr& response)
{
std::cerr << "Upload size: " << response->uploadSize << std::endl;
std::cerr << "Download size: " << response->downloadSize << std::endl;
std::cerr << "Status: " << response->statusCode << std::endl;
std::cerr << "Error message: " << response->errorMsg << std::endl;
// In case of failure, print response->errorMsg
if (i == 0)
{
statusCode0 = response->statusCode;
}
else if (i == 1)
{
statusCode1 = response->statusCode;
}
else if (i == 2)
{
statusCode2 = response->statusCode;
requestCompleted = true;
}
}
);
}
int wait = 0;
while (wait < 10000)
{
if (requestCompleted) break;
std::chrono::duration<double, std::milli> duration(10);
std::this_thread::sleep_for(duration);
wait += 10;
}
std::cerr << "Done" << std::endl;
REQUIRE(statusCode0 == 200);
REQUIRE(statusCode1 == 200);
REQUIRE(statusCode2 == 200);
} }
#endif
} }

34
ws/IXSentryClient.cpp
View File

@ -141,42 +141,42 @@ namespace ix
bool SentryClient::send(const Json::Value& msg, bool SentryClient::send(const Json::Value& msg,
bool verbose) bool verbose)
{ {
HttpRequestArgs args; auto args = _httpClient.createRequest();
args.extraHeaders["X-Sentry-Auth"] = SentryClient::computeAuthHeader(); args->extraHeaders["X-Sentry-Auth"] = SentryClient::computeAuthHeader();
args.connectTimeout = 60; args->connectTimeout = 60;
args.transferTimeout = 5 * 60; args->transferTimeout = 5 * 60;
args.followRedirects = true; args->followRedirects = true;
args.verbose = verbose; args->verbose = verbose;
args.logger = [](const std::string& msg) args->logger = [](const std::string& msg)
{ {
std::cout << msg; std::cout << msg;
}; };
std::string body = computePayload(msg); std::string body = computePayload(msg);
HttpResponse response = _httpClient.post(_url, body, args); HttpResponsePtr response = _httpClient.post(_url, body, args);
if (verbose) if (verbose)
{ {
for (auto it : response.headers) for (auto it : response->headers)
{ {
std::cerr << it.first << ": " << it.second << std::endl; std::cerr << it.first << ": " << it.second << std::endl;
} }
std::cerr << "Upload size: " << response.uploadSize << std::endl; std::cerr << "Upload size: " << response->uploadSize << std::endl;
std::cerr << "Download size: " << response.downloadSize << std::endl; std::cerr << "Download size: " << response->downloadSize << std::endl;
std::cerr << "Status: " << response.statusCode << std::endl; std::cerr << "Status: " << response->statusCode << std::endl;
if (response.errorCode != HttpErrorCode::Ok) if (response->errorCode != HttpErrorCode::Ok)
{ {
std::cerr << "error message: " << response.errorMsg << std::endl; std::cerr << "error message: " << response->errorMsg << std::endl;
} }
if (response.headers["Content-Type"] != "application/octet-stream") if (response->headers["Content-Type"] != "application/octet-stream")
{ {
std::cerr << "payload: " << response.payload << std::endl; std::cerr << "payload: " << response->payload << std::endl;
} }
} }
return response.statusCode == 200; return response->statusCode == 200;
} }
} // namespace ix } // namespace ix

2
ws/ixcobra/IXCobraMetricsThreadedPublisher.cpp
View File

@ -99,7 +99,7 @@ namespace ix
void CobraMetricsThreadedPublisher::pushMessage(MessageKind messageKind, void CobraMetricsThreadedPublisher::pushMessage(MessageKind messageKind,
const Json::Value& msg) const Json::Value& msg)
{ {
// Now actually enqueue the task // Enqueue the task
{ {
// acquire lock // acquire lock
std::unique_lock<std::mutex> lock(_queue_mutex); std::unique_lock<std::mutex> lock(_queue_mutex);

1
ws/ixcobra/IXCobraMetricsThreadedPublisher.h
View File

@ -25,7 +25,6 @@ namespace ix
~CobraMetricsThreadedPublisher(); ~CobraMetricsThreadedPublisher();
/// Configuration / set keys, etc... /// Configuration / set keys, etc...
/// All input data but the channel name is encrypted with rc4
void configure(const std::string& appkey, void configure(const std::string& appkey,
const std::string& endpoint, const std::string& endpoint,
const std::string& channel, const std::string& channel,

8
ws/ws_cobra_to_sentry.cpp
View File

@ -46,9 +46,11 @@ namespace ix
std::condition_variable progressCondition; std::condition_variable progressCondition;
std::queue<Json::Value> queue; std::queue<Json::Value> queue;
SentryClient sentryClient(dsn);
auto sentrySender = [&condition, &progressCondition, &conditionVariableMutex, auto sentrySender = [&condition, &progressCondition, &conditionVariableMutex,
&queue, verbose, &errorSending, &sentCount, &queue, verbose, &errorSending, &sentCount,
&stop, &dsn] &stop, &sentryClient]
{ {
while (true) while (true)
{ {
@ -62,9 +64,7 @@ namespace ix
queue.pop(); queue.pop();
} }
SentryClient sc(dsn); if (!sentryClient.send(msg, verbose))
if (!sc.send(msg, verbose))
{ {
errorSending = true; errorSending = true;
} }

44
ws/ws_http_client.cpp
View File

@ -95,19 +95,20 @@ namespace ix
const std::string& output, const std::string& output,
bool compress) bool compress)
{ {
HttpRequestArgs args; HttpClient httpClient;
args.extraHeaders = parseHeaders(headersData); auto args = httpClient.createRequest();
args.connectTimeout = connectTimeout; args->extraHeaders = parseHeaders(headersData);
args.transferTimeout = transferTimeout; args->connectTimeout = connectTimeout;
args.followRedirects = followRedirects; args->transferTimeout = transferTimeout;
args.maxRedirects = maxRedirects; args->followRedirects = followRedirects;
args.verbose = verbose; args->maxRedirects = maxRedirects;
args.compress = compress; args->verbose = verbose;
args.logger = [](const std::string& msg) args->compress = compress;
args->logger = [](const std::string& msg)
{ {
std::cout << msg; std::cout << msg;
}; };
args.onProgressCallback = [](int current, int total) -> bool args->onProgressCallback = [](int current, int total) -> bool
{ {
std::cerr << "\r" << "Downloaded " std::cerr << "\r" << "Downloaded "
<< current << " bytes out of " << total; << current << " bytes out of " << total;
@ -116,8 +117,7 @@ namespace ix
HttpParameters httpParameters = parsePostParameters(data); HttpParameters httpParameters = parsePostParameters(data);
HttpClient httpClient; HttpResponsePtr response;
HttpResponse response;
if (headersOnly) if (headersOnly)
{ {
response = httpClient.head(url, args); response = httpClient.head(url, args);
@ -133,21 +133,21 @@ namespace ix
std::cerr << std::endl; std::cerr << std::endl;
for (auto it : response.headers) for (auto it : response->headers)
{ {
std::cerr << it.first << ": " << it.second << std::endl; std::cerr << it.first << ": " << it.second << std::endl;
} }
std::cerr << "Upload size: " << response.uploadSize << std::endl; std::cerr << "Upload size: " << response->uploadSize << std::endl;
std::cerr << "Download size: " << response.downloadSize << std::endl; std::cerr << "Download size: " << response->downloadSize << std::endl;
std::cerr << "Status: " << response.statusCode << std::endl; std::cerr << "Status: " << response->statusCode << std::endl;
if (response.errorCode != HttpErrorCode::Ok) if (response->errorCode != HttpErrorCode::Ok)
{ {
std::cerr << "error message: " << response.errorMsg << std::endl; std::cerr << "error message: " << response->errorMsg << std::endl;
} }
if (!headersOnly && response.errorCode == HttpErrorCode::Ok) if (!headersOnly && response->errorCode == HttpErrorCode::Ok)
{ {
if (save || !output.empty()) if (save || !output.empty())
{ {
@ -160,14 +160,14 @@ namespace ix
std::cout << "Writing to disk: " << filename << std::endl; std::cout << "Writing to disk: " << filename << std::endl;
std::ofstream out(filename); std::ofstream out(filename);
out.write((char*)&response.payload.front(), response.payload.size()); out.write((char*)&response->payload.front(), response->payload.size());
out.close(); out.close();
} }
else else
{ {
if (response.headers["Content-Type"] != "application/octet-stream") if (response->headers["Content-Type"] != "application/octet-stream")
{ {
std::cout << "payload: " << response.payload << std::endl; std::cout << "payload: " << response->payload << std::endl;
} }
else else
{ {