14 SSLClient::SSLClient(
const std::shared_ptr<Service>& service,
const std::shared_ptr<SSLContext>& context,
const std::string& address,
int port)
15 : _id(CppCommon::UUID::Sequential()),
17 _io_service(_service->GetAsioService()),
18 _strand(*_io_service),
19 _strand_required(_service->IsStrandRequired()),
23 _stream(*_io_service, *_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!");
48 SSLClient::SSLClient(
const std::shared_ptr<Service>& service,
const std::shared_ptr<SSLContext>& context,
const std::string& address,
const std::string& scheme)
49 : _id(CppCommon::UUID::Sequential()),
51 _io_service(_service->GetAsioService()),
52 _strand(*_io_service),
53 _strand_required(_service->IsStrandRequired()),
58 _stream(*_io_service, *_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!");
83 SSLClient::SSLClient(
const std::shared_ptr<Service>& service,
const std::shared_ptr<SSLContext>& context,
const asio::ip::tcp::endpoint& endpoint)
84 : _id(CppCommon::UUID::Sequential()),
86 _io_service(_service->GetAsioService()),
87 _strand(*_io_service),
88 _strand_required(_service->IsStrandRequired()),
89 _address(endpoint.address().to_string()),
90 _port(endpoint.port()),
93 _stream(*_io_service, *_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_service, *_context);
159 _endpoint = asio::ip::tcp::endpoint(asio::ip::make_address(_address), (
unsigned short)_port);
162 socket().connect(_endpoint, ec);
177 socket().set_option(asio::ip::tcp::socket::keep_alive(
true));
180 socket().set_option(asio::ip::tcp::no_delay(
true));
200 _stream.handshake(asio::ssl::stream_base::client, ec);
218 if (_send_buffer_main.empty())
230 _stream = asio::ssl::stream<asio::ip::tcp::socket>(*_io_service, *_context);
235 asio::ip::tcp::resolver::query query(_address, (_scheme.empty() ? std::to_string(_port) : _scheme));
236 auto endpoints = resolver->resolver().resolve(query, ec);
250 _endpoint = asio::connect(
socket(), endpoints, ec);
265 socket().set_option(asio::ip::tcp::socket::keep_alive(
true));
268 socket().set_option(asio::ip::tcp::no_delay(
true));
288 _stream.handshake(asio::ssl::stream_base::client, ec);
306 if (_send_buffer_main.empty())
312 bool SSLClient::DisconnectInternal()
314 if (!
IsConnected() || _resolving || _connecting || _handshaking)
317 auto self(this->shared_from_this());
323 _handshaking =
false;
358 auto self(this->shared_from_this());
367 _stream = asio::ssl::stream<asio::ip::tcp::socket>(*_io_service, *_context);
370 auto async_connect_handler =
make_alloc_handler(_connect_storage, [
this,
self](std::error_code ec1)
381 socket().set_option(asio::ip::tcp::socket::keep_alive(
true));
384 socket().set_option(asio::ip::tcp::no_delay(
true));
405 auto async_handshake_handler =
make_alloc_handler(_connect_storage, [
this,
self](std::error_code ec2)
407 _handshaking =
false;
424 if (_send_buffer_main.empty())
431 DisconnectInternalAsync(
true);
434 if (_strand_required)
435 _stream.async_handshake(asio::ssl::stream_base::client, bind_executor(_strand, async_handshake_handler));
437 _stream.async_handshake(asio::ssl::stream_base::client, async_handshake_handler);
449 _endpoint = asio::ip::tcp::endpoint(asio::ip::make_address(_address), (
unsigned short)_port);
451 if (_strand_required)
452 socket().async_connect(_endpoint, bind_executor(_strand, async_connect_handler));
454 socket().async_connect(_endpoint, async_connect_handler);
456 if (_strand_required)
457 _strand.post(connect_handler);
459 _io_service->post(connect_handler);
470 auto self(this->shared_from_this());
479 _stream = asio::ssl::stream<asio::ip::tcp::socket>(*_io_service, *_context);
482 auto async_resolve_handler =
make_alloc_handler(_connect_storage, [
this,
self](std::error_code ec1, asio::ip::tcp::resolver::results_type endpoints)
493 auto async_connect_handler =
make_alloc_handler(_connect_storage, [
this,
self](std::error_code ec2,
const asio::ip::tcp::endpoint&
endpoint)
507 socket().set_option(asio::ip::tcp::socket::keep_alive(
true));
510 socket().set_option(asio::ip::tcp::no_delay(
true));
531 auto async_handshake_handler =
make_alloc_handler(_connect_storage, [
this,
self](std::error_code ec3)
533 _handshaking =
false;
550 if (_send_buffer_main.empty())
557 DisconnectInternalAsync(
true);
560 if (_strand_required)
561 _stream.async_handshake(asio::ssl::stream_base::client, bind_executor(_strand, async_handshake_handler));
563 _stream.async_handshake(asio::ssl::stream_base::client, async_handshake_handler);
573 if (_strand_required)
574 asio::async_connect(
socket(), endpoints, bind_executor(_strand, async_connect_handler));
576 asio::async_connect(
socket(), endpoints, async_connect_handler);
588 asio::ip::tcp::resolver::query query(_address, (_scheme.empty() ? std::to_string(_port) : _scheme));
589 if (_strand_required)
590 resolver->resolver().async_resolve(query, bind_executor(_strand, async_resolve_handler));
592 resolver->resolver().async_resolve(query, async_resolve_handler);
594 if (_strand_required)
595 _strand.post(connect_handler);
597 _io_service->post(connect_handler);
602 bool SSLClient::DisconnectInternalAsync(
bool dispatch)
604 if (!
IsConnected() || _resolving || _connecting || _handshaking)
608 auto self(this->shared_from_this());
611 if (!
IsConnected() || _resolving || _connecting || _handshaking)
620 auto async_shutdown_handler =
make_alloc_handler(_connect_storage, [
this,
self](std::error_code ec2) { DisconnectInternal(); });
621 if (_strand_required)
622 _stream.async_shutdown(bind_executor(_strand, async_shutdown_handler));
624 _stream.async_shutdown(async_shutdown_handler);
626 if (_strand_required)
629 _strand.dispatch(disconnect_handler);
631 _strand.post(disconnect_handler);
636 _io_service->dispatch(disconnect_handler);
638 _io_service->post(disconnect_handler);
650 CppCommon::Thread::Yield();
663 assert((buffer !=
nullptr) &&
"Pointer to the buffer should not be null!");
664 if (buffer ==
nullptr)
670 size_t sent = asio::write(_stream, asio::buffer(buffer, size), ec);
690 size_t SSLClient::Send(
const void* buffer,
size_t size,
const CppCommon::Timespan& timeout)
698 assert((buffer !=
nullptr) &&
"Pointer to the buffer should not be null!");
699 if (buffer ==
nullptr)
704 std::condition_variable cv;
705 asio::error_code error;
706 asio::system_timer timer(_stream.get_executor());
709 auto async_done_handler = [&](asio::error_code ec)
711 std::unique_lock<std::mutex> lck(mtx);
722 timer.expires_from_now(timeout.chrono());
723 timer.async_wait([&](
const asio::error_code& ec) { async_done_handler(ec ? ec : asio::error::timed_out); });
727 _stream.async_write_some(asio::buffer(buffer, size), [&](std::error_code ec,
size_t write) { async_done_handler(ec); sent = write; });
730 std::unique_lock<std::mutex> lck(mtx);
731 cv.wait(lck, [&]() {
return done == 2; });
744 if (error && (error != asio::error::timed_out))
761 assert((buffer !=
nullptr) &&
"Pointer to the buffer should not be null!");
762 if (buffer ==
nullptr)
766 std::scoped_lock locker(_send_lock);
769 bool send_required = _send_buffer_main.empty() || _send_buffer_flush.empty();
772 if (((_send_buffer_main.size() + size) > _send_buffer_limit) && (_send_buffer_limit > 0))
774 SendError(asio::error::no_buffer_space);
779 const uint8_t* bytes = (
const uint8_t*)buffer;
780 _send_buffer_main.insert(_send_buffer_main.end(), bytes, bytes + size);
783 _bytes_pending = _send_buffer_main.size();
791 auto self(this->shared_from_this());
792 auto send_handler = [
this,
self]()
797 if (_strand_required)
798 _strand.dispatch(send_handler);
800 _io_service->dispatch(send_handler);
813 assert((buffer !=
nullptr) &&
"Pointer to the buffer should not be null!");
814 if (buffer ==
nullptr)
820 size_t received = _stream.read_some(asio::buffer(buffer, size), ec);
824 _bytes_received += received;
842 std::string text(size, 0);
843 text.resize(
Receive(text.data(), text.size()));
855 assert((buffer !=
nullptr) &&
"Pointer to the buffer should not be null!");
856 if (buffer ==
nullptr)
861 std::condition_variable cv;
862 asio::error_code error;
863 asio::system_timer timer(_stream.get_executor());
866 auto async_done_handler = [&](asio::error_code ec)
868 std::unique_lock<std::mutex> lck(mtx);
879 timer.expires_from_now(timeout.chrono());
880 timer.async_wait([&](
const asio::error_code& ec) { async_done_handler(ec ? ec : asio::error::timed_out); });
884 _stream.async_read_some(asio::buffer(buffer, size), [&](std::error_code ec,
size_t read) { async_done_handler(ec); received = read; });
887 std::unique_lock<std::mutex> lck(mtx);
888 cv.wait(lck, [&]() {
return done == 2; });
894 _bytes_received += received;
901 if (error && (error != asio::error::timed_out))
912 std::string text(size, 0);
913 text.resize(
Receive(text.data(), text.size(), timeout));
923 void SSLClient::TryReceive()
933 auto self(this->shared_from_this());
934 auto async_receive_handler =
make_alloc_handler(_receive_storage, [
this,
self](std::error_code ec,
size_t size)
945 _bytes_received += size;
951 if (_receive_buffer.size() == size)
954 if (((2 * size) > _receive_buffer_limit) && (_receive_buffer_limit > 0))
956 SendError(asio::error::no_buffer_space);
957 DisconnectInternalAsync(
true);
961 _receive_buffer.resize(2 * size);
971 DisconnectInternalAsync(
true);
974 if (_strand_required)
975 _stream.async_read_some(asio::buffer(_receive_buffer.data(), _receive_buffer.size()), bind_executor(_strand, async_receive_handler));
977 _stream.async_read_some(asio::buffer(_receive_buffer.data(), _receive_buffer.size()), async_receive_handler);
980 void SSLClient::TrySend()
989 if (_send_buffer_flush.empty())
991 std::scoped_lock locker(_send_lock);
994 _send_buffer_flush.swap(_send_buffer_main);
995 _send_buffer_flush_offset = 0;
999 _bytes_sending += _send_buffer_flush.size();
1003 if (_send_buffer_flush.empty())
1012 auto self(this->shared_from_this());
1013 auto async_write_handler =
make_alloc_handler(_send_storage, [
this,
self](std::error_code ec,
size_t size)
1024 _bytes_sending -= size;
1025 _bytes_sent += size;
1028 _send_buffer_flush_offset += size;
1031 if (_send_buffer_flush_offset == _send_buffer_flush.size())
1034 _send_buffer_flush.clear();
1035 _send_buffer_flush_offset = 0;
1048 DisconnectInternalAsync(
true);
1051 if (_strand_required)
1052 _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));
1054 _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);
1057 void SSLClient::ClearBuffers()
1060 std::scoped_lock locker(_send_lock);
1063 _send_buffer_main.clear();
1064 _send_buffer_flush.clear();
1065 _send_buffer_flush_offset = 0;
1073 void SSLClient::SendError(std::error_code ec)
1076 if ((ec == asio::error::connection_aborted) ||
1077 (ec == asio::error::connection_refused) ||
1078 (ec == asio::error::connection_reset) ||
1079 (ec == asio::error::eof) ||
1080 (ec == asio::error::operation_aborted))
1084 if (ec == asio::ssl::error::stream_truncated)
1086 if (ec.category() == asio::error::get_ssl_category())
1088 if ((ERR_GET_REASON(ec.value()) == SSL_R_DECRYPTION_FAILED_OR_BAD_RECORD_MAC) ||
1089 (ERR_GET_REASON(ec.value()) == SSL_R_PROTOCOL_IS_SHUTDOWN) ||
1090 (ERR_GET_REASON(ec.value()) == SSL_R_WRONG_VERSION_NUMBER))
1094 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)
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.
virtual void ReceiveAsync()
Receive data from the server (asynchronous)
bool IsHandshaked() const noexcept
Is the session handshaked?
virtual void onHandshaked()
Handle session handshaked notification.
std::shared_ptr< SSLContext > & context() noexcept
Get the client SSL context.
bool IsConnected() const noexcept
Is the client connected?
virtual void onEmpty()
Handle empty send buffer notification.
std::shared_ptr< Service > & service() noexcept
Get the Asio service.
virtual bool SendAsync(const void *buffer, size_t size)
Send data to the server (asynchronous)
virtual bool ConnectAsync()
Connect the client (asynchronous)
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)
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.
void SetupReceiveBufferSize(size_t size)
Setup option: receive buffer size.
asio::ip::tcp::endpoint & endpoint() noexcept
Get the client endpoint.
virtual bool Reconnect()
Reconnect the client (synchronous)
virtual void onReceived(const void *buffer, size_t size)
Handle buffer received 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.
asio::ssl::stream< asio::ip::tcp::socket >::next_layer_type & socket() noexcept
Get the client socket.
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.
AllocateHandler< THandler > make_alloc_handler(HandlerStorage &storage, THandler handler)
Helper function to wrap a handler object to add custom allocation.
C++ Server project definitions.