Skip to content

ipgallery.common.cpp / Microservice

Go to a project
Commit 75dc2fb7 by Amir Aharon
Options

Merge branch 'Feature-RestEvpp' into 'develop' 

Feature rest evpp client

See merge request !5
parents 0c193196 f94c6ff9
Expand all
Inline Side-by-side
Showing with 748 additions and 23 deletions
.vscode/launch.json
......@@ -51,6 +51,19 @@
"environment": []
},
{
"name": "C++ Client Local (GDB)",
"type": "cppdbg",
"request": "launch",
//"preLaunchTask": "start-gdbserver",
"miDebuggerPath": "/usr/bin/gdb",
"targetArchitecture": "x64",
"program": "${workspaceRoot}/build/bin/test_MicroserviceClient",
"args": [],
"stopAtEntry": false,
"cwd": "${workspaceRoot}",
"environment": []
},
{
"name": "C++ Attach (GDB)",
"type": "cppdbg",
"request": "launch",
......
.vscode/settings.json
......@@ -7,6 +7,51 @@
"tuple": "cpp",
"utility": "cpp",
"bitset": "cpp",
"chrono": "cpp"
"chrono": "cpp",
"thread": "cpp",
"cctype": "cpp",
"clocale": "cpp",
"cmath": "cpp",
"csignal": "cpp",
"cstdarg": "cpp",
"cstddef": "cpp",
"cstdio": "cpp",
"cstdlib": "cpp",
"cstring": "cpp",
"ctime": "cpp",
"cwchar": "cpp",
"cwctype": "cpp",
"atomic": "cpp",
"strstream": "cpp",
"codecvt": "cpp",
"condition_variable": "cpp",
"cstdint": "cpp",
"deque": "cpp",
"forward_list": "cpp",
"unordered_map": "cpp",
"unordered_set": "cpp",
"vector": "cpp",
"exception": "cpp",
"fstream": "cpp",
"future": "cpp",
"initializer_list": "cpp",
"iomanip": "cpp",
"iosfwd": "cpp",
"iostream": "cpp",
"limits": "cpp",
"memory": "cpp",
"mutex": "cpp",
"new": "cpp",
"ostream": "cpp",
"numeric": "cpp",
"ratio": "cpp",
"sstream": "cpp",
"stdexcept": "cpp",
"streambuf": "cpp",
"system_error": "cpp",
"cinttypes": "cpp",
"type_traits": "cpp",
"typeindex": "cpp",
"typeinfo": "cpp"
}
}
\ No newline at end of file
CMakeLists.txt
......@@ -3,17 +3,17 @@ project(Microservice)
# version stuff
set (Microservice_VERSION_MAJOR 1)
set (Microservice_VERSION_MINOR 5)
set (Microservice_VERSION_PATCH 0)
set (Microservice_VERSION_PATCH 1)
set(Microservice_VERSION_STRING ${Microservice_VERSION_MAJOR}.${Microservice_VERSION_MINOR}.${Microservice_VERSION_PATCH})
# type build flags
#set(CMAKE_BUILD_TYPE Release)
set(CMAKE_BINARY_DIR build)
set(BUILD_DIR build)
set(CMAKE_BUILD_TYPE Debug)
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11 -m64 -g")
set(CMAKE_CXX_FLAGS_DEBUG "-O0")
set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/lib)
set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin)
set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${BUILD_DIR}/lib)
set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${BUILD_DIR}/bin)
set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
# linked libs and their locations
......@@ -161,7 +161,7 @@ set(CPACK_PACKAGE_DEFAULT_LOCATION "./dist")
include(CPack)
#add_custom_target(dist COMMAND ${CMAKE_MAKE_PROGRAM} package_source)
# RUN make package
# gradle uploadArchives -Pcversion=[version] -Ppublish_file=[the gz file]
#gradle uploadArchives -Pcversion=[version] -Ppublish_file=[the gz file]
# install lib files
#
install(TARGETS Microservice DESTINATION lib)
......
README.md
## C++ Microservice Framework
* to create microservice docker run script/build_microservice_docker.sh [version]
## VERSIONS:
# 1.5.1
- Fully functioning evpp rest client , async only
# 1.5.0
- add new rest server Evpp
- add new Evpp http client, currently NOT TO BE USED!
......
compile_commands.json
......@@ -21,11 +21,21 @@
},
{
"directory": "/home/develop/project",
"command": "/usr/bin/c++ -DMicroservice_EXPORTS -std=c++11 -m64 -g -O0 -fPIC -I/home/develop/project/src -isystem /home/develop/project/../3party/cereal-1.2.1/include -isystem /home/develop/project/../3party/rapidjson-cereal-1.2.1 -isystem /home/develop/project/../3party/cppmetrics-0.1.1-Linux/include -isystem /home/develop/project/../3party/civetweb/include -isystem /home/develop/project/../3party/cpprest/Release/include -isystem /home/develop/project/../3party/rabbitmq -isystem /home/develop/project/../3party/flatbuffers/include -isystem /home/develop/project/../3party/poco-1.7.8/Foundation/include -isystem /home/develop/project/../3party/evpp/build-release/include -isystem /usr/include/Poco -isystem /usr/include/hiredis -o CMakeFiles/Microservice.dir/src/utils/EvppRequest.cpp.o -c /home/develop/project/src/utils/EvppRequest.cpp",
"file": "/home/develop/project/src/utils/EvppRequest.cpp"
},
{
"directory": "/home/develop/project",
"command": "/usr/bin/c++ -DMicroservice_EXPORTS -std=c++11 -m64 -g -O0 -fPIC -I/home/develop/project/src -isystem /home/develop/project/../3party/cereal-1.2.1/include -isystem /home/develop/project/../3party/rapidjson-cereal-1.2.1 -isystem /home/develop/project/../3party/cppmetrics-0.1.1-Linux/include -isystem /home/develop/project/../3party/civetweb/include -isystem /home/develop/project/../3party/cpprest/Release/include -isystem /home/develop/project/../3party/rabbitmq -isystem /home/develop/project/../3party/flatbuffers/include -isystem /home/develop/project/../3party/poco-1.7.8/Foundation/include -isystem /home/develop/project/../3party/evpp/build-release/include -isystem /usr/include/Poco -isystem /usr/include/hiredis -o CMakeFiles/Microservice.dir/src/utils/ScheduledTimer.cpp.o -c /home/develop/project/src/utils/ScheduledTimer.cpp",
"file": "/home/develop/project/src/utils/ScheduledTimer.cpp"
},
{
"directory": "/home/develop/project",
"command": "/usr/bin/c++ -DMicroservice_EXPORTS -std=c++11 -m64 -g -O0 -fPIC -I/home/develop/project/src -isystem /home/develop/project/../3party/cereal-1.2.1/include -isystem /home/develop/project/../3party/rapidjson-cereal-1.2.1 -isystem /home/develop/project/../3party/cppmetrics-0.1.1-Linux/include -isystem /home/develop/project/../3party/civetweb/include -isystem /home/develop/project/../3party/cpprest/Release/include -isystem /home/develop/project/../3party/rabbitmq -isystem /home/develop/project/../3party/flatbuffers/include -isystem /home/develop/project/../3party/poco-1.7.8/Foundation/include -isystem /home/develop/project/../3party/evpp/build-release/include -isystem /usr/include/Poco -isystem /usr/include/hiredis -o CMakeFiles/Microservice.dir/src/utils/EvppResponse.cpp.o -c /home/develop/project/src/utils/EvppResponse.cpp",
"file": "/home/develop/project/src/utils/EvppResponse.cpp"
},
{
"directory": "/home/develop/project",
"command": "/usr/bin/c++ -DMicroservice_EXPORTS -std=c++11 -m64 -g -O0 -fPIC -I/home/develop/project/src -isystem /home/develop/project/../3party/cereal-1.2.1/include -isystem /home/develop/project/../3party/rapidjson-cereal-1.2.1 -isystem /home/develop/project/../3party/cppmetrics-0.1.1-Linux/include -isystem /home/develop/project/../3party/civetweb/include -isystem /home/develop/project/../3party/cpprest/Release/include -isystem /home/develop/project/../3party/rabbitmq -isystem /home/develop/project/../3party/flatbuffers/include -isystem /home/develop/project/../3party/poco-1.7.8/Foundation/include -isystem /home/develop/project/../3party/evpp/build-release/include -isystem /usr/include/Poco -isystem /usr/include/hiredis -o CMakeFiles/Microservice.dir/src/utils/CommonUtils.cpp.o -c /home/develop/project/src/utils/CommonUtils.cpp",
"file": "/home/develop/project/src/utils/CommonUtils.cpp"
},
......
doc/todo.txt
- memory leak on performance testing of SendZmqRestRequests
- upon receiving the response , forward it to a new task to be carried by another thread
- catch http exception when callback url is down
- add Qihoo360/evpp as the http client with async, maybe also as server
\ No newline at end of file
+ add Qihoo360/evpp as the http client with async, maybe also as server
\ No newline at end of file
script/build_microservice_docker.sh
......@@ -7,6 +7,12 @@ if [ -z "$1" ]
exit
fi
echo "cleaning old stuff.."
rm CPack*
rm -rf cmake-build-debug _CPack_Packages CMakeFiles
rm CMakeCache.txt
rm cmake_install.cmake
echo "building the base image..."
docker build -f Dockerfile.develop -t municipalitybank.com:5050/ipgallery.common.cpp/microservice/develop .
echo "run build env"
......@@ -22,7 +28,7 @@ echo "build"
# set to exit on fail
set -e
docker exec -it devenv make
echo "packaging..."
echo "packaging... version: " $1
# make package
docker exec -it devenv make package
......
src/Microservice_Client.cpp
......@@ -162,4 +162,18 @@ ClientRespAsyncTask cMicroservice_Client::AsyncSend(ClientAsyncTaskParamsPtr &cl
});
}
void cMicroservice_Client::AsyncCreate(MSCommandParams* p_cmdParams,CrudHandler crudHandler){
p_commandClient_->AsyncCreate(p_cmdParams,crudHandler);
}
void cMicroservice_Client::AsyncRead(MSCommandParams* p_cmdParams,CrudHandler crudHandler){
p_commandClient_->AsyncRead(p_cmdParams,crudHandler);
}
void cMicroservice_Client::AsyncUpdate(MSCommandParams* p_cmdParams,CrudHandler crudHandler){
p_commandClient_->AsyncUpdate(p_cmdParams,crudHandler);
}
void cMicroservice_Client::AsyncDelete(MSCommandParams* p_cmdParams,CrudHandler crudHandler){
p_commandClient_->AsyncDelete(p_cmdParams,crudHandler);
}
src/Microservice_Client.h
......@@ -76,7 +76,6 @@ struct ClientAsyncTaskParams
typedef std::shared_ptr<ClientAsyncTaskParams> ClientAsyncTaskParamsPtr;
static const char *const NOT_MSGQ_CLIENT = "Not a MsgQueue Client";
static const char *const NOT_PUBSUB_CLIENT = "Not a PubSub Client";
......@@ -90,6 +89,7 @@ struct ClientAsyncTaskParamsFactory
static ClientAsyncTaskParamsPtr CreateMsgQAsyncTask(IResponse* p_IResponse, IContainer* p_IContainer) {
return std::make_shared<ClientAsyncTaskParams>(p_IResponse,p_IContainer,ClientAsyncTaskParams::eMsgQ);
}
};
class cMicroservice_Client {
......@@ -145,7 +145,11 @@ public:
ClientRespAsyncTask AsyncDelete(ClientAsyncTaskParamsPtr& clientAsyncTaskParamsPtr);
ClientRespAsyncTask AsyncSend(ClientAsyncTaskParamsPtr& clientAsyncTaskParamsPtr);
//
// async only clients
void AsyncCreate(MSCommandParams* p_cmdParams,CrudHandler crudHandler);
void AsyncRead(MSCommandParams* p_cmdParams,CrudHandler crudHandler);
void AsyncUpdate(MSCommandParams* p_cmdParams,CrudHandler crudHandler);
void AsyncDelete(MSCommandParams* p_cmdParams,CrudHandler crudHandler);
};
#endif /* MICROSERVICE_CLIENT_H */
......
src/common/Microservice_Iface.h
......@@ -31,6 +31,7 @@ class Microservice_PubSubContext;
typedef std::map<std::string, std::deque<std::string> > DequeStringMap;
typedef std::function<void(cMicroservice_BaseRestResponse* p_baseRestResponse)> CrudHandler;
namespace nsMicroservice_Iface
{
......@@ -116,6 +117,7 @@ namespace nsMicroservice_Iface
virtual cMicroservice_Enums::eLogLevel getLevel() = 0;
};
struct ICommandClient : public virtual IClient
{
......@@ -147,6 +149,22 @@ namespace nsMicroservice_Iface
}
};
typedef std::shared_ptr<HandleCommandData> HttpCommandDataPtr;
struct HandleAsyncCommandData
{
MSCommandParams* p_cmd_params;
CommandCounters* p_command_counters;
cMicroservice_Enums::eCrudMethod method;
CrudHandler crudHandler;
bool finished;
HandleAsyncCommandData(MSCommandParams* p_cmd_params,
CommandCounters* p_command_counters,
cMicroservice_Enums::eCrudMethod method,
CrudHandler crudHandler) :
p_cmd_params(p_cmd_params), p_command_counters(p_command_counters),method(method), crudHandler(crudHandler), finished(false) {
}
};
typedef std::shared_ptr<HandleAsyncCommandData> HandleAsyncCommandDataPtr;
ILogger* p_logger_;
static constexpr const char* TYPE = "Command";
......@@ -195,6 +213,14 @@ namespace nsMicroservice_Iface
virtual void SetLogger(ILogger* logger) { p_logger_ = logger; }
/**
* FULLY ASYNC COMMANDS
*
*/
virtual MSRetStat AsyncCreate(MSCommandParams* p_cmd_params,CrudHandler crudHandler) { return MSRetStat(); }
virtual MSRetStat AsyncRead(MSCommandParams* p_cmd_params,CrudHandler crudHandler) { return MSRetStat(); }
virtual MSRetStat AsyncUpdate(MSCommandParams* p_cmd_params,CrudHandler crudHandler) { return MSRetStat(); }
virtual MSRetStat AsyncDelete(MSCommandParams* p_cmd_params,CrudHandler crudHandler) { return MSRetStat(); }
};
struct IMetricsFactory
......
src/impl/clients/MSICommandClientEvppImpl.h
......@@ -3,7 +3,7 @@
#include "common/Microservice_Iface.h"
#include <evpp/event_loop_thread.h>
#include <evpp/httpc/response.h>
#include <utils/EvppResponse.h>
using namespace nsMicroservice_Iface;
......@@ -25,10 +25,16 @@ class MSICommandClientEvppImpl : public ICommandClient
* @param cmdDataPtr
*/
void HandleCommandAndCallback(ICommandClient::HttpCommandDataPtr &cmdDataPtr,cMicroservice_Enums::eCrudMethod crudMethod);
template <typename Req>
MSRetStat HandleAsyncCommandAndCallback(ICommandClient::HandleAsyncCommandDataPtr& handleAsyncCommandDataPtr);
void handleRequestFailed(const ICommandClient::HttpCommandDataPtr &cmdDataPtr,const char* error);
void handleRequestSuccess(const ICommandClient::HttpCommandDataPtr &cmdDataPtr,
const std::shared_ptr<evpp::httpc::Response> &responsePtr);
const std::shared_ptr<evpp::httpc::EvppResponse> &responsePtr);
void handleAsyncRequestFailed(const ICommandClient::HandleAsyncCommandDataPtr &asyncCmdDataPtr,const char* error);
void handleAsyncRequestSuccess(const ICommandClient::HandleAsyncCommandDataPtr &asyncCmdDataPtr,
const std::shared_ptr<evpp::httpc::EvppResponse> &responsePtr);
public:
MSICommandClientEvppImpl();
......@@ -39,6 +45,11 @@ class MSICommandClientEvppImpl : public ICommandClient
MSRetStat Update(MSCommandParams *p_cmd_params, cMicroservice_BaseRestResponse *p_response) override;
MSRetStat Delete(MSCommandParams *p_cmd_params, cMicroservice_BaseRestResponse *p_response) override;
MSRetStat AsyncCreate(MSCommandParams* p_cmd_params,CrudHandler crudHandler) override;
MSRetStat AsyncRead(MSCommandParams* p_cmd_params,CrudHandler crudHandler) override;
MSRetStat AsyncUpdate(MSCommandParams* p_cmd_params,CrudHandler crudHandler) override;
MSRetStat AsyncDelete(MSCommandParams* p_cmd_params,CrudHandler crudHandler) override;
};
#endif // MSICOMMANDCLIENTEVPPIMPL_H
\ No newline at end of file
src/impl/servers/Microservice_IRestServerEvppImpl.h
......@@ -8,8 +8,8 @@
#include <evpp/http/http_server.h>
struct EvppConnParams {
const evpp::http::ContextPtr& ctx_;
const evpp::http::HTTPSendResponseCallback& cb_;
const evpp::http::ContextPtr ctx_;
const evpp::http::HTTPSendResponseCallback cb_;
EvppConnParams(const evpp::http::ContextPtr& ctx, const evpp::http::HTTPSendResponseCallback& cb) :
ctx_(ctx),cb_(cb) {}
......
src/utils/EvppRequest.cpp 0 → 100644
#include <evpp/libevent.h>
#include <evpp/httpc/conn_pool.h>
#include "EvppResponse.h"
#include "EvppRequest.h"
#include <evpp/httpc/url_parser.h>
namespace evpp {
namespace httpc {
const std::string EvppRequest::empty_ = "";
EvppRequest::EvppRequest(ConnPool* pool, EventLoop* loop, evhttp_cmd_type req_type, const std::string& http_uri, const std::string& body)
: pool_(pool), loop_(loop), req_type_(req_type), host_(pool->host()), uri_(http_uri), body_(body) {
}
EvppRequest::EvppRequest(EventLoop* loop, evhttp_cmd_type req_type, const std::string& http_url, const std::string& body, Duration timeout)
: pool_(nullptr), loop_(loop), req_type_(req_type), body_(body) {
//TODO performance compare
#if LIBEVENT_VERSION_NUMBER >= 0x02001500
struct evhttp_uri* evuri = evhttp_uri_parse(http_url.c_str());
uri_ = evhttp_uri_get_path(evuri);
const char* query = evhttp_uri_get_query(evuri);
if (query && strlen(query) > 0) {
uri_ += "?";
uri_ += query;
}
host_ = evhttp_uri_get_host(evuri);
int port = evhttp_uri_get_port(evuri);
if (port < 0) {
port = 80;
}
conn_.reset(new Conn(loop, host_, port, timeout));
evhttp_uri_free(evuri);
#else
URLParser p(http_url);
conn_.reset(new Conn(loop, p.host, port, timeout));
if (p.query.empty()) {
uri_ = p.path;
} else {
uri_ = p.path + "?" + p.query;
}
host_ = p.host;
#endif
}
EvppRequest::~EvppRequest() {
assert(loop_->IsInLoopThread());
}
void EvppRequest::Execute(const EvppHandler& h) {
handler_ = h;
loop_->RunInLoop(std::bind(&EvppRequest::ExecuteInLoop, this));
}
void EvppRequest::ExecuteInLoop() {
//DLOG_TRACE;
assert(loop_->IsInLoopThread());
//evhttp_cmd_type req_type = EVHTTP_REQ_GET;
std::string errmsg;
struct evhttp_request* req = nullptr;
if (conn_) {
assert(pool_ == nullptr);
if (!conn_->Init()) {
errmsg = "conn init fail";
goto failed;
}
} else {
assert(pool_);
conn_ = pool_->Get(loop_);
if (!conn_->Init()) {
errmsg = "conn init fail";
goto failed;
}
}
req = evhttp_request_new(&EvppRequest::HandleResponse, this);
if (!req) {
errmsg = "evhttp_request_new fail";
goto failed;
}
if (evhttp_add_header(req->output_headers, "host", conn_->host().c_str())) {
evhttp_request_free(req);
errmsg = "evhttp_add_header failed";
goto failed;
}
/**
* add custom headers
*/
if (!headers_map_.empty()){
for (auto entry : headers_map_){
evhttp_add_header(req->output_headers, entry.first.c_str(), entry.second.c_str());
}
}
if (!body_.empty()) {
//req_type = EVHTTP_REQ_POST;
if (evbuffer_add(req->output_buffer, body_.c_str(), body_.size())) {
evhttp_request_free(req);
errmsg = "evbuffer_add fail";
goto failed;
}
}
if (evhttp_make_request(conn_->evhttp_conn(), req, req_type_, uri_.c_str()) != 0) {
// At here conn_ has owned this req, so don't need to free it.
errmsg = "evhttp_make_request fail";
goto failed;
}
return;
failed:
// Retry
if (retried_ < retry_number_) {
LOG_WARN << "this=" << this << " http request failed : " << errmsg << " retried=" << retried_ << " max retry_time=" << retry_number_ << ". Try again.";
Retry();
return;
}
std::shared_ptr<EvppResponse> response(new EvppResponse(this, nullptr));
handler_(response);
}
void EvppRequest::Retry() {
retried_ += 1;
if (retry_interval_.IsZero()) {
ExecuteInLoop();
} else {
loop_->RunAfter(retry_interval_.Milliseconds(), std::bind(&EvppRequest::ExecuteInLoop, this));
}
}
void EvppRequest::HandleResponse(struct evhttp_request* r, void* v) {
EvppRequest* thiz = (EvppRequest*)v;
assert(thiz);
thiz->HandleResponse(r);
}
void EvppRequest::HandleResponse(struct evhttp_request* r) {
assert(loop_->IsInLoopThread());
if (r) {
if (r->response_code == HTTP_OK || retried_ >= retry_number_) {
LOG_WARN << "this=" << this << " response_code=" << r->response_code << " retried=" << retried_ << " max retry_time=" << retry_number_;
std::shared_ptr<EvppResponse> response(new EvppResponse(this, r));
//Recycling the http Connection object
if (pool_) {
pool_->Put(conn_);
}
handler_(response);
return;
}
}
// Retry
if (retried_ < retry_number_) {
LOG_WARN << "this=" << this << " response_code=" << (r ? r->response_code : 0) << " retried=" << retried_ << " max retry_time=" << retry_number_ << ". Try again";
Retry();
return;
}
// Eventually this Request failed
std::shared_ptr<EvppResponse> response(new EvppResponse(this, r));
// Recycling the http Connection object
if (pool_) {
pool_->Put(conn_);
}
handler_(response);
}
} // httpc
} // evpp
src/utils/EvppRequest.h 0 → 100644
#pragma once
#include "evpp/libevent.h"
#include <evpp/inner_pre.h>
#include <evpp/event_loop.h>
#include <evpp/httpc/conn.h>
struct evhttp_connection;
namespace evpp {
namespace httpc {
class ConnPool;
class EvppResponse;
class Conn;
typedef std::function<void(const std::shared_ptr<EvppResponse>&)> EvppHandler;
class EVPP_EXPORT EvppRequest {
public:
// @brief Create a HTTP Request and create Conn from pool.
// Do a HTTP GET request if body is empty or HTTP POST request if body is not empty.
// @param[in] pool -
// @param[in] loop -
// @param[in] uri_with_param - The URI of the HTTP request with parameters
// @param[in] body -
// @return -
EvppRequest(ConnPool* pool, EventLoop* loop, evhttp_cmd_type req_type, const std::string& uri_with_param, const std::string& body);
// @brief Create a HTTP Request and create Conn myself
// Do a HTTP GET request if body is empty or HTTP POST request if body is not empty.
// @param[in] loop -
// @param[in] url - The URL of the HTTP request
// @param[in] body -
// @param[in] timeout -
// @return -
EvppRequest(EventLoop* loop, evhttp_cmd_type req_type,const std::string& url, const std::string& body, Duration timeout);
~EvppRequest();
void Execute(const EvppHandler& h);
const std::shared_ptr<Conn>& conn() const {
return conn_;
}
struct evhttp_connection* evhttp_conn() const {
return conn_->evhttp_conn();
}
const std::string& uri() const {
return uri_;
}
const std::string& host() const {
return host_;
}
void set_retry_number(int v) {
retry_number_ = v;
}
void set_retry_interval(Duration d) {
retry_interval_ = d;
}
void add_header(const std::string& header, const std::string& value){
headers_map_[header] = value;
}
void add_headers(std::map<std::string,std::string>& headers_map){
headers_map_.insert(headers_map.begin(), headers_map.end());
}
private:
static void HandleResponse(struct evhttp_request* r, void* v);
void HandleResponse(struct evhttp_request* r);
void ExecuteInLoop();
void Retry();
protected:
static const std::string empty_;
private:
ConnPool* pool_ = nullptr;
EventLoop* loop_;
std::string host_;
std::string uri_; // The URI of the HTTP request with parameters
std::string body_;
std::shared_ptr<Conn> conn_;
EvppHandler handler_;
evhttp_cmd_type req_type_;
std::map<std::string,std::string> headers_map_;
// The retried times
int retried_ = 0;
// The max retry times. Set to 0 if you don't want to retry when failed.
// The total execution times is retry_number_+1
int retry_number_ = 2;
// Default : 1ms
//Duration retry_interval_ = Duration(0.001);
Duration retry_interval_ = Duration(0);
};
typedef std::shared_ptr<EvppRequest> EvppRequestPtr;
class GetEvppRequest : public EvppRequest {
public:
GetEvppRequest(EventLoop* loop, const std::string& url, const std::string& body, Duration timeout)
: EvppRequest(loop, EVHTTP_REQ_GET,url, body, timeout) {}
GetEvppRequest(ConnPool* pool, EventLoop* loop, const std::string& uri_with_param)
: EvppRequest(pool, loop, EVHTTP_REQ_GET, uri_with_param, empty_) {}
GetEvppRequest(EventLoop* loop, const std::string& url, Duration timeout)
: EvppRequest(loop, EVHTTP_REQ_GET, url, empty_, timeout) {}
};
class PostEvppRequest : public EvppRequest {
public:
PostEvppRequest(ConnPool* pool, EventLoop* loop, const std::string& uri_with_param, const std::string& body)
: EvppRequest(pool, loop, EVHTTP_REQ_POST,uri_with_param, body) {}
PostEvppRequest(EventLoop* loop, const std::string& url, const std::string& body, Duration timeout)
: EvppRequest(loop, EVHTTP_REQ_POST,url, body, timeout) {}
PostEvppRequest(EventLoop* loop, const std::string& url, Duration timeout)
: EvppRequest(loop, EVHTTP_REQ_POST, url, empty_, timeout) {}
};
class PutEvppRequest : public EvppRequest {
public:
PutEvppRequest(ConnPool* pool, EventLoop* loop, const std::string& uri_with_param, const std::string& body)
: EvppRequest(pool, loop, EVHTTP_REQ_PUT,uri_with_param, body) {}
PutEvppRequest(EventLoop* loop, const std::string& url, const std::string& body, Duration timeout)
: EvppRequest(loop, EVHTTP_REQ_PUT,url, body, timeout) {}
PutEvppRequest(EventLoop* loop, const std::string& url, Duration timeout)
: EvppRequest(loop, EVHTTP_REQ_PUT, url, empty_, timeout) {}
};
class DeleteEvppRequest : public EvppRequest {
public:
DeleteEvppRequest(ConnPool* pool, EventLoop* loop, const std::string& uri_with_param, const std::string& body)
: EvppRequest(pool, loop, EVHTTP_REQ_DELETE, uri_with_param, body) {}
DeleteEvppRequest(EventLoop* loop, const std::string& url, const std::string& body, Duration timeout)
: EvppRequest(loop, EVHTTP_REQ_DELETE, url, body, timeout) {}
DeleteEvppRequest(EventLoop* loop, const std::string& url, Duration timeout)
: EvppRequest(loop, EVHTTP_REQ_DELETE, url, empty_, timeout) {}
};
} // httpc
} // evpp
src/utils/EvppResponse.cpp 0 → 100644
#include <evpp/libevent.h>
#include <evpp/httpc/conn.h>
#include <evpp/httpc/conn_pool.h>
#include "EvppResponse.h"
#include "EvppRequest.h"
namespace evpp {
namespace httpc {
EvppResponse::EvppResponse(EvppRequest* r, struct evhttp_request* evreq)
: request_(r), evreq_(evreq), http_code_(0) {
if (evreq) {
http_code_ = evreq->response_code;
#if LIBEVENT_VERSION_NUMBER >= 0x02001500
struct evbuffer* evbuf = evhttp_request_get_input_buffer(evreq);
size_t buffer_size = evbuffer_get_length(evbuf);
if (buffer_size > 0) {
this->body_ = evpp::Slice((char*)evbuffer_pullup(evbuf, -1), buffer_size);
}
#else
if (evreq->input_buffer->off > 0) {
this->body_ = evpp::Slice((char*)evreq->input_buffer->buffer, evreq->input_buffer->off);
}
#endif
}
}
EvppResponse::~EvppResponse() {
}
const char* EvppResponse::FindHeader(const char* key) {
if (http_code_ > 0) {
assert(this->evreq_);
return evhttp_find_header(this->evreq_->input_headers, key);
}
return nullptr;
}
}
}
src/utils/EvppResponse.h 0 → 100644
#pragma once
#include <map>
#include "evpp/inner_pre.h"
#include "evpp/event_loop.h"
#include "evpp/slice.h"
struct evhttp_request;
namespace evpp {
namespace httpc {
class EvppRequest;
class EVPP_EXPORT EvppResponse {
public:
typedef std::map<evpp::Slice, evpp::Slice> Headers;
EvppResponse(EvppRequest* r, struct evhttp_request* evreq);
~EvppResponse();
int http_code() const {
return http_code_;
}
const evpp::Slice& body() const {
return body_;
}
const EvppRequest* request() const {
return request_;
}
const char* FindHeader(const char* key);
private:
EvppRequest* request_;
struct evhttp_request* evreq_;
int http_code_;
evpp::Slice body_;
};
}
}
\ No newline at end of file
test/Microservice_ClientTest.cpp
......@@ -30,6 +30,8 @@
#include <evpp/httpc/conn.h>
#include <evpp/httpc/response.h>
#include <memory>
#include <utils/EvppRequest.h>
#include <utils/EvppResponse.h>
......@@ -165,7 +167,7 @@ void pubsubtest(cMicroservice_Client *p_Client) {
// for convenience
using json = nlohmann::json;
static const int ITERATIONS = 10000;
static const int ITERATIONS = 10;
static const char *const JSON_CONTENT = "{\n"
......@@ -438,7 +440,7 @@ void SendZmqRestRequests(const Microservice_App &msApp, cMicroservice_Client *p_
#include <atomic>
void runEvppClientTest(){
void runEvppTest(){
std::atomic_int counter(1);
bool responsed = false;
evpp::EventLoopThread t;
......@@ -463,9 +465,150 @@ void runEvppClientTest(){
t.Stop(true);
}
template <typename Req>
void runEvppClientTest(evpp::EventLoopThread& eventLoop,
const std::string url ,
std::atomic_int& testPassed){
std::atomic_int counter(1);
bool responsed = false;
evpp::EventLoopThread t;
for (int i = 0; i < ITERATIONS; i++)
{
std::shared_ptr<Req> r(new Req(eventLoop.loop(), url, evpp::Duration(3.0)));
auto f = [&counter, &responsed, r](const std::shared_ptr<evpp::httpc::EvppResponse> &response) {
std::cout << __PRETTY_FUNCTION__ << " http_code=" << response->http_code() << " [" << response->body().ToString() << "]\n";
//std::string header = response->FindHeader("Connection");
register int count = counter++;
if (count >= ITERATIONS)
responsed = true;
};
r->Execute(f);
}
while (!responsed) {
usleep(1);
}
testPassed++;
}
void runEvppRequestsTest(){
std::atomic_int testPassedNumber(0);
bool responsed = false;
evpp::EventLoopThread t;
t.Start(true);
//std::string uri = "https://postman-echo.com";
std::string uri = "http://172.16.1.244:8080";
std::string body = "This is expected to be sent back as part of response body.";
runEvppClientTest<evpp::httpc::GetEvppRequest>(t,uri + std::string("/get"),testPassedNumber);
runEvppClientTest<evpp::httpc::PostEvppRequest>(t,uri + std::string("/post"),testPassedNumber);
runEvppClientTest<evpp::httpc::PutEvppRequest>(t,uri + std::string("/put"),testPassedNumber);
runEvppClientTest<evpp::httpc::DeleteEvppRequest>(t,uri + std::string("/delete"),testPassedNumber);
while (testPassedNumber < 4) {
usleep(1);
}
t.Stop(true);
}
void printObjectDoc(rapidjson::Document &t_ObjectDoc){
std::ostringstream c_OutputStream;
if(!t_ObjectDoc.IsNull()) {
rapidjson::StringBuffer buffer;
rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
t_ObjectDoc.Accept(writer);
c_OutputStream << nsMicroservice_Constants::SUCCESS_REST_RESPONSE_TEMPLATE << buffer.GetString() << '}';
} else {
c_OutputStream << nsMicroservice_Constants::SUCCESS_NULL_REST_RESPONSE_TEMPLATE << '}';
}
std::cout << c_OutputStream.str().c_str() << std::endl;
}
void runMSClientEvppTest(){
std::string body = "This is expected to be sent back as part of response body.";
auto p_msClient = ClientFactory::createEvppCommandImpl("evpp-test");
std::atomic_int testPassedNumber(0);
std::map<std::string,std::string> headers;
headers["Content-Type"] = "application/json";
//std::string uri = "http://echo.jpillora.com";
std::string uri = "https://postman-echo.com";
MSCommandParams msCmndParams;
msCmndParams.WithEntity(uri)
.WithRequestParams("q=base")
.WithHeadersMap(&headers)
.WithContent(body);
auto f = [&testPassedNumber](cMicroservice_BaseRestResponse* p_baseRestResponse){
if(p_baseRestResponse->IsSuccess()){
std::cout << __PRETTY_FUNCTION__ << ": Success, Object =>" << std::endl;
printObjectDoc(p_baseRestResponse->GetObjectNode());
testPassedNumber++;
} else {
std::cout << __PRETTY_FUNCTION__ << ": Error" << std::endl;
}
};
for (int i = 0; i < ITERATIONS; i++)
{
msCmndParams.WithParamsString("post");
p_msClient->AsyncCreate(&msCmndParams, f);
msCmndParams.WithParamsString("get");
p_msClient->AsyncRead(&msCmndParams, f);
msCmndParams.WithParamsString("put");
p_msClient->AsyncUpdate(&msCmndParams, f);
msCmndParams.WithParamsString("delete");
p_msClient->AsyncDelete(&msCmndParams, f);
}
while (testPassedNumber < 4 * ITERATIONS)
{
usleep(1);
}
}
static int responsed = 0;
static void HandleHTTPResponse(const std::shared_ptr<evpp::httpc::EvppResponse>& response, evpp::httpc::PostEvppRequest* request) {
LOG_INFO << "http_code=" << response->http_code()
<< " URL=http://" << request->host() << request->uri()
<< " BODY = [" << response->body().ToString() << "]";
const char* header = response->FindHeader("Connection");
if (header) {
LOG_INFO << "HTTP HEADER Connection=" << header;
}
responsed++;
assert(request == response->request());
delete request; // The request MUST BE deleted in EventLoop thread.
}
void test_evpp(){
evpp::EventLoopThread t;
string url = "https://postman-echo.com/post";
t.Start(true);
evpp::httpc::PostEvppRequest* r = new evpp::httpc::PostEvppRequest(t.loop(), url, evpp::Duration(3.0));
r->Execute(std::bind(&HandleHTTPResponse, std::placeholders::_1, r));
r = new evpp::httpc::PostEvppRequest(t.loop(), url, evpp::Duration(3.0));
r->Execute(std::bind(&HandleHTTPResponse, std::placeholders::_1, r));
r = new evpp::httpc::PostEvppRequest(t.loop(), url, evpp::Duration(3.0));
r->Execute(std::bind(&HandleHTTPResponse, std::placeholders::_1, r));
r = new evpp::httpc::PostEvppRequest(t.loop(), url, evpp::Duration(3.0));
r->Execute(std::bind(&HandleHTTPResponse, std::placeholders::_1, r));
while (responsed != 4) {
usleep(1);
}
t.Stop(true);
LOG_INFO << "EventLoopThread stopped.";
}
int main(int argc, char *argv[])
{
auto duration = CommonUtils::measureFunc<>(runEvppClientTest);
//auto duration = CommonUtils::measureFunc<>(runEvppRequestsTest);
auto duration = CommonUtils::measureFunc<>(runMSClientEvppTest);
// auto duration = CommonUtils::measureFunc<>(test_evpp);
std::cout <<" Testing " << ITERATIONS << " with map serialization json took: " << duration << "msec" << '\n';
//runRestZmqTest();
......
test/Microservice_Test.cpp
......@@ -109,14 +109,47 @@ public:
rpj_Doc.AddMember(rapidjson::StringRef(it->first.c_str()),rapidjson::StringRef(dequeIt->c_str()),rpj_Alloc);
}
}
ReadSync(pc_reqCtx);
ReadAsyncEvpp(pc_reqCtx);
//ReadSync(pc_reqCtx);
// ReadAsync2(pc_reqCtx);
//this->WriteObjectToResponse(pc_reqCtx,rpj_Doc);
// this->WriteObjectToResponse(pc_reqCtx,rpj_Doc);
// add metric
long value = rand() % 1000 + 1;
p_histo->update(value);
}
void ReadAsyncEvpp(cMicroservice_RequestContext *pc_reqCtx) {
std::string body = "This is expected to be sent back as part of response body.";
//auto p_msClient = ClientFactory::createEvppCommandImpl("evpp-test");
std::map<std::string,std::string> headers;
headers["Content-Type"] = "application/json";
//std::string uri = "http://echo.jpillora.com";
std::string uri = "https://postman-echo.com";
auto clientAsyncTaskParamsPtr = ClientAsyncTaskParamsFactory::CreateCommndParamsAsyncTask(pc_reqCtx->mpti_Response,
pc_reqCtx->mpti_Container);
MSCommandParams msCmndParams;
msCmndParams.WithEntity(uri)
.WithRequestParams("q=base")
.WithHeadersMap(&headers)
.WithContent(body);
msCmndParams.WithParamsString("post");
p_rest_client_->AsyncCreate(&msCmndParams, [clientAsyncTaskParamsPtr](cMicroservice_BaseRestResponse* p_baseRestResponse){
if(p_baseRestResponse->IsSuccess()){
clientAsyncTaskParamsPtr->p_IContainer_->WriteObjectToResponse(
clientAsyncTaskParamsPtr->p_IResponse_.get(),
p_baseRestResponse->GetObjectNode());
} else {
clientAsyncTaskParamsPtr->p_IContainer_->SendErrorResp(
clientAsyncTaskParamsPtr->p_IResponse_.get(),
p_baseRestResponse->GetError());
}
});
}
void ReadSync(cMicroservice_RequestContext *pc_reqCtx) {
Microservice_RestResponse rest_response;
std::map<std::string,std::string> headers;
......@@ -380,8 +413,9 @@ int main(int argc, char *argv[])
.withMetrics()
.withMonitoring() // need to add reload
.withPubSub(NULL)
.withServiceDiscovery(NULL)
.addClient(ClientFactory::createEvppCommandImpl("other-service", "localhost", 32010, true, 10, false,"localhost:6379"))
.withServiceDiscovery(NULL) //auto p_msClient = ClientFactory::createEvppCommandImpl("evpp-test");
.addClient(ClientFactory::createEvppCommandImpl("other-service"))
//.addClient(ClientFactory::createEvppCommandImpl("other-service", "localhost", 32010, true, 10, false,"localhost:6379"))
// .addClient(ClientFactory::createHttpImplMsClient("other-service", "localhost", 32010, true, 10, false,"localhost:6379"))
.addClient(ClientFactory::createZmqMsgQImp("zmq-service", msApp.name(), 0,
Microservice_ZMQServerParams::eProtocol::eIpc))
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment