35 #ifndef _ABSTRACT_TCP_SERVER2_H_ 36 #define _ABSTRACT_TCP_SERVER2_H_ 41 #include <boost/noncopyable.hpp> 42 #include <boost/shared_ptr.hpp> 48 #include <boost/asio.hpp> 49 #include <boost/asio/ssl.hpp> 50 #include <boost/array.hpp> 51 #include <boost/noncopyable.hpp> 52 #include <boost/shared_ptr.hpp> 53 #include <boost/enable_shared_from_this.hpp> 54 #include <boost/interprocess/detail/atomic.hpp> 55 #include <boost/thread/thread.hpp> 61 #undef ELECTRONEUM_DEFAULT_LOG_CATEGORY 62 #define ELECTRONEUM_DEFAULT_LOG_CATEGORY "net" 64 #define ABSTRACT_SERVER_SEND_QUE_MAX_COUNT 1000 83 template<
class t_protocol_handler>
85 :
public boost::enable_shared_from_this<connection<t_protocol_handler> >,
86 private boost::noncopyable,
100 typename t_protocol_handler::config_type
config;
105 explicit connection( boost::asio::io_service& io_service,
106 boost::shared_ptr<shared_state>
state,
111 boost::shared_ptr<shared_state>
state,
120 bool start(
bool is_income,
bool is_multithreaded);
138 virtual bool do_send(
const void* ptr,
size_t cb);
139 virtual bool do_send_chunk(
const void* ptr,
size_t cb);
140 virtual bool send_done();
141 virtual bool close();
142 virtual bool call_run_once_service_io();
143 virtual bool request_callback();
144 virtual boost::asio::io_service& get_io_service();
145 virtual bool add_ref();
146 virtual bool release();
148 boost::shared_ptr<connection<t_protocol_handler> > safe_shared_from_this();
151 void handle_receive(
const boost::system::error_code& e,
152 std::size_t bytes_transferred);
155 void handle_read(
const boost::system::error_code& e,
156 std::size_t bytes_transferred);
159 void handle_write(
const boost::system::error_code& e,
size_t cb);
162 void reset_timer(boost::posix_time::milliseconds ms,
bool add);
163 boost::posix_time::milliseconds get_default_timeout();
164 boost::posix_time::milliseconds get_timeout_from_bytes_read(
size_t bytes);
167 unsigned int host_count(
const std::string &host,
int delta = 0);
170 boost::array<char, 8192> buffer_;
171 size_t buffer_ssl_init_fill;
179 size_t m_reference_count = 0;
180 boost::shared_ptr<connection<t_protocol_handler> > m_self_ref;
190 boost::mutex m_throttle_speed_in_mutex;
191 boost::mutex m_throttle_speed_out_mutex;
193 boost::asio::deadline_timer m_timer;
195 bool m_ready_to_close;
206 template<
class t_protocol_handler>
208 :
private boost::noncopyable
210 enum try_connect_result_t
234 bool run_server(
size_t threads_count,
bool wait =
true,
const boost::thread::attributes& attrs = boost::thread::attributes());
244 const std::atomic<bool>&
get_stop_signal() const noexcept {
return m_stop_signal_sent; }
262 template<
class t_callback>
267 assert(m_state !=
nullptr);
268 return m_state->config;
275 assert(m_state !=
nullptr);
276 auto connections_count = m_state->sock_count > 0 ? (m_state->sock_count - 1) : 0;
277 return connections_count;
294 template <
class t_handler>
310 template<
class t_handler>
313 boost::shared_ptr<idle_callback_conext<t_handler>> ptr(
new idle_callback_conext<t_handler>(io_service_, t_callback, timeout_ms));
315 ptr->m_timer.expires_from_now(boost::posix_time::milliseconds(ptr->m_period));
320 template<
class t_handler>
324 if(!ptr->call_handler())
326 ptr->m_timer.expires_from_now(boost::posix_time::milliseconds(ptr->m_period));
331 template<
class t_handler>
334 io_service_.post(t_callback);
340 bool worker_thread();
342 void handle_accept(
const boost::system::error_code& e);
344 bool is_thread_worker();
346 const boost::shared_ptr<typename connection<t_protocol_handler>::shared_state> m_state;
352 : io_service(), work(io_service)
355 boost::asio::io_service io_service;
356 boost::asio::io_service::work work;
358 std::unique_ptr<worker> m_io_service_local_instance;
359 boost::asio::io_service& io_service_;
362 boost::asio::ip::tcp::acceptor acceptor_;
365 std::atomic<bool> m_stop_signal_sent;
369 size_t m_threads_count;
370 std::vector<boost::shared_ptr<boost::thread> > m_threads;
371 boost::thread::id m_main_thread_id;
372 critical_section m_threads_lock;
380 boost::mutex connections_mutex;
381 std::set<connection_ptr> connections_;
388 #include "abstract_tcp_server2.inl" virtual bool call_handler()
size_t get_threads_count()
virtual ~i_connection_filter()
idle_callback_conext(boost::asio::io_service &io_serice, t_handler &h, uint64_t period)
t_protocol_handler::connection_context t_connection_context
boosted_tcp_server(t_connection_type connection_type)
std::unique_ptr< void, close > socket
Unique ZMQ socket handle, calls zmq_close on destruction.
void send_stop_signal()
Stop the server.
bool init_server(uint32_t port, const std::string address="0.0.0.0", ssl_options_t ssl_options=ssl_support_t::e_ssl_support_autodetect)
boost::asio::deadline_timer m_timer
bool run_server(size_t threads_count, bool wait=true, const boost::thread::attributes &attrs=boost::thread::attributes())
Run the server's io_service loop.
i_connection_filter * pfilter
virtual bool call_handler()
bool add_connection(t_connection_context &out, boost::asio::ip::tcp::socket &&sock, network_address real_remote, epee::net_utils::ssl_support_t ssl_support=epee::net_utils::ssl_support_t::e_ssl_support_autodetect)
boost::asio::io_service & get_io_service()
const std::atomic< bool > & get_stop_signal() const noexcept
bool start(bool is_income, bool is_multithreaded)
Start the first asynchronous operation for the connection.
bool add_idle_handler(t_handler t_callback, uint64_t timeout_ms)
void set_connection_filter(i_connection_filter *pfilter)
std::map< std::string, t_connection_type > server_type_map
connection(boost::asio::io_service &io_service, boost::shared_ptr< shared_state > state, t_connection_type connection_type, epee::net_utils::ssl_support_t ssl_support)
Construct a connection with the given io_service.
bool timed_wait_server_stop(uint64_t wait_mseconds)
wait for service workers stop
boost::shared_ptr< connection< t_protocol_handler > > connection_ptr
Represents a single connection from a client.
idle_callback_conext_base(boost::asio::io_service &io_serice)
unsigned __int64 uint64_t
virtual ~idle_callback_conext_base()
long get_connections_count() const
try_connect_result_t try_connect(connection_ptr new_connection_l, const std::string &adr, const std::string &port, boost::asio::ip::tcp::socket &sock_, const boost::asio::ip::tcp::endpoint &remote_endpoint, const std::string &bind_ip, uint32_t conn_timeout, epee::net_utils::ssl_support_t ssl_support)
virtual bool is_remote_host_allowed(const epee::net_utils::network_address &address)=0
bool connect_async(const std::string &adr, const std::string &port, uint32_t conn_timeot, const t_callback &cb, const std::string &bind_ip="0.0.0.0", epee::net_utils::ssl_support_t ssl_support=epee::net_utils::ssl_support_t::e_ssl_support_autodetect)
void set_default_remote(epee::net_utils::network_address remote)
bool is_stop_signal_sent() const noexcept
boost::endian::big_uint16_t port
t_protocol_handler::connection_context t_connection_context
void create_server_type_map()
void set_threads_prefix(const std::string &prefix_name)
virtual ~connection() noexcept(false)
const T & move(const T &t)
void get_context(t_connection_context &context_)
implementaion for throttling of connection (count and rate-limit speed etc)
bool global_timer_handler(boost::shared_ptr< idle_callback_conext< t_handler >> ptr)
t_protocol_handler::config_type config
bool async_call(t_handler t_callback)
base for connection, contains e.g. the ratelimit hooks
bool connect(const std::string &adr, const std::string &port, uint32_t conn_timeot, t_connection_context &cn, const std::string &bind_ip="0.0.0.0", epee::net_utils::ssl_support_t ssl_support=epee::net_utils::ssl_support_t::e_ssl_support_autodetect)
bool speed_limit_is_enabled() const
tells us should we be sleeping here (e.g. do not sleep on RPC connections)
t_protocol_handler::config_type & get_config_object()