#include "TCP_Client.h" #include "Core/Statistics/Frequency_Limit.h" #include "Socket_p.h" #include #include #ifdef _WIN32 #include #include #else #include #include #include #include #endif namespace Psc::socket { static bool fd_writable_now(int fd) { #ifdef _WIN32 fd_set wfds; FD_ZERO(&wfds); FD_SET((SOCKET)fd, &wfds); timeval tv{}; tv.tv_sec = 0; tv.tv_usec = 0; int r = select(0, nullptr, &wfds, nullptr, &tv); return r > 0 && FD_ISSET((SOCKET)fd, &wfds); #else fd_set wfds; FD_ZERO(&wfds); FD_SET(fd, &wfds); timeval tv{}; tv.tv_sec = 0; tv.tv_usec = 0; int r = select(fd + 1, nullptr, &wfds, nullptr, &tv); return r > 0 && FD_ISSET(fd, &wfds); #endif } void TCP_Client::create() { send_buffer.init(max_buffer_size); auto rsf = socket::create_socket_fd(Socket_Type::TCP); if (!rsf) { last_err = rsf.error(); socket_logger->c_debug("Tcp_Client::create", {}, LOG_POS + last_err.to_string()); return; } socket_fd = rsf.value(); set_state(Not_Set_Field); auto r = socket::set_reuse(socket_fd, true) .and_then([this]() { return socket::set_block(socket_fd, false); }) .and_then([this]() { return socket::TCP::set_keep_alive(socket_fd, true); }) .and_then([this]() { return socket::TCP::set_no_delay(socket_fd, true); }) .and_then([this]() { return socket::set_reuse_addr(socket_fd, true); }) ; if (!r) { last_err = r.error(); socket_logger->error("Tcp_Client::create", {}, Psc::to_string(last_err) + " " + LOG_POS); (void)socket::close(socket_fd); socket_fd = -1; set_state(Not_Created); return; } // 这里最好检查 dest_address 是否已设置(你自己加一个标志位也行) set_state(Reconnecting); next_reconnect_tp = std::chrono::steady_clock::now(); tick(); // 立即尝试一次 } void TCP_Client::schedule_recreate(Enum_Err e) { last_err = e; (void)socket::close(socket_fd); set_state(Not_Created); next_reconnect_tp = std::chrono::steady_clock::now() + std::chrono::milliseconds(reconnect_backoff_ms); } bool TCP_Client::start_connect() { if (socket_fd == -1) return false; auto r = socket::connect(socket_fd, dest_address); if (!r) { last_err = r.error(); if (last_err.nerr == NetError::in_progress) { set_state(Connecting); return false; // 还没完成 } if (last_err.nerr == NetError::already_connected) { set_state(Wait_Check); return true; } socket_logger->c_debug({}, {}, "连接失败:"+ VAR_STR_2(last_err, *this) + "\n"); // 其它错误:进入退避重连 schedule_recreate(last_err); return false; } return false; } bool TCP_Client::check_writeable() { return true; } void TCP_Client::tick() { if (socket_fd == -1) return; auto st = state.load(); auto now = std::chrono::steady_clock::now(); if (st == Not_Created) { create(); return; } if (st == Reconnecting) { if (now >= next_reconnect_tp) { (void) start_connect(); } return; } if (st == Connecting || st == Wait_Check) { set_state(Connected); return; } if (st == Connected) { flush_send_buffer(); return; } } std::string TCP_Client::read() { if (socket_fd == -1) return ""; // 如果还没连上,先推进连接状态 tick(); if (state.load(std::memory_order_relaxed) != Connected) return ""; auto ret = socket::recv_all(socket_fd); if (!ret) { Enum_Err e = ret.error(); // 下面这几个名字按你 NetError 实际枚举改: // - would_block: 非阻塞没数据,不算错 // - in_progress/not_connected: 连接没完成 if (e.nerr == NetError::would_block) return ""; static Frequency_Limit drop_log_fl; if (drop_log_fl.test()) { socket_logger->c_debug("Tcp_Client::read recv失败", {}, to_string() + " " + Ret_To_String(ret) + " " + LOG_POS); } // 断线类错误:进入退避重连(必要时你也可以 close+重建fd) schedule_recreate(e); return ""; } auto r = ret.value(); return r; } void TCP_Client::flush_send_buffer() { if (socket_fd == -1) return; if (state.load(std::memory_order_relaxed) != Connected) return; if (send_buffer.empty()) return; constexpr uint32_t patch_len = 1500; std::string buf; buf.resize(patch_len); while (true) { uint32_t n = send_buffer.peek_best_effort(buf.data(), patch_len); if (n == 0) break; auto r = socket::send(socket_fd, buf.data(), n); if (!r) { last_err = r.error(); if (last_err.nerr == NetError::would_block) break; if (last_err.nerr == NetError::connection_aborted) { schedule_recreate(last_err); } else if (last_err.nerr == NetError::connection_reset) { schedule_recreate(last_err); } else { socket_logger->c_debug("Tcp_Client::send", {}, "触发重连 " + VAR_STR_2(last_err, *this)); schedule_recreate(last_err); } return; } send_buffer.skip_best_effort((uint32_t)r.value()); } } void TCP_Client::send(const std::string& data) { if (data.empty()) return; (void)send_buffer.write_best_effort((const uint8_t*)data.data(), (uint32_t)data.size()); if (state != Connected) return; tick(); flush_send_buffer(); } std::string TCP_Client::to_string() { return VAR_STR_4(socket_fd, address, dest_address, state); } void TCP_Client::close() { if (socket_fd != -1) { auto r = socket::close(socket_fd); (void)r; socket_fd = -1; set_state(Not_Created); } } void TCP_Client::set_state(State s) { auto old_state = state.load(); std::ostringstream oss; oss << to_string() << "状态变化:" << Psc::to_string(s) << "===>" << Psc::to_string(old_state) << std::endl; socket_logger->debug({}, {}, oss.str()); if (on_state_change) on_state_change(state, s); state = s; } TCP_Client::~TCP_Client() { TCP_Client::close(); } } // namespace Psc::socket