Try Qt socket for SBB.

Mysticial · Mysticial · commit f785aa3f05b8 · 2025-07-06T03:19:34.000-07:00
diff --git a/Common/Cpp/Sockets/ClientSocket.h b/Common/Cpp/Sockets/ClientSocket.h
@@ -10,9 +10,11 @@
 
 
 #ifdef _WIN32
-#include "ClientSocket_WinSocket.h"
+//#include "ClientSocket_WinSocket.h"
+#include "ClientSocket_Qt.h"
 namespace PokemonAutomation{
-    using ClientSocket = ClientSocket_WinSocket;
+//    using ClientSocket = ClientSocket_WinSocket;
+    using ClientSocket = ClientSocket_Qt;
 }
 #else
 #include "ClientSocket_POSIX.h"
diff --git a/Common/Cpp/Sockets/ClientSocket_Qt.h b/Common/Cpp/Sockets/ClientSocket_Qt.h
@@ -0,0 +1,179 @@
+/*  Client Socket (Qt)
+ *
+ *  From: https://github.com/PokemonAutomation/
+ *
+ */
+
+#ifndef PokemonAutomation_ClientSocket_Qt_H
+#define PokemonAutomation_ClientSocket_Qt_H
+
+#include <mutex>
+#include <condition_variable>
+#include <QThread>
+#include <QTcpSocket>
+//#include "Common/Cpp/Concurrency/SpinPause.h"
+#include "AbstractClientSocket.h"
+
+//#include <iostream>
+//using std::cout;
+//using std::endl;
+
+namespace PokemonAutomation{
+
+
+
+class ClientSocket_Qt final : public QThread, public AbstractClientSocket{
+    Q_OBJECT
+
+    struct SendData{
+        const void* data;
+        size_t bytes;
+        std::mutex lock;
+        std::condition_variable cv;
+    };
+
+
+public:
+    ClientSocket_Qt()
+        : m_socket(nullptr)
+    {
+//        cout << "ClientSocket_Qt()" << endl;
+        start();
+
+//        cout << "ClientSocket_Qt() - waiting" << endl;
+        std::unique_lock<std::mutex> lg(m_lock);
+        m_cv.wait(lg, [this]{ return m_socket != nullptr; });
+    }
+
+    virtual ~ClientSocket_Qt(){
+//        cout << "~ClientSocket_Qt()" << endl;
+        close();
+    }
+    virtual void close() noexcept override{
+//        cout << "close()" << endl;
+        m_state.store(State::DESTRUCTING, std::memory_order_release);
+        quit();
+        wait();
+    }
+
+    virtual void connect(const std::string& address, uint16_t port) override{
+//        cout << "connect()" << endl;
+        emit internal_connect(address, port);
+    }
+
+    virtual size_t blocking_send(const void* data, size_t bytes) override{
+//        cout << "blocking_send() - start: " << std::string((const char*)data, bytes) << endl;
+
+        SendData send_data;
+        send_data.data = data;
+        send_data.bytes = bytes;
+
+        emit send(&send_data);
+
+        std::unique_lock<std::mutex> lg(send_data.lock);
+        send_data.cv.wait(lg, [&]{
+            return send_data.data == nullptr || m_socket == nullptr;
+        });
+
+//        cout << "blocking_send() - end: " << std::string((const char*)data, bytes) << endl;
+        return send_data.bytes;
+    }
+
+
+signals:
+    void internal_connect(const std::string& address, uint16_t port);
+    void send(void* data);
+
+private:
+    virtual void run() override{
+        QTcpSocket socket;
+
+        QThread::connect(
+            &socket, &QTcpSocket::connected,
+            &socket, [this]{
+//                cout << "connected()" << endl;
+                m_state.store(State::CONNECTED, std::memory_order_release);
+                m_listeners.run_method_unique(&Listener::on_connect_finished, "");
+            }
+        );
+        QThread::connect(
+            &socket, &QTcpSocket::readyRead,
+            &socket, [this]{
+//                cout << "readyRead()" << endl;
+                constexpr size_t BUFFER_SIZE = 4096;
+                char buffer[BUFFER_SIZE];
+                qint64 bytes = m_socket->read(buffer, BUFFER_SIZE);
+                if (bytes > 0){
+//                    cout << "Received: " << std::string(buffer, bytes) << endl;
+                    m_listeners.run_method_unique(&Listener::on_receive_data, buffer, bytes);
+                }
+            }
+        );
+        QThread::connect(
+            this, &ClientSocket_Qt::internal_connect,
+            &socket, [this](const std::string& address, uint16_t port){
+                m_state.store(State::CONNECTING, std::memory_order_release);
+                m_socket->connectToHost(QHostAddress(QString::fromStdString(address)), port);
+            }
+        );
+        QThread::connect(
+            this, &ClientSocket_Qt::send,
+            &socket, [this](void* params){
+//                cout << "internal_send() - enter " << endl;
+
+                SendData& data = *(SendData*)params;
+                size_t sent = 0;
+
+                size_t bytes = data.bytes;
+
+                const char* ptr = (const char*)data.data;
+                while (bytes > 0 && m_socket->state() == QAbstractSocket::ConnectedState){
+                    qint64 current_sent = m_socket->write((const char*)ptr, bytes);
+                    if (current_sent <= 0){
+                        break;
+                    }
+                    sent += current_sent;
+                    ptr += current_sent;
+                    bytes -= current_sent;
+                    data.bytes = sent;
+                }
+
+                m_socket->flush();
+
+                std::lock_guard<std::mutex> lg(data.lock);
+                data.data = nullptr;
+                data.bytes = sent;
+                data.cv.notify_all();
+
+//                cout << "internal_send() - exit " << endl;
+            }
+        );
+
+
+        {
+            std::lock_guard<std::mutex> lg(m_lock);
+            m_socket = &socket;
+        }
+        m_cv.notify_all();
+
+        exec();
+
+        {
+            std::lock_guard<std::mutex> lg(m_lock);
+            m_socket = nullptr;
+        }
+        m_cv.notify_all();
+    }
+
+private:
+    std::mutex m_lock;
+    std::condition_variable m_cv;
+    QTcpSocket* m_socket;
+};
+
+
+
+
+
+}
+#endif
diff --git a/SerialPrograms/CMakeLists.txt b/SerialPrograms/CMakeLists.txt
@@ -193,6 +193,7 @@ file(GLOB MAIN_SOURCES
     ../Common/Cpp/Sockets/ClientSocket.cpp
     ../Common/Cpp/Sockets/ClientSocket.h
     ../Common/Cpp/Sockets/ClientSocket_POSIX.h
+    ../Common/Cpp/Sockets/ClientSocket_Qt.h
     ../Common/Cpp/Sockets/ClientSocket_WinSocket.h
     ../Common/Cpp/StreamConverters.cpp
     ../Common/Cpp/StreamConverters.h
diff --git a/SerialPrograms/Source/NintendoSwitch/Controllers/SysbotBase/SysbotBase_Connection.cpp b/SerialPrograms/Source/NintendoSwitch/Controllers/SysbotBase/SysbotBase_Connection.cpp
@@ -49,7 +49,8 @@ TcpSysbotBase_Connection::TcpSysbotBase_Connection(
 )
     : m_logger(logger)
     , m_supports_command_queue(false)
-    , m_last_receive(WallClock::min())
+    , m_last_ping_send(WallClock::min())
+    , m_last_ping_receive(WallClock::min())
 {
     QHostAddress address;
     int port;
@@ -122,7 +123,7 @@ std::string pretty_print(uint64_t x){
 
 void TcpSysbotBase_Connection::thread_loop(){
     std::unique_lock<std::mutex> lg(m_lock);
-    WallClock send_time = current_time();
+    m_last_ping_send = current_time();
     while (true){
         ClientSocket::State state = m_socket.state();
         if (state == ClientSocket::State::DESTRUCTING || state == ClientSocket::State::NOT_RUNNING){
@@ -131,30 +132,14 @@ void TcpSysbotBase_Connection::thread_loop(){
 
         WallClock now = current_time();
 
-        if (m_last_receive == WallClock::min()){
-            send_time = now;
-            write_data("getVersion\r\n");
-        }else if (send_time < m_last_receive && now - m_last_receive < std::chrono::seconds(1)){
-            std::chrono::microseconds latency = std::chrono::duration_cast<std::chrono::microseconds>(m_last_receive - send_time);
-            std::string str = "Response Time: " + pretty_print(latency.count()) + " ms";
-            if (latency < 10ms){
-                set_status_line1(str, COLOR_BLUE);
-            }else if (latency < 50ms){
-                set_status_line1(str, COLOR_DARKGREEN);
-            }else{
-                set_status_line1(str, COLOR_ORANGE);
-            }
-            send_time = current_time();
-            write_data("getVersion\r\n");
-//            cout << std::chrono::duration_cast<std::chrono::microseconds>(current_time() - send_time) << endl;
-        }else{
-            std::chrono::milliseconds time_since = std::chrono::duration_cast<std::chrono::milliseconds>(now - m_last_receive);
+        std::chrono::milliseconds time_since = std::chrono::duration_cast<std::chrono::milliseconds>(now - m_last_ping_receive);
+        if (time_since > std::chrono::seconds(5)){
             std::string str = "Last Ack: " + pretty_print(time_since.count()) + " seconds ago";
             set_status_line1(str, COLOR_RED);
-            send_time = current_time();
-            write_data("getVersion\r\n");
-//            cout << std::chrono::duration_cast<std::chrono::microseconds>(current_time() - send_time) << endl;
         }
+
+        m_last_ping_send = current_time();
+        write_data("getVersion\r\n");
         m_cv.wait_for(lg, std::chrono::seconds(1));
     }
 }
@@ -185,11 +170,6 @@ void TcpSysbotBase_Connection::on_receive_data(const void* data, size_t bytes){
 //    cout << "on_receive_data(): " << std::string((const char*)data, bytes - 2) << endl;
 
     WallClock now = current_time();
-    {
-        std::lock_guard<std::mutex> lg(m_lock);
-        m_last_receive = now;
-    }
-
 
     try{
         const char* ptr = (const char*)data;
@@ -206,7 +186,8 @@ void TcpSysbotBase_Connection::on_receive_data(const void* data, size_t bytes){
                 std::string(
                     m_receive_buffer.begin(),
                     m_receive_buffer.end()
-                )
+                ),
+                now
             );
             m_receive_buffer.clear();
         }
@@ -215,7 +196,7 @@ void TcpSysbotBase_Connection::on_receive_data(const void* data, size_t bytes){
 
     }catch (...){}
 }
-void TcpSysbotBase_Connection::process_message(const std::string& message){
+void TcpSysbotBase_Connection::process_message(const std::string& message, WallClock timestamp){
 //    cout << "sys-botbase Response: " << message << endl;
 
     m_listeners.run_method_unique(&Listener::on_message, message);
@@ -229,6 +210,20 @@ void TcpSysbotBase_Connection::process_message(const std::string& message){
         set_status_line0("sys-botbase: Version " + str, COLOR_BLUE);
 
         std::lock_guard<std::mutex> lg(m_lock);
+        m_last_ping_receive = timestamp;
+
+        if (m_last_ping_send != WallClock::min()){
+            std::chrono::microseconds latency = std::chrono::duration_cast<std::chrono::microseconds>(timestamp - m_last_ping_send);
+            std::string text = "Response Time: " + pretty_print(latency.count()) + " ms";
+            if (latency < 10ms){
+                set_status_line1(text, COLOR_BLUE);
+            }else if (latency < 50ms){
+                set_status_line1(text, COLOR_DARKGREEN);
+            }else{
+                set_status_line1(text, COLOR_ORANGE);
+            }
+        }
+
         if (!m_thread.joinable()){
             set_mode(str);
         }
diff --git a/SerialPrograms/Source/NintendoSwitch/Controllers/SysbotBase/SysbotBase_Connection.h b/SerialPrograms/Source/NintendoSwitch/Controllers/SysbotBase/SysbotBase_Connection.h
@@ -55,7 +55,7 @@ class TcpSysbotBase_Connection : public ControllerConnection, private ClientSock
     virtual void on_connect_finished(const std::string& error_message) override;
     virtual void on_receive_data(const void* data, size_t bytes) override;
 
-    void process_message(const std::string& message);
+    void process_message(const std::string& message, WallClock timestamp);
     void set_mode(const std::string& sbb_version);
 
 private:
@@ -66,7 +66,8 @@ class TcpSysbotBase_Connection : public ControllerConnection, private ClientSock
 
     std::string m_connecting_message;
 //    std::string m_version;
-    WallClock m_last_receive;
+    WallClock m_last_ping_send;
+    WallClock m_last_ping_receive;
     std::deque<char> m_receive_buffer;
 
     SpinLock m_send_lock;
