mirror of
https://github.com/qicosmos/rest_rpc.git
synced 2026-08-29 08:34:47 +08:00
remove files
This commit is contained in:
qicosmos
@@ -1,49 +0,0 @@
|
||||
#ifndef REST_RPC_CODEC_H_
|
||||
#define REST_RPC_CODEC_H_
|
||||
|
||||
#include <msgpack.hpp>
|
||||
|
||||
namespace rest_rpc {
|
||||
namespace rpc_service {
|
||||
|
||||
using buffer_type = msgpack::sbuffer;
|
||||
struct msgpack_codec {
|
||||
const static size_t init_size = 2 * 1024;
|
||||
|
||||
template<typename... Args>
|
||||
static buffer_type pack_args(Args&&... args) {
|
||||
buffer_type buffer(init_size);
|
||||
msgpack::pack(buffer, std::forward_as_tuple(std::forward<Args>(args)...));
|
||||
return buffer;
|
||||
}
|
||||
|
||||
template<typename Arg, typename... Args,
|
||||
typename = typename std::enable_if<std::is_enum<Arg>::value>::type>
|
||||
static std::string pack_args_str(Arg arg, Args&&... args) {
|
||||
buffer_type buffer(init_size);
|
||||
msgpack::pack(buffer, std::forward_as_tuple((int)arg, std::forward<Args>(args)...));
|
||||
return std::string(buffer.data(), buffer.size());
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
buffer_type pack(T&& t) const {
|
||||
buffer_type buffer;
|
||||
msgpack::pack(buffer, std::forward<T>(t));
|
||||
return buffer;
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
T unpack(char const* data, size_t length) {
|
||||
try {
|
||||
msgpack::unpack(&msg_, data, length);
|
||||
return msg_.get().as<T>();
|
||||
} catch (...) { throw std::invalid_argument("unpack failed: Args not match!"); }
|
||||
}
|
||||
|
||||
private:
|
||||
msgpack::unpacked msg_;
|
||||
};
|
||||
} // namespace rpc_service
|
||||
} // namespace rest_rpc
|
||||
|
||||
#endif // REST_RPC_CODEC_H_
|
||||
@@ -1,370 +0,0 @@
|
||||
#ifndef REST_RPC_CONNECTION_H_
|
||||
#define REST_RPC_CONNECTION_H_
|
||||
|
||||
#include <iostream>
|
||||
#include <memory>
|
||||
#include <array>
|
||||
#include <deque>
|
||||
#include "use_asio.hpp"
|
||||
#include "const_vars.h"
|
||||
#include "router.h"
|
||||
#include "cplusplus_14.h"
|
||||
|
||||
using boost::asio::ip::tcp;
|
||||
|
||||
namespace rest_rpc {
|
||||
namespace rpc_service {
|
||||
struct ssl_configure {
|
||||
std::string cert_file;
|
||||
std::string key_file;
|
||||
};
|
||||
|
||||
class connection : public std::enable_shared_from_this<connection>, private asio::noncopyable {
|
||||
public:
|
||||
connection(boost::asio::io_service& io_service, std::size_t timeout_seconds)
|
||||
: socket_(io_service),
|
||||
body_(INIT_BUF_SIZE),
|
||||
timer_(io_service),
|
||||
timeout_seconds_(timeout_seconds),
|
||||
has_closed_(false) {
|
||||
}
|
||||
|
||||
~connection() {
|
||||
close();
|
||||
}
|
||||
|
||||
void start() {
|
||||
if (is_ssl() && !has_shake_) {
|
||||
async_handshake();
|
||||
}
|
||||
else {
|
||||
read_head();
|
||||
}
|
||||
}
|
||||
|
||||
tcp::socket& socket() { return socket_; }
|
||||
|
||||
bool has_closed() const { return has_closed_; }
|
||||
uint64_t request_id() const {
|
||||
return req_id_;
|
||||
}
|
||||
|
||||
void response(uint64_t req_id, std::string data, request_type req_type = request_type::req_res) {
|
||||
auto len = data.size();
|
||||
assert(len < MAX_BUF_LEN);
|
||||
|
||||
std::unique_lock<std::mutex> lock(write_mtx_);
|
||||
write_queue_.emplace_back(message_type{ req_id, req_type, std::make_shared<std::string>(std::move(data)) });
|
||||
if (write_queue_.size() > 1) {
|
||||
return;
|
||||
}
|
||||
|
||||
write();
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
void pack_and_response(uint64_t req_id, T data) {
|
||||
auto result = msgpack_codec::pack_args_str(result_code::OK, std::move(data));
|
||||
response(req_id, std::move(result));
|
||||
}
|
||||
|
||||
void set_conn_id(int64_t id) { conn_id_ = id; }
|
||||
|
||||
int64_t conn_id() const { return conn_id_; }
|
||||
|
||||
const std::vector<char>& body() const {
|
||||
return body_;
|
||||
}
|
||||
|
||||
std::string remote_address() const {
|
||||
if (has_closed_) {
|
||||
return "";
|
||||
}
|
||||
|
||||
return socket_.remote_endpoint().address().to_string();
|
||||
}
|
||||
|
||||
void publish(const std::string& key, const std::string& data) {
|
||||
auto result = msgpack_codec::pack_args_str(result_code::OK, key, data);
|
||||
response(0, std::move(result), request_type::sub_pub);
|
||||
}
|
||||
|
||||
void set_callback(std::function<void(std::string, std::string, std::weak_ptr<connection>)> callback) {
|
||||
callback_ = std::move(callback);
|
||||
}
|
||||
|
||||
void init_ssl_context(const ssl_configure& ssl_conf) {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
unsigned long ssl_options = boost::asio::ssl::context::default_workarounds
|
||||
| boost::asio::ssl::context::no_sslv2
|
||||
| boost::asio::ssl::context::single_dh_use;
|
||||
try {
|
||||
boost::asio::ssl::context ssl_context(boost::asio::ssl::context::sslv23);
|
||||
ssl_context.set_options(ssl_options);
|
||||
ssl_context.set_password_callback([](std::size_t size,
|
||||
boost::asio::ssl::context_base::password_purpose purpose) {return "123456"; });
|
||||
|
||||
boost::system::error_code ec;
|
||||
if (rpcfs::exists(ssl_conf.cert_file, ec)) {
|
||||
ssl_context.use_certificate_chain_file(ssl_conf.cert_file);
|
||||
}
|
||||
|
||||
if (rpcfs::exists(ssl_conf.key_file, ec))
|
||||
ssl_context.use_private_key_file(ssl_conf.key_file, boost::asio::ssl::context::pem);
|
||||
|
||||
//ssl_context_callback(ssl_context);
|
||||
ssl_stream_ = std::make_unique<boost::asio::ssl::stream<boost::asio::ip::tcp::socket&>>(socket_, ssl_context);
|
||||
}
|
||||
catch (const std::exception& e) {
|
||||
print(e);
|
||||
}
|
||||
#endif
|
||||
}
|
||||
|
||||
private:
|
||||
void read_head() {
|
||||
reset_timer();
|
||||
auto self(this->shared_from_this());
|
||||
async_read_head([this, self](boost::system::error_code ec, std::size_t length) {
|
||||
if (!socket_.is_open()) {
|
||||
//LOG(INFO) << "socket already closed";
|
||||
return;
|
||||
}
|
||||
|
||||
if (!ec) {
|
||||
//const uint32_t body_len = *((int*)(head_));
|
||||
//req_id_ = *((std::uint64_t*)(head_ + sizeof(int32_t)));
|
||||
rpc_header* header = (rpc_header*)(head_);
|
||||
req_id_ = header->req_id;
|
||||
const uint32_t body_len = header->body_len;
|
||||
req_type_ = header->req_type;
|
||||
if (body_len > 0 && body_len < MAX_BUF_LEN) {
|
||||
if (body_.size() < body_len) { body_.resize(body_len); }
|
||||
read_body(body_len);
|
||||
return;
|
||||
}
|
||||
|
||||
if (body_len == 0) { // nobody, just head, maybe as heartbeat.
|
||||
cancel_timer();
|
||||
read_head();
|
||||
}
|
||||
else {
|
||||
print("invalid body len");
|
||||
close();
|
||||
}
|
||||
}
|
||||
else {
|
||||
print(ec);
|
||||
close();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void read_body(std::size_t size) {
|
||||
auto self(this->shared_from_this());
|
||||
async_read(size, [this, self](boost::system::error_code ec, std::size_t length) {
|
||||
cancel_timer();
|
||||
|
||||
if (!socket_.is_open()) {
|
||||
//LOG(INFO) << "socket already closed";
|
||||
return;
|
||||
}
|
||||
|
||||
if (!ec) {
|
||||
read_head();
|
||||
if (req_type_ == request_type::req_res) {
|
||||
router& _router = router::get();
|
||||
_router.route<connection>(body_.data(), length, this->shared_from_this());
|
||||
}
|
||||
else if (req_type_ == request_type::sub_pub) {
|
||||
try {
|
||||
msgpack_codec codec;
|
||||
auto p = codec.unpack<std::tuple<std::string, std::string>>(body_.data(), length);
|
||||
callback_(std::move(std::get<0>(p)), std::move(std::get<1>(p)), this->shared_from_this());
|
||||
}
|
||||
catch (const std::exception& ex) {
|
||||
print(ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
else {
|
||||
//LOG(INFO) << ec.message();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void write() {
|
||||
auto& msg = write_queue_.front();
|
||||
write_size_ = (uint32_t)msg.content->size();
|
||||
std::array<boost::asio::const_buffer, 4> write_buffers;
|
||||
write_buffers[0] = boost::asio::buffer(&write_size_, sizeof(uint32_t));
|
||||
write_buffers[1] = boost::asio::buffer(&msg.req_id, sizeof(uint64_t));
|
||||
write_buffers[2] = boost::asio::buffer(&msg.req_type, sizeof(request_type));
|
||||
write_buffers[3] = boost::asio::buffer(msg.content->data(), write_size_);
|
||||
|
||||
auto self = this->shared_from_this();
|
||||
async_write(write_buffers,
|
||||
[this, self](boost::system::error_code ec, std::size_t length) {
|
||||
on_write(ec, length);
|
||||
});
|
||||
}
|
||||
|
||||
void on_write(boost::system::error_code ec, std::size_t length) {
|
||||
if (ec) {
|
||||
print(ec);
|
||||
close(false);
|
||||
return;
|
||||
}
|
||||
|
||||
if (has_closed()) { return; }
|
||||
|
||||
std::unique_lock<std::mutex> lock(write_mtx_);
|
||||
write_queue_.pop_front();
|
||||
|
||||
if (!write_queue_.empty()) {
|
||||
write();
|
||||
}
|
||||
}
|
||||
|
||||
void async_handshake() {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
auto self = this->shared_from_this();
|
||||
ssl_stream_->async_handshake(boost::asio::ssl::stream_base::server,
|
||||
[this, self](const boost::system::error_code& error) {
|
||||
if (error) {
|
||||
print(error);
|
||||
close();
|
||||
return;
|
||||
}
|
||||
|
||||
has_shake_ = true;
|
||||
read_head();
|
||||
});
|
||||
#endif
|
||||
}
|
||||
|
||||
bool is_ssl() const {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
return ssl_stream_ != nullptr;
|
||||
#else
|
||||
return false;
|
||||
#endif
|
||||
}
|
||||
|
||||
template<typename Handler>
|
||||
void async_read_head(Handler handler) {
|
||||
if (is_ssl()) {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
boost::asio::async_read(*ssl_stream_, boost::asio::buffer(head_, HEAD_LEN), std::move(handler));
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
boost::asio::async_read(socket_, boost::asio::buffer(head_, HEAD_LEN), std::move(handler));
|
||||
}
|
||||
}
|
||||
|
||||
template<typename Handler>
|
||||
void async_read(size_t size_to_read, Handler handler) {
|
||||
if (is_ssl()) {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
boost::asio::async_read(*ssl_stream_, boost::asio::buffer(body_.data(), size_to_read), std::move(handler));
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
boost::asio::async_read(socket_, boost::asio::buffer(body_.data(), size_to_read), std::move(handler));
|
||||
}
|
||||
}
|
||||
|
||||
template<typename BufferType, typename Handler>
|
||||
void async_write(const BufferType& buffers, Handler handler) {
|
||||
if (is_ssl()) {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
boost::asio::async_write(*ssl_stream_, buffers, std::move(handler));
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
boost::asio::async_write(socket_, buffers, std::move(handler));
|
||||
}
|
||||
}
|
||||
|
||||
void reset_timer() {
|
||||
if (timeout_seconds_ == 0) { return; }
|
||||
|
||||
auto self(this->shared_from_this());
|
||||
timer_.expires_from_now(std::chrono::seconds(timeout_seconds_));
|
||||
timer_.async_wait([this, self](const boost::system::error_code& ec) {
|
||||
if (has_closed()) { return; }
|
||||
|
||||
if (ec) { return; }
|
||||
|
||||
//LOG(INFO) << "rpc connection timeout";
|
||||
close(false);
|
||||
});
|
||||
}
|
||||
|
||||
void cancel_timer() {
|
||||
if (timeout_seconds_ == 0) { return; }
|
||||
|
||||
timer_.cancel();
|
||||
}
|
||||
|
||||
void close(bool close_ssl = true) {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
if (close_ssl && ssl_stream_) {
|
||||
boost::system::error_code ec;
|
||||
ssl_stream_->shutdown(ec);
|
||||
ssl_stream_ = nullptr;
|
||||
}
|
||||
#endif
|
||||
if (has_closed_) {
|
||||
return;
|
||||
}
|
||||
|
||||
boost::system::error_code ignored_ec;
|
||||
socket_.shutdown(tcp::socket::shutdown_both, ignored_ec);
|
||||
socket_.close(ignored_ec);
|
||||
has_closed_ = true;
|
||||
has_shake_ = false;
|
||||
}
|
||||
|
||||
template<typename... Args>
|
||||
void print(Args... args) {
|
||||
#ifdef _DEBUG
|
||||
std::initializer_list<int>{( std::cout << args << ' ', 0)...};
|
||||
std::cout << "\n";
|
||||
#endif
|
||||
}
|
||||
|
||||
void print(const boost::system::error_code& ec) {
|
||||
print(ec.value(), ec.message());
|
||||
}
|
||||
|
||||
void print(const std::exception& ex) {
|
||||
print(ex.what());
|
||||
}
|
||||
|
||||
tcp::socket socket_;
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
std::unique_ptr<boost::asio::ssl::stream<boost::asio::ip::tcp::socket&>> ssl_stream_ = nullptr;
|
||||
#endif
|
||||
bool has_shake_ = false;
|
||||
char head_[HEAD_LEN];
|
||||
std::vector<char> body_;
|
||||
std::uint64_t req_id_;
|
||||
request_type req_type_;
|
||||
|
||||
uint32_t write_size_ = 0;
|
||||
std::mutex write_mtx_;
|
||||
|
||||
asio::steady_timer timer_;
|
||||
std::size_t timeout_seconds_;
|
||||
int64_t conn_id_ = 0;
|
||||
bool has_closed_;
|
||||
|
||||
std::deque<message_type> write_queue_;
|
||||
std::function<void(std::string, std::string, std::weak_ptr<connection>)> callback_;
|
||||
};
|
||||
} // namespace rpc_service
|
||||
} // namespace rest_rpc
|
||||
|
||||
#endif // REST_RPC_CONNECTION_H_
|
||||
@@ -1,43 +0,0 @@
|
||||
#pragma once
|
||||
#include <cstdint>
|
||||
|
||||
namespace rest_rpc {
|
||||
|
||||
enum class result_code : std::int16_t {
|
||||
OK = 0,
|
||||
FAIL = 1,
|
||||
};
|
||||
|
||||
enum class error_code {
|
||||
OK,
|
||||
UNKNOWN,
|
||||
FAIL,
|
||||
TIMEOUT,
|
||||
CANCEL,
|
||||
BADCONNECTION,
|
||||
};
|
||||
|
||||
enum class request_type : uint8_t {
|
||||
req_res,
|
||||
sub_pub
|
||||
};
|
||||
|
||||
struct message_type {
|
||||
std::uint64_t req_id;
|
||||
request_type req_type;
|
||||
std::shared_ptr<std::string> content;
|
||||
};
|
||||
|
||||
|
||||
#pragma pack (1)
|
||||
struct rpc_header {
|
||||
uint32_t body_len;
|
||||
uint64_t req_id;
|
||||
request_type req_type;
|
||||
};
|
||||
#pragma pack ()
|
||||
|
||||
static const size_t MAX_BUF_LEN = 1048576 * 10;
|
||||
static const size_t HEAD_LEN = 13;
|
||||
static const size_t INIT_BUF_SIZE = 2 * 1024;
|
||||
}
|
||||
@@ -1,108 +0,0 @@
|
||||
#ifndef REST_RPC_CPLUSPLUS_14_H_
|
||||
#define REST_RPC_CPLUSPLUS_14_H_
|
||||
|
||||
#include <type_traits>
|
||||
#include <memory>
|
||||
#include <tuple>
|
||||
|
||||
#if __cplusplus == 201103L
|
||||
|
||||
namespace std {
|
||||
template<class T>
|
||||
struct unique_if {
|
||||
typedef unique_ptr<T> single_object;
|
||||
};
|
||||
|
||||
template<class T>
|
||||
struct unique_if<T[]> {
|
||||
typedef unique_ptr<T[]> unknown_bound;
|
||||
};
|
||||
|
||||
template<class T, size_t N>
|
||||
struct unique_if<T[N]> {
|
||||
typedef void known_bound;
|
||||
};
|
||||
|
||||
template<class T, class... Args>
|
||||
typename unique_if<T>::single_object make_unique(Args&&... args) {
|
||||
return unique_ptr<T>(new T(forward<Args>(args)...));
|
||||
}
|
||||
|
||||
template<class T>
|
||||
typename unique_if<T>::unknown_bound make_unique(size_t n) {
|
||||
typedef typename remove_extent<T>::type U;
|
||||
return unique_ptr<T>(new U[n]());
|
||||
}
|
||||
|
||||
template<class T, class... Args>
|
||||
typename unique_if<T>::known_bound make_unique(Args&&...) = delete;
|
||||
|
||||
template<size_t... Ints>
|
||||
struct index_sequence {
|
||||
using type = index_sequence;
|
||||
using value_type = size_t;
|
||||
static constexpr std::size_t size() noexcept { return sizeof...(Ints); }
|
||||
};
|
||||
|
||||
// --------------------------------------------------------------
|
||||
|
||||
template<class Sequence1, class Sequence2>
|
||||
struct _merge_and_renumber;
|
||||
|
||||
template<size_t... I1, size_t... I2>
|
||||
struct _merge_and_renumber<index_sequence<I1...>, index_sequence<I2...>>
|
||||
: index_sequence<I1..., (sizeof...(I1) + I2)...> {};
|
||||
|
||||
// --------------------------------------------------------------
|
||||
|
||||
template<size_t N>
|
||||
struct make_index_sequence : _merge_and_renumber<typename make_index_sequence<N / 2>::type,
|
||||
typename make_index_sequence<N - N / 2>::type> {};
|
||||
|
||||
template<>
|
||||
struct make_index_sequence<0> : index_sequence<> {};
|
||||
template<>
|
||||
struct make_index_sequence<1> : index_sequence<0> {};
|
||||
|
||||
template<typename... T>
|
||||
using index_sequence_for = make_index_sequence<sizeof...(T)>;
|
||||
|
||||
template<bool B, class T = void>
|
||||
using enable_if_t = typename enable_if<B, T>::type;
|
||||
|
||||
template<typename T>
|
||||
using remove_const_t = typename remove_const<T>::type;
|
||||
|
||||
template<typename T>
|
||||
using remove_reference_t = typename remove_reference<T>::type;
|
||||
|
||||
template<int I, typename T>
|
||||
using tuple_element_t = typename tuple_element<I, T>::type;
|
||||
|
||||
template<typename T>
|
||||
using decay_t = typename decay<T>::type;
|
||||
|
||||
template<typename F, typename Tuple, size_t... Idx>
|
||||
auto apply_helper(F&& f, Tuple&& tp, std::index_sequence<Idx...>)
|
||||
-> decltype(std::forward<F>(f)(std::get<Idx>(std::forward<Tuple>(tp))...)) {
|
||||
return std::forward<F>(f)(std::get<Idx>(std::forward<Tuple>(tp))...);
|
||||
}
|
||||
|
||||
template<typename F, typename Tuple>
|
||||
auto apply(F&& f, Tuple&& tp)
|
||||
-> decltype(apply_helper(std::forward<F>(f), std::forward<Tuple>(tp),
|
||||
std::make_index_sequence<std::tuple_size<decay_t<Tuple>>::value>{})) {
|
||||
return apply_helper(std::forward<F>(f), std::forward<Tuple>(tp),
|
||||
std::make_index_sequence<std::tuple_size<decay_t<Tuple>>::value>{});
|
||||
}
|
||||
|
||||
template<typename F, typename... Args>
|
||||
auto invoke(F&& f, Args&&... args) -> decltype(std::forward<F>(f)(std::forward<Args>(args)...)) {
|
||||
return std::forward<F>(f)(std::forward<Args>(args)...);
|
||||
}
|
||||
|
||||
} // namespace std
|
||||
|
||||
#endif
|
||||
|
||||
#endif // REST_RPC_CPLUSPLUS_14_H_
|
||||
@@ -1,62 +0,0 @@
|
||||
#ifndef REST_RPC_IO_SERVICE_POOL_H_
|
||||
#define REST_RPC_IO_SERVICE_POOL_H_
|
||||
|
||||
#include <vector>
|
||||
#include <memory>
|
||||
#include "use_asio.hpp"
|
||||
|
||||
namespace rest_rpc {
|
||||
namespace rpc_service {
|
||||
class io_service_pool : private asio::noncopyable {
|
||||
public:
|
||||
explicit io_service_pool(std::size_t pool_size) : next_io_service_(0) {
|
||||
if (pool_size == 0) throw std::runtime_error("io_service_pool size is 0");
|
||||
|
||||
for (std::size_t i = 0; i < pool_size; ++i) {
|
||||
io_service_ptr io_service(new boost::asio::io_service);
|
||||
work_ptr work(new boost::asio::io_service::work(*io_service));
|
||||
io_services_.push_back(io_service);
|
||||
work_.push_back(work);
|
||||
}
|
||||
}
|
||||
|
||||
void run() {
|
||||
std::vector<std::shared_ptr<std::thread>> threads;
|
||||
for (std::size_t i = 0; i < io_services_.size(); ++i) {
|
||||
threads.emplace_back(
|
||||
std::make_shared<std::thread>([](io_service_ptr svr) { svr->run(); }, io_services_[i]));
|
||||
}
|
||||
|
||||
for (std::size_t i = 0; i < threads.size(); ++i) threads[i]->join();
|
||||
}
|
||||
|
||||
void stop() {
|
||||
for (std::size_t i = 0; i < io_services_.size(); ++i) {
|
||||
io_services_[i]->stop();
|
||||
}
|
||||
}
|
||||
|
||||
boost::asio::io_service& get_io_service() {
|
||||
boost::asio::io_service& io_service = *io_services_[next_io_service_];
|
||||
++next_io_service_;
|
||||
if (next_io_service_ == io_services_.size()) next_io_service_ = 0;
|
||||
return io_service;
|
||||
}
|
||||
|
||||
private:
|
||||
typedef std::shared_ptr<boost::asio::io_service> io_service_ptr;
|
||||
typedef std::shared_ptr<boost::asio::io_service::work> work_ptr;
|
||||
|
||||
/// The pool of io_services.
|
||||
std::vector<io_service_ptr> io_services_;
|
||||
|
||||
/// The work that keeps the io_services running.
|
||||
std::vector<work_ptr> work_;
|
||||
|
||||
/// The next io_service to use for a connection.
|
||||
std::size_t next_io_service_;
|
||||
};
|
||||
} // namespace rpc_service
|
||||
} // namespace rest_rpc
|
||||
|
||||
#endif // REST_RPC_IO_SERVICE_POOL_H_
|
||||
@@ -1,106 +0,0 @@
|
||||
#ifndef REST_RPC_META_UTIL_HPP
|
||||
#define REST_RPC_META_UTIL_HPP
|
||||
|
||||
#include "cplusplus_14.h"
|
||||
|
||||
namespace rest_rpc{
|
||||
|
||||
template <typename... Args, typename Func, std::size_t... Idx>
|
||||
void for_each(const std::tuple<Args...>& t, Func&& f, std::index_sequence<Idx...>) {
|
||||
(void)std::initializer_list<int> { (f(std::get<Idx>(t)), void(), 0)...};
|
||||
}
|
||||
|
||||
template <typename... Args, typename Func, std::size_t... Idx>
|
||||
void for_each_i(const std::tuple<Args...>& t, Func&& f, std::index_sequence<Idx...>) {
|
||||
(void)std::initializer_list<int> { (f(std::get<Idx>(t), std::integral_constant<size_t, Idx>{}), void(), 0)...};
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
struct function_traits;
|
||||
|
||||
template<typename Ret, typename Arg, typename... Args>
|
||||
struct function_traits<Ret(Arg, Args...)>
|
||||
{
|
||||
public:
|
||||
enum { arity = sizeof...(Args)+1 };
|
||||
typedef Ret function_type(Arg, Args...);
|
||||
typedef Ret return_type;
|
||||
using stl_function_type = std::function<function_type>;
|
||||
typedef Ret(*pointer)(Arg, Args...);
|
||||
|
||||
typedef std::tuple<Arg, Args...> tuple_type;
|
||||
typedef std::tuple<std::remove_const_t<std::remove_reference_t<Arg>>, std::remove_const_t<std::remove_reference_t<Args>>...> bare_tuple_type;
|
||||
using args_tuple = std::tuple<std::string, Arg, std::remove_const_t<std::remove_reference_t<Args>>...>;
|
||||
using args_tuple_2nd = std::tuple<std::string, std::remove_const_t<std::remove_reference_t<Args>>...>;
|
||||
};
|
||||
|
||||
template<typename Ret>
|
||||
struct function_traits<Ret()> {
|
||||
public:
|
||||
enum { arity = 0 };
|
||||
typedef Ret function_type();
|
||||
typedef Ret return_type;
|
||||
using stl_function_type = std::function<function_type>;
|
||||
typedef Ret(*pointer)();
|
||||
|
||||
typedef std::tuple<> tuple_type;
|
||||
typedef std::tuple<> bare_tuple_type;
|
||||
using args_tuple = std::tuple<std::string>;
|
||||
using args_tuple_2nd = std::tuple<std::string>;
|
||||
};
|
||||
|
||||
template<typename Ret, typename... Args>
|
||||
struct function_traits<Ret(*)(Args...)> : function_traits<Ret(Args...)>{};
|
||||
|
||||
template <typename Ret, typename... Args>
|
||||
struct function_traits<std::function<Ret(Args...)>> : function_traits<Ret(Args...)>{};
|
||||
|
||||
template <typename ReturnType, typename ClassType, typename... Args>
|
||||
struct function_traits<ReturnType(ClassType::*)(Args...)> : function_traits<ReturnType(Args...)>{};
|
||||
|
||||
template <typename ReturnType, typename ClassType, typename... Args>
|
||||
struct function_traits<ReturnType(ClassType::*)(Args...) const> : function_traits<ReturnType(Args...)>{};
|
||||
|
||||
template<typename Callable>
|
||||
struct function_traits : function_traits<decltype(&Callable::operator())>{};
|
||||
|
||||
template<typename T>
|
||||
using remove_const_reference_t = std::remove_const_t<std::remove_reference_t<T>>;
|
||||
|
||||
template<size_t... Is>
|
||||
auto make_tuple_from_sequence(std::index_sequence<Is...>)->decltype(std::make_tuple(Is...)) {
|
||||
std::make_tuple(Is...);
|
||||
}
|
||||
|
||||
template<size_t N>
|
||||
constexpr auto make_tuple_from_sequence()->decltype(make_tuple_from_sequence(std::make_index_sequence<N>{})) {
|
||||
return make_tuple_from_sequence(std::make_index_sequence<N>{});
|
||||
}
|
||||
|
||||
namespace detail {
|
||||
template <class Tuple, class F, std::size_t...Is>
|
||||
void tuple_switch(const std::size_t i, Tuple&& t, F&& f, std::index_sequence<Is...>) {
|
||||
(void)std::initializer_list<int> {
|
||||
(i == Is && (
|
||||
(void)std::forward<F>(f)(std::integral_constant<size_t, Is>{}), 0))...
|
||||
};
|
||||
}
|
||||
} // namespace detail
|
||||
|
||||
template <class Tuple, class F>
|
||||
inline void tuple_switch(const std::size_t i, Tuple&& t, F&& f) {
|
||||
constexpr auto N =
|
||||
std::tuple_size<std::remove_reference_t<Tuple>>::value;
|
||||
|
||||
detail::tuple_switch(i, std::forward<Tuple>(t), std::forward<F>(f),
|
||||
std::make_index_sequence<N>{});
|
||||
}
|
||||
|
||||
template<int N, typename... Args>
|
||||
using nth_type_of = std::tuple_element_t<N, std::tuple<Args...>>;
|
||||
|
||||
template<typename... Args>
|
||||
using last_type_of = nth_type_of<sizeof...(Args)-1, Args...>;
|
||||
}
|
||||
|
||||
#endif //REST_RPC_META_UTIL_HPP
|
||||
@@ -1,2 +0,0 @@
|
||||
#include "rest_rpc/rpc_server.h"
|
||||
#include "rest_rpc/rpc_client.hpp"
|
||||
@@ -1,190 +0,0 @@
|
||||
#ifndef REST_RPC_ROUTER_H_
|
||||
#define REST_RPC_ROUTER_H_
|
||||
|
||||
#include <functional>
|
||||
#include "use_asio.hpp"
|
||||
#include "codec.h"
|
||||
#include "meta_util.hpp"
|
||||
|
||||
namespace rest_rpc {
|
||||
enum class ExecMode { sync, async };
|
||||
const constexpr ExecMode Async = ExecMode::async;
|
||||
|
||||
namespace rpc_service {
|
||||
class connection;
|
||||
|
||||
class router : asio::noncopyable {
|
||||
public:
|
||||
static router& get() {
|
||||
static router instance;
|
||||
return instance;
|
||||
}
|
||||
|
||||
template<ExecMode model, typename Function>
|
||||
void register_handler(std::string const& name, Function f) {
|
||||
return register_nonmember_func<model>(name, std::move(f));
|
||||
}
|
||||
|
||||
template<ExecMode model, typename Function, typename Self>
|
||||
void register_handler(std::string const& name, const Function& f, Self* self) {
|
||||
return register_member_func<model>(name, f, self);
|
||||
}
|
||||
|
||||
void remove_handler(std::string const& name) { this->map_invokers_.erase(name); }
|
||||
|
||||
template<typename T>
|
||||
void route(const char* data, std::size_t size, std::weak_ptr<T> conn) {
|
||||
auto conn_sp = conn.lock();
|
||||
if (!conn_sp) {
|
||||
return;
|
||||
}
|
||||
|
||||
auto req_id = conn_sp->request_id();
|
||||
std::string result;
|
||||
try {
|
||||
msgpack_codec codec;
|
||||
auto p = codec.unpack<std::tuple<std::string>>(data, size);
|
||||
auto& func_name = std::get<0>(p);
|
||||
auto it = map_invokers_.find(func_name);
|
||||
if (it == map_invokers_.end()) {
|
||||
result = codec.pack_args_str(result_code::FAIL, "unknown function: " + func_name);
|
||||
conn_sp->response(req_id, std::move(result));
|
||||
return;
|
||||
}
|
||||
|
||||
ExecMode model;
|
||||
it->second(conn, data, size, result, model);
|
||||
if (model == ExecMode::sync) {
|
||||
if (result.size() >= MAX_BUF_LEN) {
|
||||
result = codec.pack_args_str(result_code::FAIL, "the response result is out of range: more than 10M " + func_name);
|
||||
}
|
||||
conn_sp->response(req_id, std::move(result));
|
||||
}
|
||||
}
|
||||
catch (const std::exception & ex) {
|
||||
msgpack_codec codec;
|
||||
result = codec.pack_args_str(result_code::FAIL, ex.what());
|
||||
conn_sp->response(req_id, std::move(result));
|
||||
}
|
||||
}
|
||||
|
||||
router() = default;
|
||||
|
||||
private:
|
||||
router(const router&) = delete;
|
||||
router(router&&) = delete;
|
||||
|
||||
template<typename F, size_t... I, typename Arg, typename... Args>
|
||||
static typename std::result_of<F(std::weak_ptr<connection>, Args...)>::type call_helper(
|
||||
const F & f, const std::index_sequence<I...>&, std::tuple<Arg, Args...> tup, std::weak_ptr<connection> ptr) {
|
||||
return f(ptr, std::move(std::get<I + 1>(tup))...);
|
||||
}
|
||||
|
||||
template<typename F, typename Arg, typename... Args>
|
||||
static
|
||||
typename std::enable_if<std::is_void<typename std::result_of<F(std::weak_ptr<connection>, Args...)>::type>::value>::type
|
||||
call(const F & f, std::weak_ptr<connection> ptr, std::string & result, std::tuple<Arg, Args...> tp) {
|
||||
call_helper(f, std::make_index_sequence<sizeof...(Args)>{}, std::move(tp), ptr);
|
||||
result = msgpack_codec::pack_args_str(result_code::OK);
|
||||
}
|
||||
|
||||
template<typename F, typename Arg, typename... Args>
|
||||
static
|
||||
typename std::enable_if<!std::is_void<typename std::result_of<F(std::weak_ptr<connection>, Args...)>::type>::value>::type
|
||||
call(const F & f, std::weak_ptr<connection> ptr, std::string & result, std::tuple<Arg, Args...> tp) {
|
||||
auto r = call_helper(f, std::make_index_sequence<sizeof...(Args)>{}, std::move(tp), ptr);
|
||||
msgpack_codec codec;
|
||||
result = msgpack_codec::pack_args_str(result_code::OK, r);
|
||||
}
|
||||
|
||||
template<typename F, typename Self, size_t... Indexes, typename Arg, typename... Args>
|
||||
static typename std::result_of<F(Self, std::weak_ptr<connection>, Args...)>::type call_member_helper(
|
||||
const F & f, Self * self, const std::index_sequence<Indexes...>&,
|
||||
std::tuple<Arg, Args...> tup, std::weak_ptr<connection> ptr = std::shared_ptr<connection>{ nullptr }) {
|
||||
return (*self.*f)(ptr, std::move(std::get<Indexes + 1>(tup))...);
|
||||
}
|
||||
|
||||
template<typename F, typename Self, typename Arg, typename... Args>
|
||||
static typename std::enable_if<
|
||||
std::is_void<typename std::result_of<F(Self, std::weak_ptr<connection>, Args...)>::type>::value>::type
|
||||
call_member(const F & f, Self * self, std::weak_ptr<connection> ptr, std::string & result,
|
||||
std::tuple<Arg, Args...> tp) {
|
||||
call_member_helper(f, self, typename std::make_index_sequence<sizeof...(Args)>{}, std::move(tp), ptr);
|
||||
result = msgpack_codec::pack_args_str(result_code::OK);
|
||||
}
|
||||
|
||||
template<typename F, typename Self, typename Arg, typename... Args>
|
||||
static typename std::enable_if<
|
||||
!std::is_void<typename std::result_of<F(Self, std::weak_ptr<connection>, Args...)>::type>::value>::type
|
||||
call_member(const F & f, Self * self, std::weak_ptr<connection> ptr, std::string & result,
|
||||
std::tuple<Arg, Args...> tp) {
|
||||
auto r =
|
||||
call_member_helper(f, self, typename std::make_index_sequence<sizeof...(Args)>{}, std::move(tp), ptr);
|
||||
result = msgpack_codec::pack_args_str(result_code::OK, r);
|
||||
}
|
||||
|
||||
template<typename Function, ExecMode mode = ExecMode::sync>
|
||||
struct invoker {
|
||||
template<ExecMode model>
|
||||
static inline void apply(const Function& func, std::weak_ptr<connection> conn, const char* data, size_t size,
|
||||
std::string& result, ExecMode& exe_model) {
|
||||
using args_tuple = typename function_traits<Function>::args_tuple_2nd;
|
||||
exe_model = ExecMode::sync;
|
||||
msgpack_codec codec;
|
||||
try {
|
||||
auto tp = codec.unpack<args_tuple>(data, size);
|
||||
call(func, conn, result, std::move(tp));
|
||||
exe_model = model;
|
||||
}
|
||||
catch (std::invalid_argument & e) {
|
||||
result = codec.pack_args_str(result_code::FAIL, e.what());
|
||||
}
|
||||
catch (const std::exception & e) {
|
||||
result = codec.pack_args_str(result_code::FAIL, e.what());
|
||||
}
|
||||
}
|
||||
|
||||
template<ExecMode model, typename Self>
|
||||
static inline void apply_member(const Function& func, Self* self, std::weak_ptr<connection> conn,
|
||||
const char* data, size_t size, std::string& result,
|
||||
ExecMode& exe_model) {
|
||||
using args_tuple = typename function_traits<Function>::args_tuple_2nd;
|
||||
exe_model = ExecMode::sync;
|
||||
msgpack_codec codec;
|
||||
try {
|
||||
auto tp = codec.unpack<args_tuple>(data, size);
|
||||
call_member(func, self, conn, result, std::move(tp));
|
||||
exe_model = model;
|
||||
}
|
||||
catch (std::invalid_argument & e) {
|
||||
result = codec.pack_args_str(result_code::FAIL, e.what());
|
||||
}
|
||||
catch (const std::exception & e) {
|
||||
result = codec.pack_args_str(result_code::FAIL, e.what());
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
template<ExecMode model, typename Function>
|
||||
void register_nonmember_func(std::string const& name, Function f) {
|
||||
this->map_invokers_[name] = { std::bind(&invoker<Function>::template apply<model>, std::move(f), std::placeholders::_1,
|
||||
std::placeholders::_2, std::placeholders::_3,
|
||||
std::placeholders::_4, std::placeholders::_5) };
|
||||
}
|
||||
|
||||
template<ExecMode model, typename Function, typename Self>
|
||||
void register_member_func(const std::string& name, const Function& f, Self* self) {
|
||||
this->map_invokers_[name] = { std::bind(&invoker<Function>::template apply_member<model, Self>,
|
||||
f, self, std::placeholders::_1, std::placeholders::_2,
|
||||
std::placeholders::_3, std::placeholders::_4,
|
||||
std::placeholders::_5) };
|
||||
}
|
||||
|
||||
std::unordered_map<std::string,
|
||||
std::function<void(std::weak_ptr<connection>, const char*, size_t, std::string&, ExecMode& model)>>
|
||||
map_invokers_;
|
||||
};
|
||||
} // namespace rpc_service
|
||||
} // namespace rest_rpc
|
||||
|
||||
#endif // REST_RPC_ROUTER_H_
|
||||
@@ -1,796 +0,0 @@
|
||||
#pragma once
|
||||
#include <iostream>
|
||||
#include <string>
|
||||
#include <deque>
|
||||
#include <future>
|
||||
#include "use_asio.hpp"
|
||||
#include "client_util.hpp"
|
||||
#include "const_vars.h"
|
||||
#include "meta_util.hpp"
|
||||
|
||||
using namespace rest_rpc::rpc_service;
|
||||
|
||||
namespace rest_rpc {
|
||||
class req_result {
|
||||
public:
|
||||
req_result() = default;
|
||||
req_result(string_view data) : data_(data.data(), data.length()) {}
|
||||
bool success() const {
|
||||
return !has_error(data_);
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
T as() {
|
||||
if (has_error(data_)) {
|
||||
throw std::logic_error(get_error_msg(data_));
|
||||
}
|
||||
|
||||
return get_result<T>(data_);
|
||||
}
|
||||
|
||||
void as() {
|
||||
if (has_error(data_)) {
|
||||
throw std::logic_error(get_error_msg(data_));
|
||||
}
|
||||
}
|
||||
|
||||
private:
|
||||
std::string data_;
|
||||
};
|
||||
|
||||
enum class CallModel {
|
||||
future,
|
||||
callback
|
||||
};
|
||||
const constexpr auto FUTURE = CallModel::future;
|
||||
|
||||
const constexpr size_t DEFAULT_TIMEOUT = 5000; //milliseconds
|
||||
|
||||
class rpc_client : private asio::noncopyable {
|
||||
public:
|
||||
rpc_client() : socket_(ios_), work_(ios_),
|
||||
deadline_(ios_), body_(INIT_BUF_SIZE) {
|
||||
thd_ = std::make_shared<std::thread>([this] {
|
||||
ios_.run();
|
||||
});
|
||||
}
|
||||
|
||||
rpc_client(const std::string& host, unsigned short port) : socket_(ios_), work_(ios_),
|
||||
deadline_(ios_), host_(host), port_(port), body_(INIT_BUF_SIZE) {
|
||||
thd_ = std::make_shared<std::thread>([this] {
|
||||
ios_.run();
|
||||
});
|
||||
}
|
||||
|
||||
~rpc_client() {
|
||||
close();
|
||||
stop();
|
||||
}
|
||||
|
||||
void run(){
|
||||
thd_->join();
|
||||
}
|
||||
|
||||
void set_connect_timeout(size_t milliseconds) {
|
||||
connect_timeout_ = milliseconds;
|
||||
}
|
||||
|
||||
void set_reconnect_count(int reconnect_count) {
|
||||
reconnect_cnt_ = reconnect_count;
|
||||
}
|
||||
|
||||
bool connect(size_t timeout = 3, bool is_ssl = false) {
|
||||
if (has_connected_)
|
||||
return true;
|
||||
|
||||
assert(port_ != 0);
|
||||
if (is_ssl) {
|
||||
upgrade_to_ssl();
|
||||
}
|
||||
async_connect();
|
||||
return wait_conn(timeout);
|
||||
}
|
||||
|
||||
bool connect(const std::string& host, unsigned short port, bool is_ssl = false, size_t timeout = 3) {
|
||||
if (port_==0) {
|
||||
host_ = host;
|
||||
port_ = port;
|
||||
}
|
||||
|
||||
return connect(timeout, is_ssl);
|
||||
}
|
||||
|
||||
void async_connect(const std::string& host, unsigned short port) {
|
||||
if (port_ == 0) {
|
||||
host_ = host;
|
||||
port_ = port;
|
||||
}
|
||||
|
||||
async_connect();
|
||||
}
|
||||
|
||||
bool wait_conn(size_t timeout) {
|
||||
if (has_connected_) {
|
||||
return true;
|
||||
}
|
||||
|
||||
has_wait_ = true;
|
||||
std::unique_lock<std::mutex> lock(conn_mtx_);
|
||||
bool result = conn_cond_.wait_for(lock, std::chrono::seconds(timeout),
|
||||
[this] {return has_connected_.load(); });
|
||||
has_wait_ = false;
|
||||
return has_connected_;
|
||||
}
|
||||
|
||||
void enable_auto_reconnect(bool enable = true) {
|
||||
enable_reconnect_ = enable;
|
||||
}
|
||||
|
||||
void enable_auto_heartbeat(bool enable = true) {
|
||||
if (enable) {
|
||||
reset_deadline_timer(5);
|
||||
}
|
||||
else {
|
||||
deadline_.cancel();
|
||||
}
|
||||
}
|
||||
|
||||
void update_addr(const std::string& host, unsigned short port) {
|
||||
host_ = host;
|
||||
port_ = port;
|
||||
}
|
||||
|
||||
void close(bool close_ssl = true) {
|
||||
boost::system::error_code ec;
|
||||
if (close_ssl) {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
if (ssl_stream_) {
|
||||
ssl_stream_->shutdown(ec);
|
||||
ssl_stream_ = nullptr;
|
||||
}
|
||||
#endif
|
||||
}
|
||||
|
||||
if (!has_connected_)
|
||||
return;
|
||||
|
||||
has_connected_ = false;
|
||||
socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
|
||||
socket_.close(ec);
|
||||
clear_cache();
|
||||
}
|
||||
|
||||
void set_error_callback(std::function<void(boost::system::error_code)> f) {
|
||||
err_cb_ = std::move(f);
|
||||
}
|
||||
|
||||
uint64_t reqest_id() {
|
||||
return temp_req_id_;
|
||||
}
|
||||
|
||||
bool has_connected() const {
|
||||
return has_connected_;
|
||||
}
|
||||
|
||||
//sync call
|
||||
#if __cplusplus > 201402L
|
||||
template<size_t TIMEOUT, typename T = void, typename... Args>
|
||||
auto call(const std::string& rpc_name, Args&& ... args) {
|
||||
std::future<req_result> future = async_call<FUTURE>(rpc_name, std::forward<Args>(args)...);
|
||||
auto status = future.wait_for(std::chrono::milliseconds(TIMEOUT));
|
||||
if (status == std::future_status::timeout || status == std::future_status::deferred) {
|
||||
throw std::out_of_range("timeout or deferred");
|
||||
}
|
||||
|
||||
if constexpr (std::is_void_v<T>) {
|
||||
future.get().as();
|
||||
}
|
||||
else {
|
||||
return future.get().as<T>();
|
||||
}
|
||||
}
|
||||
|
||||
template<typename T = void, typename... Args>
|
||||
auto call(const std::string& rpc_name, Args&& ... args) {
|
||||
return call<DEFAULT_TIMEOUT, T>(rpc_name, std::forward<Args>(args)...);
|
||||
}
|
||||
#else
|
||||
template<size_t TIMEOUT, typename T=void, typename... Args>
|
||||
typename std::enable_if<std::is_void<T>::value>::type call(const std::string& rpc_name, Args&& ... args) {
|
||||
std::future<req_result> future = async_call<FUTURE>(rpc_name, std::forward<Args>(args)...);
|
||||
auto status = future.wait_for(std::chrono::milliseconds(TIMEOUT));
|
||||
if (status == std::future_status::timeout || status == std::future_status::deferred) {
|
||||
throw std::out_of_range("timeout or deferred");
|
||||
}
|
||||
|
||||
future.get().as();
|
||||
}
|
||||
|
||||
template<typename T = void, typename... Args>
|
||||
typename std::enable_if<std::is_void<T>::value>::type call(const std::string& rpc_name, Args&& ... args) {
|
||||
call<DEFAULT_TIMEOUT, T>(rpc_name, std::forward<Args>(args)...);
|
||||
}
|
||||
|
||||
template<size_t TIMEOUT, typename T, typename... Args>
|
||||
typename std::enable_if<!std::is_void<T>::value, T>::type call(const std::string& rpc_name, Args&& ... args) {
|
||||
std::future<req_result> future = async_call<FUTURE>(rpc_name, std::forward<Args>(args)...);
|
||||
auto status = future.wait_for(std::chrono::milliseconds(TIMEOUT));
|
||||
if (status == std::future_status::timeout || status == std::future_status::deferred) {
|
||||
throw std::out_of_range("timeout or deferred");
|
||||
}
|
||||
|
||||
return future.get().as<T>();
|
||||
}
|
||||
|
||||
template<typename T, typename... Args>
|
||||
typename std::enable_if<!std::is_void<T>::value, T>::type call(const std::string& rpc_name, Args&& ... args) {
|
||||
return call<DEFAULT_TIMEOUT, T>(rpc_name, std::forward<Args>(args)...);
|
||||
}
|
||||
#endif
|
||||
|
||||
template<CallModel model, typename... Args>
|
||||
std::future<req_result> async_call(const std::string& rpc_name, Args&&... args) {
|
||||
auto p = std::make_shared<std::promise<req_result>>();
|
||||
std::future<req_result> future = p->get_future();
|
||||
|
||||
uint64_t fu_id = 0;
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(cb_mtx_);
|
||||
fu_id_++;
|
||||
fu_id = fu_id_;
|
||||
future_map_.emplace(fu_id, std::move(p));
|
||||
}
|
||||
|
||||
msgpack_codec codec;
|
||||
auto ret = codec.pack_args(rpc_name, std::forward<Args>(args)...);
|
||||
write(fu_id, request_type::req_res, std::move(ret));
|
||||
return future;
|
||||
}
|
||||
|
||||
template<size_t TIMEOUT = DEFAULT_TIMEOUT, typename... Args>
|
||||
void async_call(const std::string& rpc_name, std::function<void(boost::system::error_code, string_view)> cb, Args&& ... args) {
|
||||
if (!has_connected_) {
|
||||
if(cb)
|
||||
cb(boost::asio::error::make_error_code(boost::asio::error::not_connected), "not connected");
|
||||
return;
|
||||
}
|
||||
|
||||
uint64_t cb_id = 0;
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(cb_mtx_);
|
||||
callback_id_++;
|
||||
callback_id_ |= (uint64_t(1) << 63);
|
||||
cb_id = callback_id_;
|
||||
auto call = std::make_shared<call_t>(ios_, std::move(cb), TIMEOUT);
|
||||
call->start_timer();
|
||||
callback_map_.emplace(cb_id, call);
|
||||
}
|
||||
|
||||
msgpack_codec codec;
|
||||
auto ret = codec.pack_args(rpc_name, std::forward<Args>(args)...);
|
||||
write(cb_id, request_type::req_res, std::move(ret));
|
||||
}
|
||||
|
||||
void stop() {
|
||||
if (thd_ != nullptr) {
|
||||
ios_.stop();
|
||||
thd_->join();
|
||||
thd_ = nullptr;
|
||||
}
|
||||
}
|
||||
|
||||
template<typename Func>
|
||||
void subscribe(std::string key, Func f) {
|
||||
auto it = sub_map_.find(key);
|
||||
if (it != sub_map_.end()) {
|
||||
assert("duplicated subscribe");
|
||||
return;
|
||||
}
|
||||
|
||||
sub_map_.emplace(key, std::move(f));
|
||||
send_subscribe(key, "");
|
||||
key_token_set_.emplace(std::move(key), "");
|
||||
}
|
||||
|
||||
template<typename Func>
|
||||
void subscribe(std::string key, std::string token, Func f) {
|
||||
auto composite_key = key + token;
|
||||
auto it = sub_map_.find(composite_key);
|
||||
if (it != sub_map_.end()) {
|
||||
assert("duplicated subscribe");
|
||||
return;
|
||||
}
|
||||
|
||||
sub_map_.emplace(std::move(composite_key), std::move(f));
|
||||
send_subscribe(key, token);
|
||||
key_token_set_.emplace(std::move(key), std::move(token));
|
||||
}
|
||||
|
||||
template<typename T, size_t TIMEOUT = 3>
|
||||
void publish(std::string key, T&& t) {
|
||||
msgpack_codec codec;
|
||||
auto buf = codec.pack(std::move(t));
|
||||
call<TIMEOUT>("publish", std::move(key), "", std::string(buf.data(), buf.size()));
|
||||
}
|
||||
|
||||
template<typename T, size_t TIMEOUT=3>
|
||||
void publish_by_token(std::string key, std::string token, T&& t) {
|
||||
msgpack_codec codec;
|
||||
auto buf = codec.pack(std::move(t));
|
||||
call<TIMEOUT>("publish_by_token", std::move(key), std::move(token), std::string(buf.data(), buf.size()));
|
||||
}
|
||||
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
void set_ssl_context_callback(std::function<void(boost::asio::ssl::context&)> ssl_context_callback) {
|
||||
ssl_context_callback_ = std::move(ssl_context_callback);
|
||||
}
|
||||
#endif
|
||||
|
||||
private:
|
||||
void async_connect() {
|
||||
assert(port_ != 0);
|
||||
auto addr = boost::asio::ip::address::from_string(host_);
|
||||
socket_.async_connect({ addr, port_ }, [this](const boost::system::error_code& ec) {
|
||||
if (has_connected_) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (ec) {
|
||||
//std::cout << ec.message() << std::endl;
|
||||
|
||||
has_connected_ = false;
|
||||
|
||||
if (reconnect_cnt_ <= 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (reconnect_cnt_ > 0) {
|
||||
reconnect_cnt_--;
|
||||
}
|
||||
|
||||
async_reconnect();
|
||||
}
|
||||
else {
|
||||
//std::cout<<"connected ok"<<std::endl;
|
||||
if (is_ssl()) {
|
||||
handshake();
|
||||
return;
|
||||
}
|
||||
|
||||
has_connected_ = true;
|
||||
do_read();
|
||||
resend_subscribe();
|
||||
if (has_wait_)
|
||||
conn_cond_.notify_one();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void async_reconnect() {
|
||||
reset_socket();
|
||||
async_connect();
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(connect_timeout_));
|
||||
}
|
||||
|
||||
void reset_deadline_timer(size_t timeout) {
|
||||
deadline_.expires_from_now(std::chrono::seconds(timeout));
|
||||
deadline_.async_wait([this, timeout](const boost::system::error_code& ec) {
|
||||
if (!ec) {
|
||||
if (has_connected_) {
|
||||
write(0, request_type::req_res, buffer_type(0));
|
||||
}
|
||||
}
|
||||
|
||||
reset_deadline_timer(timeout);
|
||||
});
|
||||
}
|
||||
|
||||
void write(std::uint64_t req_id, request_type type, buffer_type&& message) {
|
||||
size_t size = message.size();
|
||||
assert(size < MAX_BUF_LEN);
|
||||
client_message_type msg{ req_id, type, {message.release(), size} };
|
||||
|
||||
std::unique_lock<std::mutex> lock(write_mtx_);
|
||||
outbox_.emplace_back(std::move(msg));
|
||||
if (outbox_.size() > 1) {
|
||||
// outstanding async_write
|
||||
return;
|
||||
}
|
||||
|
||||
write();
|
||||
}
|
||||
|
||||
void write() {
|
||||
auto& msg = outbox_[0];
|
||||
write_size_ = (uint32_t)msg.content.length();
|
||||
std::array<boost::asio::const_buffer, 4> write_buffers;
|
||||
write_buffers[0] = boost::asio::buffer(&write_size_, sizeof(int32_t));
|
||||
write_buffers[1] = boost::asio::buffer(&msg.req_id, sizeof(uint64_t));
|
||||
write_buffers[2] = boost::asio::buffer(&msg.req_type, sizeof(request_type));
|
||||
write_buffers[3] = boost::asio::buffer((char*)msg.content.data(), write_size_);
|
||||
|
||||
async_write(write_buffers,
|
||||
[this](const boost::system::error_code& ec, const size_t length) {
|
||||
if (ec) {
|
||||
has_connected_ = false;
|
||||
close(false);
|
||||
error_callback(ec);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
std::unique_lock<std::mutex> lock(write_mtx_);
|
||||
if (outbox_.empty()) {
|
||||
return;
|
||||
}
|
||||
|
||||
::free((char*)outbox_.front().content.data());
|
||||
outbox_.pop_front();
|
||||
|
||||
if (!outbox_.empty()) {
|
||||
// more messages to send
|
||||
this->write();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void do_read() {
|
||||
async_read_head([this](const boost::system::error_code& ec, const size_t length) {
|
||||
if (!socket_.is_open()) {
|
||||
//LOG(INFO) << "socket already closed";
|
||||
has_connected_ = false;
|
||||
return;
|
||||
}
|
||||
|
||||
if (!ec) {
|
||||
//const uint32_t body_len = *((uint32_t*)(head_));
|
||||
//auto req_id = *((std::uint64_t*)(head_ + sizeof(int32_t)));
|
||||
//auto req_type = *(request_type*)(head_ + sizeof(int32_t) + sizeof(int64_t));
|
||||
rpc_header* header = (rpc_header*)(head_);
|
||||
const uint32_t body_len = header->body_len;
|
||||
if (body_len > 0 && body_len < MAX_BUF_LEN) {
|
||||
if (body_.size() < body_len) { body_.resize(body_len); }
|
||||
read_body(header->req_id, header->req_type, body_len);
|
||||
return;
|
||||
}
|
||||
|
||||
if (body_len == 0 || body_len > MAX_BUF_LEN) {
|
||||
//LOG(INFO) << "invalid body len";
|
||||
close();
|
||||
error_callback(asio::error::make_error_code(asio::error::message_size));
|
||||
return;
|
||||
}
|
||||
}
|
||||
else {
|
||||
close(false);
|
||||
error_callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void read_body(std::uint64_t req_id, request_type req_type, size_t body_len) {
|
||||
async_read(body_len,
|
||||
[this, req_id, req_type, body_len](boost::system::error_code ec, std::size_t length) {
|
||||
//cancel_timer();
|
||||
|
||||
if (!socket_.is_open()) {
|
||||
//LOG(INFO) << "socket already closed";
|
||||
call_back(req_id, asio::error::make_error_code(asio::error::connection_aborted), {});
|
||||
return;
|
||||
}
|
||||
|
||||
if (!ec) {
|
||||
//entier body
|
||||
if (req_type == request_type::req_res) {
|
||||
call_back(req_id, ec, { body_.data(), body_len });
|
||||
}
|
||||
else if (req_type == request_type::sub_pub) {
|
||||
callback_sub(ec, { body_.data(), body_len });
|
||||
}
|
||||
else {
|
||||
close();
|
||||
error_callback(asio::error::make_error_code(asio::error::invalid_argument));
|
||||
return;
|
||||
}
|
||||
|
||||
do_read();
|
||||
}
|
||||
else {
|
||||
//LOG(INFO) << ec.message();
|
||||
has_connected_ = false;
|
||||
close();
|
||||
error_callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void send_subscribe(const std::string& key, const std::string& token) {
|
||||
msgpack_codec codec;
|
||||
auto ret = codec.pack_args(key, token);
|
||||
write(0, request_type::sub_pub, std::move(ret));
|
||||
}
|
||||
|
||||
void resend_subscribe() {
|
||||
if (key_token_set_.empty())
|
||||
return;
|
||||
|
||||
for (auto& pair : key_token_set_) {
|
||||
send_subscribe(pair.first, pair.second);
|
||||
}
|
||||
}
|
||||
|
||||
void call_back(uint64_t req_id, const boost::system::error_code& ec, string_view data) {
|
||||
temp_req_id_ = req_id;
|
||||
auto cb_flag = req_id >> 63;
|
||||
if (cb_flag) {
|
||||
std::shared_ptr<call_t> cl = nullptr;
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(cb_mtx_);
|
||||
cl = std::move(callback_map_[req_id]);
|
||||
}
|
||||
|
||||
assert(cl);
|
||||
if (!cl->has_timeout()) {
|
||||
cl->cancel();
|
||||
cl->callback(ec, data);
|
||||
}
|
||||
else {
|
||||
cl->callback(asio::error::make_error_code(asio::error::timed_out), {});
|
||||
}
|
||||
|
||||
std::unique_lock<std::mutex> lock(cb_mtx_);
|
||||
callback_map_.erase(req_id);
|
||||
}
|
||||
else {
|
||||
std::unique_lock<std::mutex> lock(cb_mtx_);
|
||||
auto& f = future_map_[req_id];
|
||||
if (ec) {
|
||||
//LOG<<ec.message();
|
||||
if (!f) {
|
||||
//std::cout << "invalid req_id" << std::endl;
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
assert(f);
|
||||
f->set_value(req_result{ data });
|
||||
future_map_.erase(req_id);
|
||||
}
|
||||
}
|
||||
|
||||
void callback_sub(const boost::system::error_code& ec, string_view result) {
|
||||
rpc_service::msgpack_codec codec;
|
||||
try {
|
||||
auto tp = codec.unpack<std::tuple<int, std::string, std::string>>(result.data(), result.size());
|
||||
auto code = std::get<0>(tp);
|
||||
auto& key = std::get<1>(tp);
|
||||
auto& data = std::get<2>(tp);
|
||||
|
||||
auto it = sub_map_.find(key);
|
||||
if (it == sub_map_.end()) {
|
||||
return;
|
||||
}
|
||||
|
||||
it->second(data);
|
||||
}
|
||||
catch (const std::exception& /*ex*/) {
|
||||
error_callback(asio::error::make_error_code(asio::error::invalid_argument));
|
||||
}
|
||||
}
|
||||
|
||||
void clear_cache() {
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(write_mtx_);
|
||||
while (!outbox_.empty()) {
|
||||
::free((char*)outbox_.front().content.data());
|
||||
outbox_.pop_front();
|
||||
}
|
||||
}
|
||||
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(cb_mtx_);
|
||||
callback_map_.clear();
|
||||
future_map_.clear();
|
||||
}
|
||||
}
|
||||
|
||||
void reset_socket(){
|
||||
boost::system::error_code igored_ec;
|
||||
socket_.close(igored_ec);
|
||||
socket_ = decltype(socket_)(ios_);
|
||||
if(!socket_.is_open()){
|
||||
socket_.open(boost::asio::ip::tcp::v4());
|
||||
}
|
||||
}
|
||||
|
||||
class call_t : asio::noncopyable, public std::enable_shared_from_this<call_t> {
|
||||
public:
|
||||
call_t(asio::io_service& ios, std::function<void(boost::system::error_code, string_view)> cb, size_t timeout) : timer_(ios),
|
||||
cb_(std::move(cb)), timeout_(timeout){
|
||||
}
|
||||
|
||||
void start_timer() {
|
||||
if (timeout_ == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
timer_.expires_from_now(std::chrono::milliseconds(timeout_));
|
||||
auto self = this->shared_from_this();
|
||||
timer_.async_wait([this, self](boost::system::error_code ec) {
|
||||
if (ec) {
|
||||
return;
|
||||
}
|
||||
|
||||
has_timeout_ = true;
|
||||
});
|
||||
}
|
||||
|
||||
void callback(boost::system::error_code ec, string_view data) {
|
||||
cb_(ec, data);
|
||||
}
|
||||
|
||||
bool has_timeout() const {
|
||||
return has_timeout_;
|
||||
}
|
||||
|
||||
void cancel() {
|
||||
if (timeout_ == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
boost::system::error_code ec;
|
||||
timer_.cancel(ec);
|
||||
}
|
||||
|
||||
private:
|
||||
boost::asio::steady_timer timer_;
|
||||
std::function<void(boost::system::error_code, string_view)> cb_;
|
||||
size_t timeout_;
|
||||
bool has_timeout_ = false;
|
||||
};
|
||||
|
||||
void error_callback(const boost::system::error_code& ec) {
|
||||
if (err_cb_) {
|
||||
err_cb_(ec);
|
||||
}
|
||||
|
||||
if (enable_reconnect_) {
|
||||
async_connect();
|
||||
}
|
||||
}
|
||||
|
||||
void set_default_error_cb() {
|
||||
err_cb_ = [this](boost::system::error_code){
|
||||
async_connect();
|
||||
};
|
||||
}
|
||||
|
||||
bool is_ssl() const {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
return ssl_stream_ != nullptr;
|
||||
#else
|
||||
return false;
|
||||
#endif
|
||||
}
|
||||
|
||||
|
||||
void handshake() {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
ssl_stream_->async_handshake(boost::asio::ssl::stream_base::client,
|
||||
[this](const boost::system::error_code& ec) {
|
||||
if (!ec) {
|
||||
has_connected_ = true;
|
||||
do_read();
|
||||
resend_subscribe();
|
||||
if (has_wait_)
|
||||
conn_cond_.notify_one();
|
||||
}
|
||||
else {
|
||||
error_callback(ec);
|
||||
close();
|
||||
}
|
||||
});
|
||||
#endif
|
||||
}
|
||||
|
||||
void upgrade_to_ssl() {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
if (ssl_stream_)
|
||||
return;
|
||||
|
||||
boost::asio::ssl::context ssl_context(boost::asio::ssl::context::sslv23);
|
||||
ssl_context.set_default_verify_paths();
|
||||
boost::system::error_code ec;
|
||||
ssl_context.set_options(boost::asio::ssl::context::default_workarounds, ec);
|
||||
if (ssl_context_callback_) {
|
||||
ssl_context_callback_(ssl_context);
|
||||
}
|
||||
ssl_stream_ = std::make_unique<boost::asio::ssl::stream<boost::asio::ip::tcp::socket&>>(socket_, ssl_context);
|
||||
//verify peer TODO
|
||||
#else
|
||||
assert(is_ssl());//please add definition CINATRA_ENABLE_SSL, not allowed coming in this branch
|
||||
#endif
|
||||
}
|
||||
|
||||
template<typename Handler>
|
||||
void async_read_head(Handler handler) {
|
||||
if (is_ssl()) {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
boost::asio::async_read(*ssl_stream_, boost::asio::buffer(head_, HEAD_LEN), std::move(handler));
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
boost::asio::async_read(socket_, boost::asio::buffer(head_, HEAD_LEN), std::move(handler));
|
||||
}
|
||||
}
|
||||
|
||||
template<typename Handler>
|
||||
void async_read(size_t size_to_read, Handler handler) {
|
||||
if (is_ssl()) {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
boost::asio::async_read(*ssl_stream_, boost::asio::buffer(body_.data(), size_to_read), std::move(handler));
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
boost::asio::async_read(socket_, boost::asio::buffer(body_.data(), size_to_read), std::move(handler));
|
||||
}
|
||||
}
|
||||
|
||||
template<typename BufferType, typename Handler>
|
||||
void async_write(const BufferType& buffers, Handler handler) {
|
||||
if (is_ssl()) {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
boost::asio::async_write(*ssl_stream_, buffers, std::move(handler));
|
||||
#endif
|
||||
}
|
||||
else {
|
||||
boost::asio::async_write(socket_, buffers, std::move(handler));
|
||||
}
|
||||
}
|
||||
|
||||
boost::asio::io_service ios_;
|
||||
asio::ip::tcp::socket socket_;
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
std::unique_ptr<boost::asio::ssl::stream<boost::asio::ip::tcp::socket&>> ssl_stream_;
|
||||
std::function<void(boost::asio::ssl::context&)> ssl_context_callback_;
|
||||
#endif
|
||||
boost::asio::io_service::work work_;
|
||||
std::shared_ptr<std::thread> thd_ = nullptr;
|
||||
|
||||
std::string host_;
|
||||
unsigned short port_ = 0;
|
||||
size_t connect_timeout_ = 1000;//s
|
||||
int reconnect_cnt_ = -1;
|
||||
std::atomic_bool has_connected_ = { false };
|
||||
std::mutex conn_mtx_;
|
||||
std::condition_variable conn_cond_;
|
||||
bool has_wait_ = false;
|
||||
|
||||
asio::steady_timer deadline_;
|
||||
|
||||
struct client_message_type {
|
||||
std::uint64_t req_id;
|
||||
request_type req_type;
|
||||
string_view content;
|
||||
};
|
||||
std::deque<client_message_type> outbox_;
|
||||
uint32_t write_size_ = 0;
|
||||
std::mutex write_mtx_;
|
||||
uint64_t fu_id_ = 0;
|
||||
std::function<void(boost::system::error_code)> err_cb_;
|
||||
bool enable_reconnect_ = false;
|
||||
|
||||
std::unordered_map<std::uint64_t, std::shared_ptr<std::promise<req_result>>> future_map_;
|
||||
std::unordered_map<std::uint64_t, std::shared_ptr<call_t>> callback_map_;
|
||||
std::mutex cb_mtx_;
|
||||
uint64_t callback_id_ = 0;
|
||||
|
||||
uint64_t temp_req_id_ = 0;
|
||||
|
||||
char head_[HEAD_LEN] = {};
|
||||
std::vector<char> body_;
|
||||
|
||||
std::unordered_map<std::string, std::function<void(string_view)>> sub_map_;
|
||||
std::set<std::pair<std::string, std::string>> key_token_set_;
|
||||
};
|
||||
}
|
||||
@@ -1,227 +0,0 @@
|
||||
#ifndef REST_RPC_RPC_SERVER_H_
|
||||
#define REST_RPC_RPC_SERVER_H_
|
||||
|
||||
#include <thread>
|
||||
#include <mutex>
|
||||
#include <condition_variable>
|
||||
#include "connection.h"
|
||||
#include "io_service_pool.h"
|
||||
#include "router.h"
|
||||
|
||||
using boost::asio::ip::tcp;
|
||||
|
||||
namespace rest_rpc {
|
||||
namespace rpc_service {
|
||||
using rpc_conn = std::weak_ptr<connection>;
|
||||
class rpc_server : private asio::noncopyable {
|
||||
public:
|
||||
rpc_server(short port, size_t size, size_t timeout_seconds = 15, size_t check_seconds = 10)
|
||||
: io_service_pool_(size),
|
||||
acceptor_(io_service_pool_.get_io_service(), tcp::endpoint(tcp::v4(), port)),
|
||||
timeout_seconds_(timeout_seconds),
|
||||
check_seconds_(check_seconds) {
|
||||
do_accept();
|
||||
check_thread_ = std::make_shared<std::thread>([this] { clean(); });
|
||||
pub_sub_thread_ = std::make_shared<std::thread>([this] { clean_sub_pub(); });
|
||||
}
|
||||
|
||||
rpc_server(short port, size_t size, ssl_configure ssl_conf, size_t timeout_seconds = 15, size_t check_seconds = 10) :
|
||||
rpc_server(port, size, timeout_seconds, check_seconds) {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
ssl_conf_ = std::move(ssl_conf);
|
||||
#else
|
||||
assert(false);//please add definition CINATRA_ENABLE_SSL, not allowed coming in this branch
|
||||
#endif
|
||||
}
|
||||
|
||||
~rpc_server() {
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(mtx_);
|
||||
stop_check_ = true;
|
||||
cv_.notify_all();
|
||||
}
|
||||
check_thread_->join();
|
||||
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(sub_mtx_);
|
||||
stop_check_pub_sub_ = true;
|
||||
sub_cv_.notify_all();
|
||||
}
|
||||
pub_sub_thread_->join();
|
||||
|
||||
io_service_pool_.stop();
|
||||
if(thd_){
|
||||
thd_->join();
|
||||
}
|
||||
}
|
||||
|
||||
void async_run() {
|
||||
thd_ = std::make_shared<std::thread>([this] { io_service_pool_.run(); });
|
||||
}
|
||||
|
||||
void run() {
|
||||
io_service_pool_.run();
|
||||
}
|
||||
|
||||
template<ExecMode model = ExecMode::sync, typename Function>
|
||||
void register_handler(std::string const& name, const Function& f) {
|
||||
router::get().register_handler<model>(name, f);
|
||||
}
|
||||
|
||||
template<ExecMode model = ExecMode::sync, typename Function, typename Self>
|
||||
void register_handler(std::string const& name, const Function& f, Self* self) {
|
||||
router::get().register_handler<model>(name, f, self);
|
||||
}
|
||||
|
||||
void set_conn_timeout_callback(std::function<void(int64_t)> callback) {
|
||||
conn_timeout_callback_ = std::move(callback);
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
void publish(const std::string& key, T data) {
|
||||
publish(key, "", std::move(data));
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
void publish_by_token(const std::string& key, std::string token, T data) {
|
||||
publish(key, std::move(token), std::move(data));
|
||||
}
|
||||
|
||||
std::set<std::string> get_token_list() {
|
||||
std::unique_lock<std::mutex> lock(sub_mtx_);
|
||||
return token_list_;
|
||||
}
|
||||
|
||||
private:
|
||||
void do_accept() {
|
||||
conn_.reset(new connection(io_service_pool_.get_io_service(), timeout_seconds_));
|
||||
conn_->set_callback([this](std::string key, std::string token, std::weak_ptr<connection> conn) {
|
||||
std::unique_lock<std::mutex> lock(sub_mtx_);
|
||||
sub_map_.emplace(std::move(key) + token, conn);
|
||||
if (!token.empty()) {
|
||||
token_list_.emplace(std::move(token));
|
||||
}
|
||||
});
|
||||
|
||||
acceptor_.async_accept(conn_->socket(), [this](boost::system::error_code ec) {
|
||||
if (ec) {
|
||||
//LOG(INFO) << "acceptor error: " << ec.message();
|
||||
}
|
||||
else {
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
if (!ssl_conf_.cert_file.empty()) {
|
||||
conn_->init_ssl_context(ssl_conf_);
|
||||
}
|
||||
#endif
|
||||
conn_->start();
|
||||
std::unique_lock<std::mutex> lock(mtx_);
|
||||
conn_->set_conn_id(conn_id_);
|
||||
connections_.emplace(conn_id_++, conn_);
|
||||
}
|
||||
|
||||
do_accept();
|
||||
});
|
||||
}
|
||||
|
||||
void clean() {
|
||||
while (!stop_check_) {
|
||||
std::unique_lock<std::mutex> lock(mtx_);
|
||||
cv_.wait_for(lock, std::chrono::seconds(check_seconds_));
|
||||
|
||||
for (auto it = connections_.cbegin(); it != connections_.cend();) {
|
||||
if (it->second->has_closed()) {
|
||||
if (conn_timeout_callback_) {
|
||||
conn_timeout_callback_(it->second->conn_id());
|
||||
}
|
||||
it = connections_.erase(it);
|
||||
}
|
||||
else {
|
||||
++it;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void clean_sub_pub() {
|
||||
while (!stop_check_pub_sub_) {
|
||||
std::unique_lock<std::mutex> lock(sub_mtx_);
|
||||
sub_cv_.wait_for(lock, std::chrono::seconds(10));
|
||||
|
||||
for (auto it = sub_map_.cbegin(); it != sub_map_.cend();) {
|
||||
auto conn = it->second.lock();
|
||||
if (conn == nullptr || conn->has_closed()) {
|
||||
it = sub_map_.erase(it);
|
||||
}
|
||||
else {
|
||||
++it;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
void publish(std::string key, std::string token, T data) {
|
||||
decltype(sub_map_.equal_range(key)) range;
|
||||
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(sub_mtx_);
|
||||
if (sub_map_.empty())
|
||||
return;
|
||||
|
||||
range = sub_map_.equal_range(key + token);
|
||||
}
|
||||
|
||||
std::shared_ptr<std::string> shared_data = get_shared_data<T>(std::move(data));
|
||||
for (auto it = range.first; it != range.second; ++it) {
|
||||
auto conn = it->second.lock();
|
||||
if (conn == nullptr || conn->has_closed()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
conn->publish(key + token, *shared_data);
|
||||
}
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
typename std::enable_if<std::is_assignable<std::string, T>::value, std::shared_ptr<std::string>>::type
|
||||
get_shared_data(std::string data) {
|
||||
return std::make_shared<std::string>(std::move(data));
|
||||
}
|
||||
|
||||
template<typename T>
|
||||
typename std::enable_if<!std::is_assignable<std::string, T>::value, std::shared_ptr<std::string>>::type
|
||||
get_shared_data(T data) {
|
||||
msgpack_codec codec;
|
||||
auto buf = codec.pack(std::move(data));
|
||||
return std::make_shared<std::string>(buf.data(), buf.size());
|
||||
}
|
||||
|
||||
io_service_pool io_service_pool_;
|
||||
tcp::acceptor acceptor_;
|
||||
std::shared_ptr<connection> conn_;
|
||||
std::shared_ptr<std::thread> thd_;
|
||||
std::size_t timeout_seconds_;
|
||||
|
||||
std::unordered_map<int64_t, std::shared_ptr<connection>> connections_;
|
||||
int64_t conn_id_ = 0;
|
||||
std::mutex mtx_;
|
||||
std::shared_ptr<std::thread> check_thread_;
|
||||
size_t check_seconds_;
|
||||
bool stop_check_ = false;
|
||||
std::condition_variable cv_;
|
||||
|
||||
std::function<void(int64_t)> conn_timeout_callback_;
|
||||
std::unordered_multimap<std::string, std::weak_ptr<connection>> sub_map_;
|
||||
std::set<std::string> token_list_;
|
||||
std::mutex sub_mtx_;
|
||||
std::condition_variable sub_cv_;
|
||||
|
||||
std::shared_ptr<std::thread> pub_sub_thread_;
|
||||
bool stop_check_pub_sub_ = false;
|
||||
|
||||
ssl_configure ssl_conf_;
|
||||
};
|
||||
} // namespace rpc_service
|
||||
} // namespace rest_rpc
|
||||
|
||||
#endif // REST_RPC_RPC_SERVER_H_
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,66 +0,0 @@
|
||||
#pragma once
|
||||
|
||||
#if defined(ASIO_STANDALONE)
|
||||
//MSVC : define environment path 'ASIO_STANDALONE_INCLUDE', e.g. 'E:\bdlibs\asio-1.10.6\include'
|
||||
|
||||
#include <asio.hpp>
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
#include <asio/ssl.hpp>
|
||||
#endif
|
||||
#include <asio/steady_timer.hpp>
|
||||
#include <asio/detail/noncopyable.hpp>
|
||||
namespace boost
|
||||
{
|
||||
namespace asio
|
||||
{
|
||||
using namespace ::asio;
|
||||
}
|
||||
namespace system {
|
||||
using ::std::error_code;
|
||||
}
|
||||
}
|
||||
#else
|
||||
#include <boost/asio.hpp>
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
#include <boost/asio/ssl.hpp>
|
||||
#endif
|
||||
#include <boost/asio/steady_timer.hpp>
|
||||
using namespace boost;
|
||||
using tcp_socket = boost::asio::ip::tcp::socket;
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
using ssl_socket = boost::asio::ssl::stream<boost::asio::ip::tcp::socket>;
|
||||
#endif
|
||||
#endif
|
||||
|
||||
#if __cplusplus > 201402L
|
||||
#include <string_view>
|
||||
using string_view = std::string_view;
|
||||
#else
|
||||
#ifdef ASIO_STANDALONE
|
||||
#include "string_view.hpp"
|
||||
using namespace nonstd;
|
||||
#else
|
||||
#include <boost/utility/string_view.hpp>
|
||||
using string_view = boost::string_view;
|
||||
#endif
|
||||
#endif
|
||||
|
||||
#ifdef CINATRA_ENABLE_SSL
|
||||
#if __cplusplus > 201402L
|
||||
#if defined (__GNUC__)
|
||||
#if __GNUC__ < 8
|
||||
#include <experimental/filesystem>
|
||||
namespace rpcfs = std::experimental::filesystem;
|
||||
#else
|
||||
#include <filesystem>
|
||||
namespace rpcfs = std::filesystem;
|
||||
#endif
|
||||
#else
|
||||
#include <boost/filesystem.hpp>
|
||||
namespace rpcfs = boost::filesystem;
|
||||
#endif
|
||||
#else
|
||||
#include <boost/filesystem.hpp>
|
||||
namespace rpcfs = boost::filesystem;
|
||||
#endif
|
||||
#endif
|
||||
Reference in New Issue
Block a user