15 : _id(CppCommon::UUID::Sequential()),
17 _io_context(_service->GetAsioContext()),
18 _strand(*_io_context),
19 _strand_required(_service->IsStrandRequired()),
23 _stream(*_io_context, *_context),
35 _send_buffer_flush_offset(0),
36 _option_keep_alive(false),
37 _option_no_delay(false)
39 assert((
service !=
nullptr) &&
"Asio service is invalid!");
41 throw CppCommon::ArgumentException(
"Asio service is invalid!");
43 assert((
context !=
nullptr) &&
"SSL context is invalid!");
45 throw CppCommon::ArgumentException(
"SSL context is invalid!");
49 : _id(CppCommon::UUID::Sequential()),
51 _io_context(_service->GetAsioContext()),
52 _strand(*_io_context),
53 _strand_required(_service->IsStrandRequired()),
58 _stream(*_io_context, *_context),
70 _send_buffer_flush_offset(0),
71 _option_keep_alive(false),
72 _option_no_delay(false)
74 assert((
service !=
nullptr) &&
"Asio service is invalid!");
76 throw CppCommon::ArgumentException(
"Asio service is invalid!");
78 assert((
context !=
nullptr) &&
"SSL context is invalid!");
80 throw CppCommon::ArgumentException(
"SSL context is invalid!");
84 : _id(CppCommon::UUID::Sequential()),
86 _io_context(_service->GetAsioContext()),
87 _strand(*_io_context),
88 _strand_required(_service->IsStrandRequired()),
93 _stream(*_io_context, *_context),
105 _send_buffer_flush_offset(0),
106 _option_keep_alive(false),
107 _option_no_delay(false)
109 assert((
service !=
nullptr) &&
"Asio service is invalid!");
111 throw CppCommon::ArgumentException(
"Asio service is invalid!");
113 assert((
context !=
nullptr) &&
"SSL context is invalid!");
115 throw CppCommon::ArgumentException(
"SSL context is invalid!");
124 asio::socket_base::receive_buffer_size option;
125 _stream.next_layer().get_option(option);
126 return option.value();
131 asio::socket_base::send_buffer_size option;
132 _stream.next_layer().get_option(option);
133 return option.value();
138 asio::socket_base::receive_buffer_size option((
int)size);
139 socket().set_option(option);
144 asio::socket_base::send_buffer_size option((
int)size);
145 socket().set_option(option);
154 _stream = asio::ssl::stream<asio::ip::tcp::socket>(*_io_context, *_context);
159 _endpoint = asio::ip::tcp::endpoint(asio::ip::make_address(_address), (
unsigned short)_port);
168 socket().connect(_endpoint, ec);
186 socket().set_option(asio::ip::tcp::socket::keep_alive(
true));
189 socket().set_option(asio::ip::tcp::no_delay(
true));
212 _stream.handshake(asio::ssl::stream_base::client, ec);
230 if (_send_buffer_main.empty())
242 _stream = asio::ssl::stream<asio::ip::tcp::socket>(*_io_context, *_context);
247 auto endpoints = resolver->resolver().resolve(_address, (_scheme.empty() ? std::to_string(_port) : _scheme), ec);
267 _endpoint = asio::connect(
socket(), endpoints, ec);
285 socket().set_option(asio::ip::tcp::socket::keep_alive(
true));
288 socket().set_option(asio::ip::tcp::no_delay(
true));
311 _stream.handshake(asio::ssl::stream_base::client, ec);
329 if (_send_buffer_main.empty())
335bool SSLClient::DisconnectInternal()
337 if (!
IsConnected() || _resolving || _connecting || _handshaking)
340 auto self(this->shared_from_this());
349 _handshaking =
false;
384 auto self(this->shared_from_this());
391 _stream = asio::ssl::stream<asio::ip::tcp::socket>(*_io_context, *_context);
394 _endpoint = asio::ip::tcp::endpoint(asio::ip::make_address(_address), (
unsigned short)_port);
403 auto async_connect_handler =
make_alloc_handler(_connect_storage, [
this, self](std::error_code ec1)
415 socket().set_option(asio::ip::tcp::socket::keep_alive(
true));
418 socket().set_option(asio::ip::tcp::no_delay(
true));
444 auto async_handshake_handler =
make_alloc_handler(_connect_storage, [
this, self](std::error_code ec2)
447 _handshaking =
false;
464 if (_send_buffer_main.empty())
471 DisconnectInternalAsync(
true);
474 if (_strand_required)
475 _stream.async_handshake(asio::ssl::stream_base::client, bind_executor(_strand, async_handshake_handler));
477 _stream.async_handshake(asio::ssl::stream_base::client, async_handshake_handler);
487 if (_strand_required)
488 socket().async_connect(_endpoint, bind_executor(_strand, async_connect_handler));
490 socket().async_connect(_endpoint, async_connect_handler);
492 if (_strand_required)
493 asio::post(_strand, connect_handler);
495 asio::post(*_io_context, connect_handler);
506 auto self(this->shared_from_this());
515 _stream = asio::ssl::stream<asio::ip::tcp::socket>(*_io_context, *_context);
518 auto async_resolve_handler =
make_alloc_handler(_connect_storage, [
this, self](std::error_code ec1, asio::ip::tcp::resolver::results_type endpoints)
534 auto async_connect_handler =
make_alloc_handler(_connect_storage, [
this, self](std::error_code ec2,
const asio::ip::tcp::endpoint&
endpoint)
549 socket().set_option(asio::ip::tcp::socket::keep_alive(
true));
552 socket().set_option(asio::ip::tcp::no_delay(
true));
578 auto async_handshake_handler =
make_alloc_handler(_connect_storage, [
this, self](std::error_code ec3)
581 _handshaking =
false;
598 if (_send_buffer_main.empty())
605 DisconnectInternalAsync(
true);
608 if (_strand_required)
609 _stream.async_handshake(asio::ssl::stream_base::client, bind_executor(_strand, async_handshake_handler));
611 _stream.async_handshake(asio::ssl::stream_base::client, async_handshake_handler);
621 if (_strand_required)
622 asio::async_connect(
socket(), endpoints, bind_executor(_strand, async_connect_handler));
624 asio::async_connect(
socket(), endpoints, async_connect_handler);
636 if (_strand_required)
637 resolver->resolver().async_resolve(_address, (_scheme.empty() ? std::to_string(_port) : _scheme), bind_executor(_strand, async_resolve_handler));
639 resolver->resolver().async_resolve(_address, (_scheme.empty() ? std::to_string(_port) : _scheme), async_resolve_handler);
641 if (_strand_required)
642 asio::post(_strand, connect_handler);
644 asio::post(*_io_context, connect_handler);
649bool SSLClient::DisconnectInternalAsync(
bool dispatch)
651 if (!
IsConnected() || _resolving || _connecting || _handshaking)
655 auto self(this->shared_from_this());
658 if (!
IsConnected() || _resolving || _connecting || _handshaking)
667 auto async_shutdown_handler =
make_alloc_handler(_connect_storage, [
this, self](std::error_code ec2) { DisconnectInternal(); });
668 if (_strand_required)
669 _stream.async_shutdown(bind_executor(_strand, async_shutdown_handler));
671 _stream.async_shutdown(async_shutdown_handler);
673 if (_strand_required)
676 asio::dispatch(_strand, disconnect_handler);
678 asio::post(_strand, disconnect_handler);
683 asio::dispatch(*_io_context, disconnect_handler);
685 asio::post(*_io_context, disconnect_handler);
697 CppCommon::Thread::Yield();
710 assert((buffer !=
nullptr) &&
"Pointer to the buffer should not be null!");
711 if (buffer ==
nullptr)
717 size_t sent = asio::write(_stream, asio::buffer(buffer, size), ec);
737size_t SSLClient::Send(
const void* buffer,
size_t size,
const CppCommon::Timespan& timeout)
745 assert((buffer !=
nullptr) &&
"Pointer to the buffer should not be null!");
746 if (buffer ==
nullptr)
751 std::condition_variable cv;
752 asio::error_code error;
753 asio::system_timer timer(_stream.get_executor());
756 auto async_done_handler = [&](asio::error_code ec)
758 std::unique_lock<std::mutex> lck(mtx);
769 timer.expires_after(timeout.chrono());
770 timer.async_wait([&](
const asio::error_code& ec) { async_done_handler(ec ? ec : asio::error::timed_out); });
774 _stream.async_write_some(asio::buffer(buffer, size), [&](std::error_code ec,
size_t write) { async_done_handler(ec); sent = write; });
777 std::unique_lock<std::mutex> lck(mtx);
778 cv.wait(lck, [&]() {
return done == 2; });
791 if (error && (error != asio::error::timed_out))
808 assert((buffer !=
nullptr) &&
"Pointer to the buffer should not be null!");
809 if (buffer ==
nullptr)
813 std::scoped_lock locker(_send_lock);
816 bool send_required = _send_buffer_main.empty() || _send_buffer_flush.empty();
819 if (((_send_buffer_main.size() + size) > _send_buffer_limit) && (_send_buffer_limit > 0))
821 SendError(asio::error::no_buffer_space);
826 const uint8_t* bytes = (
const uint8_t*)buffer;
827 _send_buffer_main.insert(_send_buffer_main.end(), bytes, bytes + size);
830 _bytes_pending = _send_buffer_main.size();
838 auto self(this->shared_from_this());
839 auto send_handler = [
this, self]()
844 if (_strand_required)
845 asio::dispatch(_strand, send_handler);
847 asio::dispatch(*_io_context, send_handler);
860 assert((buffer !=
nullptr) &&
"Pointer to the buffer should not be null!");
861 if (buffer ==
nullptr)
867 size_t received = _stream.read_some(asio::buffer(buffer, size), ec);
871 _bytes_received += received;
889 std::string text(size, 0);
890 text.resize(
Receive(text.data(), text.size()));
902 assert((buffer !=
nullptr) &&
"Pointer to the buffer should not be null!");
903 if (buffer ==
nullptr)
908 std::condition_variable cv;
909 asio::error_code error;
910 asio::system_timer timer(_stream.get_executor());
913 auto async_done_handler = [&](asio::error_code ec)
915 std::unique_lock<std::mutex> lck(mtx);
926 timer.expires_after(timeout.chrono());
927 timer.async_wait([&](
const asio::error_code& ec) { async_done_handler(ec ? ec : asio::error::timed_out); });
931 _stream.async_read_some(asio::buffer(buffer, size), [&](std::error_code ec,
size_t read) { async_done_handler(ec); received = read; });
934 std::unique_lock<std::mutex> lck(mtx);
935 cv.wait(lck, [&]() {
return done == 2; });
941 _bytes_received += received;
948 if (error && (error != asio::error::timed_out))
959 std::string text(size, 0);
960 text.resize(
Receive(text.data(), text.size(), timeout));
970void SSLClient::TryReceive()
980 auto self(this->shared_from_this());
981 auto async_receive_handler =
make_alloc_handler(_receive_storage, [
this, self](std::error_code ec,
size_t size)
992 _bytes_received += size;
998 if (_receive_buffer.size() == size)
1001 if (((2 * size) > _receive_buffer_limit) && (_receive_buffer_limit > 0))
1003 SendError(asio::error::no_buffer_space);
1004 DisconnectInternalAsync(
true);
1008 _receive_buffer.resize(2 * size);
1018 DisconnectInternalAsync(
true);
1021 if (_strand_required)
1022 _stream.async_read_some(asio::buffer(_receive_buffer.data(), _receive_buffer.size()), bind_executor(_strand, async_receive_handler));
1024 _stream.async_read_some(asio::buffer(_receive_buffer.data(), _receive_buffer.size()), async_receive_handler);
1027void SSLClient::TrySend()
1036 if (_send_buffer_flush.empty())
1038 std::scoped_lock locker(_send_lock);
1041 _send_buffer_flush.swap(_send_buffer_main);
1042 _send_buffer_flush_offset = 0;
1046 _bytes_sending += _send_buffer_flush.size();
1050 if (_send_buffer_flush.empty())
1059 auto self(this->shared_from_this());
1060 auto async_write_handler =
make_alloc_handler(_send_storage, [
this, self](std::error_code ec,
size_t size)
1071 _bytes_sending -= size;
1072 _bytes_sent += size;
1075 _send_buffer_flush_offset += size;
1078 if (_send_buffer_flush_offset == _send_buffer_flush.size())
1081 _send_buffer_flush.clear();
1082 _send_buffer_flush_offset = 0;
1095 DisconnectInternalAsync(
true);
1098 if (_strand_required)
1099 _stream.async_write_some(asio::buffer(_send_buffer_flush.data() + _send_buffer_flush_offset, _send_buffer_flush.size() - _send_buffer_flush_offset), bind_executor(_strand, async_write_handler));
1101 _stream.async_write_some(asio::buffer(_send_buffer_flush.data() + _send_buffer_flush_offset, _send_buffer_flush.size() - _send_buffer_flush_offset), async_write_handler);
1104void SSLClient::ClearBuffers()
1107 std::scoped_lock locker(_send_lock);
1110 _send_buffer_main.clear();
1111 _send_buffer_flush.clear();
1112 _send_buffer_flush_offset = 0;
1120void SSLClient::SendError(std::error_code ec)
1123 if ((ec == asio::error::connection_aborted) ||
1124 (ec == asio::error::connection_refused) ||
1125 (ec == asio::error::connection_reset) ||
1126 (ec == asio::error::eof) ||
1127 (ec == asio::error::operation_aborted))
1131 if (ec == asio::ssl::error::stream_truncated)
1133 if (ec.category() == asio::error::get_ssl_category())
1135 if ((ERR_GET_REASON(ec.value()) == SSL_R_DECRYPTION_FAILED_OR_BAD_RECORD_MAC) ||
1136 (ERR_GET_REASON(ec.value()) == SSL_R_PROTOCOL_IS_SHUTDOWN) ||
1137 (ERR_GET_REASON(ec.value()) == SSL_R_WRONG_VERSION_NUMBER))
1141 onError(ec.value(), ec.category().name(), ec.message());
bool option_keep_alive() const noexcept
Get the option: keep alive.
virtual size_t Send(const void *buffer, size_t size)
Send data to the server (synchronous).
asio::ssl::stream< asio::ip::tcp::socket >::next_layer_type & socket() noexcept
Get the client socket.
virtual void onError(int error, const std::string &category, const std::string &message)
Handle error notification.
SSLClient(const std::shared_ptr< Service > &service, const std::shared_ptr< SSLContext > &context, const std::string &address, int port)
Initialize SSL client with a given Asio service, SSL context, server address and port number.
virtual void onSent(size_t sent, size_t pending)
Handle buffer sent notification.
const std::string & scheme() const noexcept
Get the scheme name.
virtual void ReceiveAsync()
Receive data from the server (asynchronous).
bool IsHandshaked() const noexcept
Is the session handshaked?
virtual void onHandshaked()
Handle session handshaked notification.
bool IsConnected() const noexcept
Is the client connected?
virtual void onEmpty()
Handle empty send buffer notification.
virtual bool SendAsync(const void *buffer, size_t size)
Send data to the server (asynchronous).
virtual bool ConnectAsync()
Connect the client (asynchronous).
virtual void onConnecting()
Handle client connecting notification.
virtual bool DisconnectAsync()
Disconnect the client (asynchronous).
virtual void onDisconnected()
Handle client disconnected notification.
virtual size_t Receive(void *buffer, size_t size)
Receive data from the server (synchronous).
virtual void onHandshaking()
Handle session handshaking notification.
void SetupSendBufferSize(size_t size)
Setup option: send buffer size.
virtual bool ReconnectAsync()
Reconnect the client (asynchronous).
virtual bool Disconnect()
Disconnect the client (synchronous).
bool option_no_delay() const noexcept
Get the option: no delay.
const std::string & address() const noexcept
Get the server address.
asio::ip::tcp::endpoint & endpoint() noexcept
Get the client endpoint.
void SetupReceiveBufferSize(size_t size)
Setup option: receive buffer size.
std::shared_ptr< Service > & service() noexcept
Get the Asio service.
virtual bool Reconnect()
Reconnect the client (synchronous).
virtual void onReceived(const void *buffer, size_t size)
Handle buffer received notification.
virtual void onDisconnecting()
Handle client disconnecting notification.
size_t option_receive_buffer_size() const
Get the option: receive buffer size.
virtual bool Connect()
Connect the client (synchronous).
virtual void onConnected()
Handle client connected notification.
std::shared_ptr< SSLContext > & context() noexcept
Get the client SSL context.
size_t option_send_buffer_size() const
Get the option: send buffer size.
uint64_t bytes_pending() const noexcept
Get the number of bytes pending sent by the client.
int port() const noexcept
Get the server port number.
AllocateHandler< THandler > make_alloc_handler(HandlerStorage &storage, THandler handler)
Helper function to wrap a handler object to add custom allocation.
C++ Server project definitions.