diff --git a/src/internal/sio_client_impl.cpp b/src/internal/sio_client_impl.cpp index 3cd98da8..d50c4e84 100644 --- a/src/internal/sio_client_impl.cpp +++ b/src/internal/sio_client_impl.cpp @@ -1,3 +1,4 @@ + // // sio_client_impl.cpp // SioChatDemo @@ -7,31 +8,38 @@ // #include "sio_client_impl.h" +#include #include #include #include #include #include +#include // Comment this out to disable handshake logging to stdout -#if DEBUG || _DEBUG -#define LOG(x) std::cout << x + +#ifndef _SAL_TIME_H +#define SAL_FUNC_DEBUG cout<< +#define SAL_FUNC_VERBOSE cout<< +#define SAL_FUNC_INFO cout<< +#define SAL_FUNC_WARN cout<< +#define SAL_FUNC_ERROR cout<< #else -#define LOG(x) -#endif +#include "SAL/Log/Log.h" +#include "SAL/OS/SocketQueue.h" +#include "SAL/OS/Sync.h" -#if SIO_TLS -// If using Asio's SSL support, you will also need to add this #include. -// Source: http://think-async.com/Asio/asio-1.10.6/doc/asio/using.html -// #include -#endif +namespace { + LOGGER_NAME("socketio.client") +} -using std::chrono::milliseconds; -using namespace std; +SAL::TimerManagerPtr getTimerMgr(); + +#endif namespace sio { /*************************public:*************************/ - client_impl::client_impl() : + client_impl::client_impl(ProtocolVersion version) : m_ping_interval(0), m_ping_timeout(0), m_network_thread(), @@ -39,16 +47,19 @@ namespace sio m_reconn_delay(5000), m_reconn_delay_max(25000), m_reconn_attempts(0xFFFFFFFF), - m_reconn_made(0) + m_reconn_made(0), + m_protocol_version(version) { using websocketpp::log::alevel; + m_client.clear_access_channels(alevel::frame_header|alevel::frame_payload); #ifndef DEBUG m_client.clear_access_channels(alevel::all); m_client.set_access_channels(alevel::connect|alevel::disconnect|alevel::app); #endif +#ifndef _SAL_TIME_H // Initialize the Asio transport policy m_client.init_asio(); - +#endif // Bind the clients we are using using std::placeholders::_1; using std::placeholders::_2; @@ -63,27 +74,16 @@ namespace sio m_packet_mgr.set_encode_callback(std::bind(&client_impl::on_encode,this,_1,_2)); } - + client_impl::~client_impl() { this->sockets_invoke_void(&sio::socket::on_close); sync_close(); } - - void client_impl::set_proxy_basic_auth(const std::string& uri, const std::string& username, const std::string& password) - { - m_proxy_base_url = uri; - m_proxy_basic_username = username; - m_proxy_basic_password = password; - } - + void client_impl::connect(const string& uri, const map& query, const map& headers, const message::ptr& auth) { - if(m_reconn_timer) - { - m_reconn_timer->cancel(); - m_reconn_timer.reset(); - } + reset_timer(m_reconn_timer); if(m_network_thread) { if(m_con_state == con_closing||m_con_state == con_closed) @@ -112,16 +112,18 @@ namespace sio string query_str_value=encode_query_string(it->second); query_str.append(query_str_value); } - m_query_string=move(query_str); - + m_query_string=std::move(query_str); m_http_headers = headers; m_auth = auth; this->reset_states(); m_abort_retries = false; +#ifdef _SAL_TIME_H + this->connect_impl(uri, m_query_string); +#else m_client.get_io_service().dispatch(std::bind(&client_impl::connect_impl,this,uri,m_query_string)); m_network_thread.reset(new thread(std::bind(&client_impl::run_loop,this)));//uri lifecycle? - +#endif } socket::ptr const& client_impl::socket(string const& nsp) @@ -143,6 +145,7 @@ namespace sio } auto it = m_sockets.find(aux); + SAL_FUNC_INFO("%d", it!= m_sockets.end()); if(it!= m_sockets.end()) { return it->second; @@ -159,7 +162,19 @@ namespace sio m_con_state = con_closing; m_abort_retries = true; this->sockets_invoke_void(&sio::socket::close); - m_client.get_io_service().dispatch(std::bind(&client_impl::close_impl, this,close::status::normal,"End by user")); +#ifdef _SAL_TIME_H + // The websocket transport drives I/O on the socket-queue thread, so the + // close must run there too — otherwise it races with inbound frame + // handling. Post() runs inline when already on that thread (the common + // case here: close() is invoked from on_fail/on_open). + SAL_FUNC_INFO("close(): posting close_impl to socket-queue thread"); + SAL::MSocketQueueManager::Post([this]() { + SAL_FUNC_INFO("close(): close_impl running on socket-queue thread"); + this->close_impl(close::status::normal, "End by user"); + }); +#else + close_impl(close::status::normal,"End by user"); +#endif } void client_impl::sync_close() @@ -167,7 +182,42 @@ namespace sio m_con_state = con_closing; m_abort_retries = true; this->sockets_invoke_void(&sio::socket::close); - m_client.get_io_service().dispatch(std::bind(&client_impl::close_impl, this,close::status::normal,"End by user")); +#ifdef _SAL_TIME_H + // Called from ~client_impl: the endpoint is about to be destroyed. + // Run close_impl on the socket-queue thread (serialized with inbound + // frame handling) and block until it has run, so no SocketQueueKpoll + // callback can touch this connection after we return. Run inline if we + // are already on that thread, to avoid self-deadlock. + if (SAL::MSocketQueueManager::IsOnQueueThread()) + { + SAL_FUNC_INFO("sync_close: force-terminating inline (already on socket-queue thread)"); + force_close_impl(); + SAL_FUNC_INFO("sync_close: close completed (inline)"); + } + else + { + SAL_FUNC_INFO("sync_close: dispatching force-close to socket-queue thread and waiting"); + SAL::Event done(false); + SAL::MSocketQueueManager::Post([this, &done]() { + SAL_FUNC_INFO("sync_close: force_close_impl running on socket-queue thread"); + this->force_close_impl(); + done.signal(); + }); + // Bounded wait. A live loop runs the task within its 1s heartbeat; + // the timeout only guards against a wedged/dead loop so teardown + // can't block forever. + if (!done.wait(5000)) + { + SAL_FUNC_ERROR("sync_close: timed out waiting for socket-queue close"); + } + else + { + SAL_FUNC_INFO("sync_close: close completed on socket-queue thread"); + } + } +#else + close_impl(close::status::normal,"End by user"); +#endif if(m_network_thread) { m_network_thread->join(); @@ -206,15 +256,17 @@ namespace sio m_sockets.erase(it); } } - +#ifndef _SAL_TIME_H asio::io_service& client_impl::get_io_service() { - return m_client.get_io_service(); + return m_client.get_io_service(); } - +#endif void client_impl::on_socket_closed(string const& nsp) { if(m_socket_close_listener)m_socket_close_listener(nsp); + SAL_FUNC_INFO("on_close if nsp==/"); + if (nsp == "/") on_close(m_con); } void client_impl::on_socket_opened(string const& nsp) @@ -225,11 +277,12 @@ namespace sio /*************************private:*************************/ void client_impl::run_loop() { - +#ifndef _SAL_TIME_H m_client.run(); m_client.reset(); m_client.get_alog().write(websocketpp::log::alevel::devel, - "run loop end"); + "run loop end"); +#endif } void client_impl::connect_impl(const string& uri, const string& queryString) @@ -237,11 +290,11 @@ namespace sio do{ websocketpp::uri uo(uri); ostringstream ss; -#if SIO_TLS - ss<<"wss://"; -#else - ss<<"ws://"; -#endif + if (uo.get_secure()) { + ss<<"wss://"; + } else { + ss<<"ws://"; + } const std::string host(uo.get_host()); // As per RFC2732, literal IPv6 address should be enclosed in "[" and "]". if(host.find(':')!=std::string::npos){ @@ -253,8 +306,9 @@ namespace sio // If a resource path was included in the URI, use that, otherwise // use the default /socket.io/. const std::string path(uo.get_resource() == "/" ? "/socket.io/" : uo.get_resource()); + int proto = (int)m_protocol_version; - ss<<":"<0){ ss<<"&sid="<replace_header(header.first, header.second); } - - if (!m_proxy_base_url.empty()) { - con->set_proxy(m_proxy_base_url, ec); - if (ec) { - m_client.get_alog().write(websocketpp::log::alevel::app, - "Set Proxy Error: " + ec.message()); - break; - } - if (!m_proxy_basic_username.empty()) { - con->set_proxy_basic_auth(m_proxy_basic_username, m_proxy_basic_password, ec); - if (ec) { - m_client.get_alog().write(websocketpp::log::alevel::app, - "Set Proxy Basic Auth Error: " + ec.message()); - break; - } - } - } m_client.connect(con); return; @@ -300,21 +337,59 @@ namespace sio void client_impl::close_impl(close::status::value const& code,string const& reason) { - LOG("Close by reason:"<cancel(); - m_reconn_timer.reset(); - } + SAL_FUNC_INFO("Close by reason: %s", reason.c_str()); + reset_timer(m_reconn_timer); if (m_con.expired()) { - cerr << "Error: No active session" << endl; + SAL_FUNC_ERROR("Error: No active session"); } else { lib::error_code ec; m_client.close(m_con, code, reason, ec); + if(ec) + { + SAL_FUNC_WARN("close failed,reason: %d / %s", (int)ec.value(), ec.message().c_str()); // linux.socket: write(): send(143) failed with error 32 + this->on_close(m_con); // force to closed even fail to avoid timeout_connection if reuse as con_opened + } + } + } + + void client_impl::force_close_impl() + { + // Must run on the socket-queue thread (see sync_close). A graceful + // close (close_impl → m_client.close) only sends the close + // frame and waits for the peer's reply before the transport is shut + // down — that shutdown (which UNREGISTERS the socket from the socket + // queue) would otherwise happen AFTER ~client_impl, leaving the + // websocketpp connection alive in the loop with handlers bound to a + // freed client_impl → use-after-free in on_close. + // + // connection::terminate() forces it now: async_shutdown unregisters the + // socket synchronously, then handle_terminate fires on_close — all + // while client_impl is still alive. terminate() is idempotent. + reset_timer(m_reconn_timer); + + // Prefer the strong ref (survives on_close's m_con.reset()). Fall back + // to the weak hdl if it is still valid. + client_type::connection_ptr con = m_con_strong; + if (!con && !m_con.expired()) + { + lib::error_code ec; + con = m_client.get_con_from_hdl(m_con, ec); + if (ec) con.reset(); + } + + if (con) + { + SAL_FUNC_INFO("force_close_impl: terminating connection"); + con->terminate(lib::error_code()); + } + else + { + SAL_FUNC_INFO("force_close_impl: no active connection"); } + m_con_strong.reset(); } void client_impl::send_impl(shared_ptr const& payload_ptr,frame::opcode::value opcode) @@ -328,19 +403,48 @@ namespace sio cerr<<"Send failed,reason:"<< ec.message()< payload) + { + this->send_impl(payload, frame::opcode::text); + }); + + update_timeout_timer(); + } + + void client_impl::timeout_reconnect(std::error_code const& ec) { if(ec) { @@ -350,18 +454,19 @@ namespace sio { m_con_state = con_opening; m_reconn_made++; + this->sockets_invoke_void(&sio::socket::on_close); this->reset_states(); - LOG("Reconnecting..."<connect_impl(m_base_url,m_query_string); } } unsigned client_impl::next_delay() const { //no jitter, fixed power root. - unsigned reconn_made = min(m_reconn_made,32);//protect the pow result to be too big. - return static_cast(min(m_reconn_delay * pow(1.5,reconn_made),m_reconn_delay_max)); + unsigned reconn_made = std::min(m_reconn_made,32);//protect the pow result to be too big. + return static_cast(std::min(m_reconn_delay * pow(1.5,reconn_made),m_reconn_delay_max)); } socket::ptr client_impl::get_socket_locked(string const& nsp) @@ -393,7 +498,7 @@ namespace sio void client_impl::on_fail(connection_hdl) { if (m_con_state == con_closing) { - LOG("Connection failed while closing." << endl); + SAL_FUNC_WARN("Connection failed while closing."); this->close(); return; } @@ -401,56 +506,59 @@ namespace sio m_con.reset(); m_con_state = con_closed; this->sockets_invoke_void(&sio::socket::on_disconnect); - LOG("Connection failed." << endl); + SAL_FUNC_ERROR("Connection failed."); if(m_reconn_madenext_delay(); if(m_reconnect_listener) m_reconnect_listener(m_reconn_made,delay); - m_reconn_timer.reset(new asio::steady_timer(m_client.get_io_service())); - asio::error_code ec; - m_reconn_timer->expires_from_now(milliseconds(delay), ec); - m_reconn_timer->async_wait(std::bind(&client_impl::timeout_reconnect,this, std::placeholders::_1)); + update_timer(m_reconn_timer, delay, &client_impl::timeout_reconnect); } else { if(m_fail_listener)m_fail_listener(); } } - + void client_impl::on_open(connection_hdl con) { if (m_con_state == con_closing) { - LOG("Connection opened while closing." << endl); + SAL_FUNC_WARN("Connection opened while closing."); this->close(); return; } - LOG("Connected." << endl); + SAL_FUNC_INFO("Connected."); m_con_state = con_opened; m_con = con; + { + // Keep a strong ref so teardown can always reach the connection + // even after on_close resets the weak m_con. + lib::error_code __ec; + m_con_strong = m_client.get_con_from_hdl(con, __ec); + } m_reconn_made = 0; this->sockets_invoke_void(&sio::socket::on_open); this->socket(""); if(m_open_listener)m_open_listener(); } - + void client_impl::on_close(connection_hdl con) { - LOG("Client Disconnected." << endl); + SAL_FUNC_INFO("Client Disconnected."); con_state m_con_state_was = m_con_state; m_con_state = con_closed; lib::error_code ec; close::status::value code = close::status::normal; client_type::connection_ptr conn_ptr = m_client.get_con_from_hdl(con, ec); if (ec) { - LOG("OnClose get conn failed"<get_local_close_code(); } - + m_con.reset(); this->clear_timers(); client::close_reason reason; @@ -468,30 +576,27 @@ namespace sio this->sockets_invoke_void(&sio::socket::on_disconnect); if(m_reconn_madenext_delay(); if(m_reconnect_listener) m_reconnect_listener(m_reconn_made,delay); - m_reconn_timer.reset(new asio::steady_timer(m_client.get_io_service())); - asio::error_code ec; - m_reconn_timer->expires_from_now(milliseconds(delay), ec); - m_reconn_timer->async_wait(std::bind(&client_impl::timeout_reconnect,this, std::placeholders::_1)); + update_timer(m_reconn_timer, delay, &client_impl::timeout_reconnect); return; } reason = client::close_reason_drop; } - + if(m_close_listener) { m_close_listener(reason); } } - + void client_impl::on_message(connection_hdl, client_type::message_ptr msg) { // Parse the incoming message according to socket.IO rules m_packet_mgr.put_payload(msg->get_payload()); } - + void client_impl::on_handshake(message::ptr const& message) { if(message && message->get_flag() == message::flag_object) @@ -514,8 +619,8 @@ namespace sio { m_ping_interval = 25000; } - it = values->find("pingTimeout"); + it = values->find("pingTimeout"); if (it!=values->end()&&it->second->get_flag() == message::flag_integer) { m_ping_timeout = (unsigned) static_pointer_cast(it->second)->get_int(); } @@ -525,26 +630,53 @@ namespace sio } // Start ping timeout - update_ping_timeout_timer(); + switch (m_protocol_version) + { + case ProtocolVersion3: + // in protocol v3, the client sends a ping, and the server answers with a pong + update_send_timer(); + break; + case ProtocolVersion4: // + // in protocol v4, the server sends a ping, and the client answers with a pong + update_timeout_timer(); + break; + } return; } failed: //just close it. - m_client.get_io_service().dispatch(std::bind(&client_impl::close_impl, this,close::status::policy_violation,"Handshake error")); + // m_client.get_io_service().dispatch(std::bind(&client_impl::close_impl, this,close::status::policy_violation,"Handshake error")); + close_impl(close::status::policy_violation,"Handshake error"); } void client_impl::on_ping() { + if (m_protocol_version < ProtocolVersion4) + { + // in protocol v4, the server sends a ping, and the client answers with a pong + SAL_FUNC_DEBUG("Got ping");// don't return; + } // Reply with pong packet. packet p(packet::frame_pong); m_packet_mgr.encode(p, [&](bool /*isBin*/,shared_ptr payload) { - this->m_client.send(this->m_con, *payload, frame::opcode::text); + this->send_impl(payload, frame::opcode::text); }); // Reset the ping timeout. - update_ping_timeout_timer(); + update_timeout_timer(); + } + + void client_impl::on_pong() + { + SAL_FUNC_DEBUG("Got pong"); + + // Clear the waiting-for-ping timer + reset_timer(m_timeout_timer); + + // Reset the ping timeout. + update_send_timer(); } void client_impl::on_decode(packet const& p) @@ -554,7 +686,7 @@ namespace sio case packet::frame_message: { socket::ptr so_ptr = get_socket_locked(p.get_nsp()); - if(so_ptr)so_ptr->on_message_packet(p); + if(so_ptr)so_ptr->on_message_packet(p);else SAL_FUNC_VERBOSE("notfound"); break; } case packet::frame_open: @@ -567,46 +699,85 @@ namespace sio case packet::frame_ping: this->on_ping(); break; + case packet::frame_pong: + this->on_pong(); + break; default: + SAL_FUNC_VERBOSE("unknown"); break; } } - + void client_impl::on_encode(bool isBinary,shared_ptr const& payload) { - LOG("encoded payload length:"<length()<length()); + // m_client.get_io_service().dispatch(std::bind(&client_impl::send_impl,this,payload,isBinary?frame::opcode::binary:frame::opcode::text)); + send_impl(payload,isBinary?frame::opcode::binary:frame::opcode::text); } - + void client_impl::clear_timers() { - LOG("clear timers"< lk(m_timer_mutex); + reset_timer(m_timeout_timer); + reset_timer(m_send_timer); + } + + void client_impl::reset_timer(TIMER &timer) { + // m_timer_mutex is recursive so callers that already hold it + // (clear_timers, update_timer) can re-enter this helper. Take + // ownership of the timer pointer before operating on it: with the lock + // held this can't race a concurrent reset/update on the same field, so + // cancel+destroy run exactly once and never on freed memory. + std::lock_guard lk(m_timer_mutex); + if (auto* t = timer.release()) { - m_ping_timeout_timer->cancel(ec); - m_ping_timeout_timer.reset(); + t->cancel(); + delete t; } - } + } - void client_impl::update_ping_timeout_timer() { - if (!m_ping_timeout_timer) { - m_ping_timeout_timer = std::unique_ptr(new asio::steady_timer(get_io_service())); + void client_impl::update_send_timer() { + if (m_protocol_version > ProtocolVersion3) { // both can come here but only v3 + // in protocol v3, the client sends a ping, and the server answers with a pong + return; } + SAL_FUNC_DEBUG("Set m_send_timer %" PRId64, (int64_t)m_ping_interval); + update_timer(m_send_timer, m_ping_interval, &client_impl::timeout_send ); + } - asio::error_code ec; - m_ping_timeout_timer->expires_from_now(milliseconds(m_ping_interval + m_ping_timeout), ec); - m_ping_timeout_timer->async_wait(std::bind(&client_impl::timeout_ping, this, std::placeholders::_1)); + void client_impl::update_timeout_timer() {// both can come here + int64_t _timeout = m_ping_timeout + (m_protocol_version > ProtocolVersion3 ? m_ping_interval : 0); + SAL_FUNC_DEBUG("Set m_timeout_timer %" PRId64, _timeout); + update_timer(m_timeout_timer, _timeout, &client_impl::timeout_wait); } - + + void client_impl::update_timer(TIMER &timer, int timeout, func_ptr name) + { +#ifdef _SAL_TIME_H + std::lock_guard lk(m_timer_mutex); + reset_timer(timer); + timer.reset(SAL::timer_cb::set_timer(timeout, std::bind(name, this, std::placeholders::_1))); +#else + if (!timer) { + timer = std::unique_ptr(new asio::steady_timer(get_io_service())); + } + + asio::error_code ec; + timer->expires_from_now(std::chrono::milliseconds(timeout), ec); + timer->async_wait(std::bind(name, this, std::placeholders::_1)); +#endif + + } + void client_impl::reset_states() { m_client.reset(); m_sid.clear(); m_packet_mgr.reset(); } - + #if SIO_TLS client_impl::context_ptr client_impl::on_tls_init(connection_hdl conn) { @@ -620,7 +791,7 @@ namespace sio { cerr<<"Init tls failed,reason:"<< ec.message()< -#if _DEBUG || DEBUG -#if SIO_TLS -#include -typedef websocketpp::config::debug_asio_tls client_config; -#else -#include -typedef websocketpp::config::debug_asio client_config; -#endif //SIO_TLS -#else -#if SIO_TLS -#include -typedef websocketpp::config::asio_tls_client client_config; -#else -#include -typedef websocketpp::config::asio_client client_config; -#endif //SIO_TLS -#endif //DEBUG +#include "WebsocketSummitAdapter.h" +#include "WebsocketSummitTimer.h" -#if SIO_TLS -#include -#endif - -#include -#include -#include +typedef websocketpp::config::summit_tls_client client_config; #include #include @@ -48,11 +27,16 @@ typedef websocketpp::config::asio_client client_config; namespace sio { using namespace websocketpp; - + typedef websocketpp::client client_type; - +#ifdef _SAL_TIME_H + typedef std::unique_ptr TIMER; +#else + typedef std::unique_ptr TIMER; +#endif + class client_impl { - + protected: enum con_state { @@ -61,33 +45,33 @@ namespace sio con_closing, con_closed }; - - client_impl(); - + + client_impl(ProtocolVersion version); + ~client_impl(); - + //set listeners and event bindings. #define SYNTHESIS_SETTER(__TYPE__,__FIELD__) \ void set_##__FIELD__(__TYPE__ const& l) \ { m_##__FIELD__ = l;} - + SYNTHESIS_SETTER(client::con_listener,open_listener) - + SYNTHESIS_SETTER(client::con_listener,fail_listener) SYNTHESIS_SETTER(client::reconnect_listener,reconnect_listener) SYNTHESIS_SETTER(client::con_listener,reconnecting_listener) - + SYNTHESIS_SETTER(client::close_listener,close_listener) - + SYNTHESIS_SETTER(client::socket_listener,socket_open_listener) - + SYNTHESIS_SETTER(client::socket_listener,socket_close_listener) - + #undef SYNTHESIS_SETTER - - + + void clear_con_listeners() { m_open_listener = nullptr; @@ -96,26 +80,26 @@ namespace sio m_reconnect_listener = nullptr; m_reconnecting_listener = nullptr; } - + void clear_socket_listeners() { m_socket_open_listener = nullptr; m_socket_close_listener = nullptr; } - + // Client Functions - such as send, etc. void connect(const std::string& uri, const std::map& queryString, const std::map& httpExtraHeaders, const message::ptr& auth); - + sio::socket::ptr const& socket(const std::string& nsp); - + // Closes the connection void close(); - + void sync_close(); - + bool opened() const { return m_con_state == con_opened; } - + std::string const& get_sessionid() const { return m_sid; } void set_reconnect_attempts(unsigned attempts) {m_reconn_attempts = attempts;} @@ -129,44 +113,51 @@ namespace sio void set_logs_quiet(); void set_logs_verbose(); - - void set_proxy_basic_auth(const std::string& uri, const std::string& username, const std::string& password); + + // void set_proxy_basic_auth(const std::string& uri, const std::string& username, const std::string& password); protected: void send(packet& p); - + void remove_socket(std::string const& nsp); - +#ifndef _SAL_TIME_H asio::io_service& get_io_service(); - +#endif void on_socket_closed(std::string const& nsp); - + void on_socket_opened(std::string const& nsp); - + private: void run_loop(); void connect_impl(const std::string& uri, const std::string& query); void close_impl(close::status::value const& code,std::string const& reason); - + + // Forceful, synchronous teardown of the active connection used at + // destruction time. Unlike close_impl (a graceful close that defers + // transport shutdown until the close handshake completes), this calls + // connection::terminate() so the socket is unregistered from the socket + // queue and on_close fires immediately — before the endpoint is freed. + void force_close_impl(); + void send_impl(std::shared_ptr const& payload_ptr,frame::opcode::value opcode); - - void ping(const asio::error_code& ec); - - void timeout_ping(const asio::error_code& ec); - void timeout_reconnect(asio::error_code const& ec); + void timeout_send(const std::error_code& ec); + + void timeout_wait(const std::error_code& ec); + + void timeout_reconnect(std::error_code const& ec); unsigned next_delay() const; socket::ptr get_socket_locked(std::string const& nsp); - + void sockets_invoke_void(void (sio::socket::*fn)(void)); - + void on_decode(packet const& pack); void on_encode(bool isBinary,shared_ptr const& payload); - + //websocket callbacks void on_fail(connection_hdl con); @@ -181,58 +172,73 @@ namespace sio void on_ping(); + void on_pong(); + void reset_states(); void clear_timers(); - void update_ping_timeout_timer(); - - #if SIO_TLS + void update_timeout_timer(); + + void update_send_timer(); + +#if SIO_TLS typedef websocketpp::lib::shared_ptr context_ptr; - + context_ptr on_tls_init(connection_hdl con); - #endif - +#endif + // Percent encode query string std::string encode_query_string(const std::string &query); // Connection pointer for client functions. connection_hdl m_con; client_type m_client; + // Strong reference to the active connection, captured at on_open. + // m_con (a weak connection_hdl) is reset by on_close, which + // runs during teardown BEFORE the forceful close — leaving us unable to + // reach the still-registered connection. This strong ref survives that + // reset so force_close_impl can always terminate/unregister it. + // Declared after m_client so it is destroyed first (connection before + // endpoint). + client_type::connection_ptr m_con_strong; // Socket.IO server settings std::string m_sid; std::string m_base_url; std::string m_query_string; std::map m_http_headers; message::ptr m_auth; - std::string m_proxy_base_url; - std::string m_proxy_basic_username; - std::string m_proxy_basic_password; + // std::string m_proxy_base_url; + // std::string m_proxy_basic_username; + // std::string m_proxy_basic_password; unsigned int m_ping_interval; unsigned int m_ping_timeout; - + std::unique_ptr m_network_thread; - + packet_manager m_packet_mgr; - - std::unique_ptr m_ping_timeout_timer; + typedef void (sio::client_impl::*func_ptr)(const std::error_code&) ; + void update_timer(TIMER &timer, int timeout, func_ptr name); + void reset_timer(TIMER &timer); + std::recursive_mutex m_timer_mutex; + TIMER m_send_timer; + TIMER m_timeout_timer; + TIMER m_reconn_timer; - std::unique_ptr m_reconn_timer; - con_state m_con_state; - + client::con_listener m_open_listener; client::con_listener m_fail_listener; client::con_listener m_reconnecting_listener; client::reconnect_listener m_reconnect_listener; client::close_listener m_close_listener; - + client::socket_listener m_socket_open_listener; client::socket_listener m_socket_close_listener; - + std::map m_sockets; - + std::mutex m_socket_mutex; unsigned m_reconn_delay; @@ -243,11 +249,13 @@ namespace sio unsigned m_reconn_made; + ProtocolVersion m_protocol_version; + std::atomic m_abort_retries { false }; friend class sio::client; friend class sio::socket; + }; } #endif // SIO_CLIENT_IMPL_H - diff --git a/src/internal/sio_packet.cpp b/src/internal/sio_packet.cpp old mode 100755 new mode 100644 index 4b810987..ab66eb1f --- a/src/internal/sio_packet.cpp +++ b/src/internal/sio_packet.cpp @@ -170,8 +170,10 @@ namespace sio { if(it->name.IsString()) { - string key(it->name.GetString(),it->name.GetStringLength()); - static_cast(ptr.get())->get_map()[key] = from_json(it->value,buffers); + message::ptr tmp = from_json(it->value, buffers); + if (!tmp) continue; // will cause crash in READ_SIO_* w/o check + string key(it->name.GetString(), it->name.GetStringLength()); + static_cast(ptr.get())->get_map()[key] = tmp; } } return ptr; diff --git a/src/internal/sio_packet.h b/src/internal/sio_packet.h old mode 100755 new mode 100644 diff --git a/src/sio_client.cpp b/src/sio_client.cpp index b3fa7765..29cb2787 100644 --- a/src/sio_client.cpp +++ b/src/sio_client.cpp @@ -12,8 +12,8 @@ using std::stringstream; namespace sio { - client::client(): - m_impl(new client_impl()) + client::client(ProtocolVersion version): + m_impl(new client_impl(version)) { } @@ -67,10 +67,10 @@ namespace sio m_impl->clear_socket_listeners(); } - void client::set_proxy_basic_auth(const std::string& uri, const std::string& username, const std::string& password) - { - m_impl->set_proxy_basic_auth(uri, username, password); - } + // void client::set_proxy_basic_auth(const std::string& uri, const std::string& username, const std::string& password) + // { + // m_impl->set_proxy_basic_auth(uri, username, password); + // } void client::connect(const std::string& uri) { diff --git a/src/sio_client.h b/src/sio_client.h index 821164aa..c3bbeac1 100644 --- a/src/sio_client.h +++ b/src/sio_client.h @@ -14,6 +14,11 @@ namespace sio { class client_impl; + + enum ProtocolVersion { + ProtocolVersion3 = 3, + ProtocolVersion4 = 4 + }; class client { public: @@ -31,7 +36,7 @@ namespace sio typedef std::function socket_listener; - client(); + client(ProtocolVersion version); ~client(); //set listeners and event bindings. @@ -87,7 +92,7 @@ namespace sio void sync_close(); - void set_proxy_basic_auth(const std::string& uri, const std::string& username, const std::string& password); + // void set_proxy_basic_auth(const std::string& uri, const std::string& username, const std::string& password); bool opened() const; diff --git a/src/sio_message.h b/src/sio_message.h old mode 100755 new mode 100644 index 6ad0d9b9..2801699a --- a/src/sio_message.h +++ b/src/sio_message.h @@ -204,7 +204,7 @@ namespace sio } string_message(std::string&& v) - :message(flag_string),_v(move(v)) + :message(flag_string),_v(std::move(v)) { } public: @@ -215,7 +215,7 @@ namespace sio static message::ptr create(std::string&& v) { - return ptr(new string_message(move(v))); + return ptr(new string_message(std::move(v))); } std::string const& get_string() const @@ -269,7 +269,7 @@ namespace sio void push(std::string&& text) { - _v.push_back(string_message::create(move(text))); + _v.push_back(string_message::create(std::move(text))); } void push(std::shared_ptr const& binary) @@ -296,7 +296,7 @@ namespace sio void insert(size_t pos,std::string&& text) { - _v.insert(_v.begin()+pos, string_message::create(move(text))); + _v.insert(_v.begin()+pos, string_message::create(std::move(text))); } void insert(size_t pos,std::shared_ptr const& binary) @@ -361,7 +361,7 @@ namespace sio void insert(const std::string & key,std::string&& text) { - _v[key] = string_message::create(move(text)); + _v[key] = string_message::create(std::move(text)); } void insert(const std::string & key,std::shared_ptr const& binary) @@ -461,7 +461,7 @@ namespace sio list(std::string&& text) { - m_vector.push_back(string_message::create(move(text))); + m_vector.push_back(string_message::create(std::move(text))); } list(std::shared_ptr const& binary) @@ -489,7 +489,7 @@ namespace sio void push(std::string&& text) { - m_vector.push_back(string_message::create(move(text))); + m_vector.push_back(string_message::create(std::move(text))); } void push(std::shared_ptr const& binary) @@ -516,7 +516,7 @@ namespace sio void insert(size_t pos,std::string&& text) { - m_vector.insert(m_vector.begin()+pos, string_message::create(move(text))); + m_vector.insert(m_vector.begin()+pos, string_message::create(std::move(text))); } void insert(size_t pos,std::shared_ptr const& binary) diff --git a/src/sio_socket.cpp b/src/sio_socket.cpp index 34be59bf..54aed14a 100644 --- a/src/sio_socket.cpp +++ b/src/sio_socket.cpp @@ -1,18 +1,17 @@ #include "sio_socket.h" #include "internal/sio_packet.h" #include "internal/sio_client_impl.h" -#include -#include #include #include #include #include -#if DEBUG || _DEBUG -#define LOG(x) std::cout << x -#else -#define LOG(x) -#endif +#include "SAL/Log/Log.h" + +namespace { + LOGGER_NAME("socketio.socket") +} + #define NULL_GUARD(_x_) \ if(_x_ == NULL) return @@ -159,7 +158,7 @@ namespace sio void ack(int msgId,string const& name,message::list const& ack_message); - void timeout_connection(const asio::error_code &ec); + void timeout_connection(const std::error_code &ec); void send_connect(); @@ -181,7 +180,7 @@ namespace sio error_listener m_error_listener; - std::unique_ptr m_connection_timer; + std::unique_ptr m_connection_timer; std::queue m_packet_queue; @@ -273,10 +272,12 @@ namespace sio NULL_GUARD(m_client); packet p(packet::type_connect, m_nsp, m_auth); m_client->send(p); - m_connection_timer.reset(new asio::steady_timer(m_client->get_io_service())); - asio::error_code ec; - m_connection_timer->expires_from_now(std::chrono::milliseconds(20000), ec); - m_connection_timer->async_wait(std::bind(&socket::impl::timeout_connection,this, std::placeholders::_1)); + m_connection_timer.reset(); + m_connection_timer.reset(SAL::timer_cb::set_timer(20000, std::bind(&socket::impl::timeout_connection,this, std::placeholders::_1))); + // m_connection_timer.reset(new asio::steady_timer(m_client->get_io_service())); + // asio::error_code ec; + // m_connection_timer->expires_from_now(std::chrono::milliseconds(20000), ec); + // m_connection_timer->async_wait(std::bind(&socket::impl::timeout_connection,this, std::placeholders::_1)); } void socket::impl::close() @@ -286,15 +287,18 @@ namespace sio { packet p(packet::type_disconnect,m_nsp); send_packet(p); - - if(!m_connection_timer) - { - m_connection_timer.reset(new asio::steady_timer(m_client->get_io_service())); - } - asio::error_code ec; - m_connection_timer->expires_from_now(std::chrono::milliseconds(3000), ec); - m_connection_timer->async_wait(std::bind(&socket::impl::on_close, this)); - } + } + // Skip the delay by calling on_close immediately. Even though it is very aggressive, the server should be able to handle that + // m_connection_timer.reset(SAL::timer_cb::set_timer(3000, std::bind(&socket::impl::on_close, this))); + on_close(); + + // if(!m_connection_timer) + // { + // m_connection_timer.reset(new asio::steady_timer(m_client->get_io_service())); + // } + // asio::error_code ec; + // m_connection_timer->expires_from_now(std::chrono::milliseconds(3000), ec); + // m_connection_timer->async_wait(std::bind(&socket::impl::on_close, this)); } void socket::impl::on_connected() @@ -348,7 +352,8 @@ namespace sio void socket::impl::on_open() { - send_connect(); + // [2023-06-27 EB]: This is not strictly necessary and causes double connect send from router + // send_connect(); } void socket::impl::on_disconnect() @@ -374,21 +379,21 @@ namespace sio // Connect open case packet::type_connect: { - LOG("Received Message type (Connect)"<on_connected(); break; } case packet::type_disconnect: { - LOG("Received Message type (Disconnect)"<on_close(); break; } case packet::type_event: case packet::type_binary_event: { - LOG("Received Message type (Event)"<get_flag() == message::flag_array) { @@ -401,7 +406,9 @@ namespace sio { mlist.push(array_ptr->get_vector()[i]); } - this->on_socketio_event(p.get_nsp(), p.get_pack_id(),name_ptr->get_string(), std::move(mlist)); + SAL_FUNC_DEBUG("Received SocketIo Event Nsp=%s, PackId=%u, NamePtr=%s", + p.get_nsp().c_str(), p.get_pack_id(), name_ptr->get_string().c_str()); + this->on_socketio_event(p.get_nsp(), (int)p.get_pack_id(),name_ptr->get_string(), std::move(mlist)); } } @@ -411,7 +418,7 @@ namespace sio case packet::type_ack: case packet::type_binary_ack: { - LOG("Received Message type (ACK)"<get_flag() == message::flag_array) { @@ -427,7 +434,7 @@ namespace sio // Error case packet::type_error: { - LOG("Received Message type (ERROR)"<on_socketio_error(p.get_message()); break; } @@ -475,7 +482,7 @@ namespace sio if(m_error_listener)m_error_listener(err_message); } - void socket::impl::timeout_connection(const asio::error_code &ec) + void socket::impl::timeout_connection(const std::error_code &ec) { NULL_GUARD(m_client); if(ec) @@ -483,7 +490,7 @@ namespace sio return; } m_connection_timer.reset(); - LOG("Connection timeout,close socket."<on_close(); }