socket: Unified Soket thread management.

.vscode/settings.json

@@ -24,8 +24,8 @@
         "-Wno-dev"
     ],
     "cmake.environment": {
-        "QT_LIB_ROOT": "C:/Qt/6.8.3",
+        "QT_LIB_ROOT": "D:/Dev/Qt6/msvc2022_64",
-        "PATH": "${env:PATH};C:/Qt/6.8.3/bin"
+        "PATH": "${env:PATH};D:/Dev/Qt6/msvc2022_64/bin"
     },
     // "cmake.environment": {
     //     "QT_LIB_ROOT": "C:/Qt/Qt5.14.2/5.14.2/msvc2017_64",
@@ -159,6 +159,7 @@
         "qreadwritelock": "cpp",
         "qdatastream": "cpp",
         "qhostaddress": "cpp",
-        "qthread": "cpp"
+        "qthread": "cpp",
+        "qobject": "cpp"
     }
 }

ClientCore/ClientCore.cpp

@@ -49,6 +49,7 @@ void ClientCore::onReadyRead()
 void ClientCore::onDisconnected()
 {
     qCritical() << QString("client %1 disconnected...").arg(remoteID_);
+    emit sigDisconnect();
 }
 
 void ClientCore::UseFrame(QSharedPointer<FrameBuffer> frame)
@@ -172,3 +173,32 @@ QString ClientCore::GetOwnID()
 {
     return ownID_;
 }
+
+SocketWorker::SocketWorker(ClientCore* core, QObject* parent) : QThread(parent), core_(core)
+{
+    connect(core_, &ClientCore::sigDisconnect, this, [this]() {
+        emit disconnected();
+        thread()->quit();
+    });
+}
+
+SocketWorker::~SocketWorker()
+{
+}
+
+void SocketWorker::SetConnectInfo(const QString& ip, qint16 port)
+{
+    ip_ = ip;
+    port_ = port;
+}
+
+void SocketWorker::run()
+{
+    emit connecting();
+    if (!core_->Connect(ip_, port_)) {
+        emit conFailed();
+        return;
+    }
+    emit conSuccess();
+    exec();
+}

ClientCore/ClientCore.h

@@ -8,9 +8,12 @@
 #include <LocalFile.h>
 #include <Protocol.h>
 #include <QDataStream>
+#include <QFuture>
 #include <QHostAddress>
 #include <QMutex>
 #include <QMutexLocker>
+#include <QPromise>
+#include <QQueue>
 #include <QTcpSocket>
 #include <QThread>
 #include <array>
@@ -29,13 +32,41 @@ public:
     bool Send(QSharedPointer<FrameBuffer> frame);
     bool Send(const char* data, qint64 len);
     template <typename T> bool Send(const T& info, FrameBufferType type, const QString& tid)
+    {
+        auto f = GetBuffer<T>(info, type, tid);
+        return Send(f);
+    }
+    template <typename T> QSharedPointer<FrameBuffer> GetBuffer(const T& info, FrameBufferType type, const QString& tid)
     {
         auto f = QSharedPointer<FrameBuffer>::create();
         f->tid = tid;
         f->data = infoPack<T>(info);
         f->type = type;
-        return Send(f);
+        return f;
     }
+    template <typename Callable> static bool asyncInvoke(QObject* context, Callable&& func)
+    {
+        auto promise = QSharedPointer<QPromise<bool>>::create();
+        QFuture<bool> future = promise->future();
+
+        QMetaObject::invokeMethod(
+            context,
+            [func = std::forward<Callable>(func), promise]() mutable {
+                try {
+                    promise->addResult(func());
+                } catch (...) {
+                    promise->addResult(false);
+                }
+                promise->finish();
+            },
+            Qt::QueuedConnection);
+
+        future.waitForFinished();
+        return future.result();
+    }
+
+signals:
+    void sigDisconnect();
 
 private:
     void onReadyRead();
@@ -71,4 +102,30 @@ public:
     std::array<std::function<void(QSharedPointer<FrameBuffer>)>, 256> frameCall_;
 };
 
+class SocketWorker : public QThread
+{
+    Q_OBJECT
+
+public:
+    SocketWorker(ClientCore* core, QObject* parent = nullptr);
+    ~SocketWorker();
+
+public:
+    void SetConnectInfo(const QString& ip, qint16 port);
+
+protected:
+    void run() override;
+
+signals:
+    void conSuccess();
+    void connecting();
+    void conFailed();
+    void disconnected();
+
+private:
+    QString ip_;
+    qint16 port_;
+    ClientCore* core_;
+};
+
 #endif   // CLIENTCORE_H

ClientCore/FileTrans.cpp

@@ -114,7 +114,7 @@ void FileTrans::RegisterFrameCall()
 void FileTrans::fbtReqSend(QSharedPointer<FrameBuffer> frame)
 {
     InfoMsg info = infoUnpack<InfoMsg>(frame->data);
-    qInfo() << QString(tr("%1 req send: %2 to %3")).arg(frame->fid).arg(info.fromPath).arg(info.toPath);
+    qInfo() << QString(tr("%1 req send: %2 to %3")).arg(frame->fid).arg(info.fromPath, info.toPath);
 
     // judge is same client's same file.
 
@@ -132,7 +132,7 @@ void FileTrans::fbtReqSend(QSharedPointer<FrameBuffer> frame)
         info.msg = QString(tr("open file failed: %1")).arg(newerPath);
         qCritical() << info.msg;
         if (!clientCore_->Send<InfoMsg>(info, FBT_CLI_CANOT_SEND, frame->fid)) {
-            qCritical() << QString(tr("open recv file:%2 failed, and reply %2 failed.")).arg(info.msg).arg(frame->fid);
+            qCritical() << QString(tr("open recv file:%2 failed, and reply %2 failed.")).arg(info.msg, frame->fid);
             downTask_->file.close();
             return;
         }
@@ -144,8 +144,10 @@ void FileTrans::fbtReqSend(QSharedPointer<FrameBuffer> frame)
     downTask_->permission = info.permissions;
 
     info.msg = QString(tr("open recv file success: %1")).arg(newerPath);
-    if (!clientCore_->Send<InfoMsg>(info, FBT_CLI_CAN_SEND, frame->fid)) {
+    auto f = clientCore_->GetBuffer(info, FBT_CLI_CAN_SEND, frame->fid);
-        qCritical() << QString(tr("open recv file:%2 success, but reply %2 failed.")).arg(info.msg).arg(frame->fid);
+    auto sendRet = ClientCore::asyncInvoke(clientCore_, [this, f]() { return clientCore_->Send(f); });
+    if (!sendRet) {
+        qCritical() << QString(tr("open recv file:%2 success, but reply %2 failed.")).arg(info.msg, frame->fid);
         downTask_->file.close();
         return;
     }
@@ -200,7 +202,7 @@ void FileTrans::fbtCanDown(QSharedPointer<FrameBuffer> frame)
 void FileTrans::fbtCanotDown(QSharedPointer<FrameBuffer> frame)
 {
     InfoMsg info = infoUnpack<InfoMsg>(frame->data);
-    qCritical() << QString(tr("request send file:%1 failed. reason:%2")).arg(info.fromPath).arg(info.msg);
+    qCritical() << QString(tr("request send file:%1 failed. reason:%2")).arg(info.fromPath, info.msg);
 }
 
 void FileTrans::fbtFileBuffer(QSharedPointer<FrameBuffer> frame)
@@ -223,7 +225,7 @@ void FileTrans::fbtFileBuffer(QSharedPointer<FrameBuffer> frame)
 void FileTrans::fbtCanotSend(QSharedPointer<FrameBuffer> frame)
 {
     InfoMsg info = infoUnpack<InfoMsg>(frame->data);
-    qCritical() << QString(tr("request file:%1 failed. reason:%2")).arg(info.fromPath).arg(info.msg);
+    qCritical() << QString(tr("request file:%1 failed. reason:%2")).arg(info.fromPath, info.msg);
     if (sendTask_->file.isOpen()) {
         sendTask_->file.close();
     }
@@ -232,7 +234,7 @@ void FileTrans::fbtCanotSend(QSharedPointer<FrameBuffer> frame)
 void FileTrans::fbtCanSend(QSharedPointer<FrameBuffer> frame)
 {
     InfoMsg info = infoUnpack<InfoMsg>(frame->data);
-    qInfo() << QString(tr("start trans file:%1 to %2")).arg(info.fromPath).arg(frame->fid);
+    qInfo() << QString(tr("start trans file:%1 to %2")).arg(info.fromPath, frame->fid);
     SendFile(sendTask_);
 }
 
@@ -255,6 +257,21 @@ void FileTrans::SendFile(const QSharedPointer<DoTransTask>& task)
     sendThread->run();
 }
 
+QFuture<bool> FileTrans::sendFrameAsync(const QSharedPointer<FrameBuffer>& frame)
+{
+    auto promise = QSharedPointer<QPromise<bool>>::create();
+    QFuture<bool> future = promise->future();
+    QMetaObject::invokeMethod(
+        clientCore_,
+        [this, frame, promise]() mutable {
+            bool ret = clientCore_->Send(frame);
+            promise->addResult(ret);
+            promise->finish();
+        },
+        Qt::QueuedConnection);
+    return future;
+}
+
 SendThread::SendThread(ClientCore* clientCore) : cliCore_(clientCore)
 {
 }
@@ -265,23 +282,36 @@ void SendThread::run()
     auto frame = QSharedPointer<FrameBuffer>::create();
     frame->tid = task_->task.remoteId;
     frame->type = FBT_CLI_FILE_BUFFER;
+    frame->call = [this](QSharedPointer<FrameBuffer> frame) { sendCall(frame); };
 
-    bool suc = true;
+    isSuccess_ = true;
     while (!task_->file.atEnd()) {
         frame->data.resize(CHUNK_BUF_SIZE);
         auto br = task_->file.read(frame->data.data(), CHUNK_BUF_SIZE);
         if (br == -1) {
             qCritical() << QString(tr("read file failed: %1")).arg(task_->file.errorString());
-            suc = false;
+            isSuccess_ = false;
             break;
         }
         frame->data.resize(br);
-        if (!cliCore_->Send(frame)) {
+
+        while (curSendCount_ >= MAX_SEND_TASK) {
+            QThread::msleep(1);
+            // shoule add abort action mark.
+        }
+
+        QMetaObject::invokeMethod(this, [this, frame] {
+            frame->sendRet = cliCore_->Send(frame);
+            if (frame->call) {
+                frame->call(frame);
+            }
+        });
+        ++curSendCount_;
+
+        if (!isSuccess_) {
             qCritical() << QString(tr("send to %1 file failed.")).arg(task_->task.remoteId);
-            suc = false;
             break;
         }
-        task_->tranSize += br;
     }
     task_->file.close();
 }
@@ -290,3 +320,13 @@ void SendThread::setTask(const QSharedPointer<DoTransTask>& task)
 {
     task_ = task;
 }
+
+void SendThread::sendCall(QSharedPointer<FrameBuffer> frame)
+{
+    if (frame->sendRet) {
+        --curSendCount_;
+        task_->tranSize += frame->data.size();
+    } else {
+        isSuccess_ = false;
+    }
+}

ClientCore/FileTrans.h

@@ -5,9 +5,13 @@
 #include <QMap>
 #include <QMutex>
 #include <QVector>
+#include <QFuture>
+#include <QPromise>
 
 #include "ClientCore.h"
 
+constexpr int MAX_SEND_TASK = 10;
+
 struct TransTask {
     bool isUpload{false};
     QString localId;
@@ -43,9 +47,12 @@ public:
 public:
     void run() override;
     void setTask(const QSharedPointer<DoTransTask>& task);
+    void sendCall(QSharedPointer<FrameBuffer> frame);
 
 private:
+    bool isSuccess_{ false };
     ClientCore* cliCore_;
+    quint32 curSendCount_{0};
     QSharedPointer<DoTransTask> task_;
 };
 
@@ -76,6 +83,7 @@ private:
 private:
     void RegisterFrameCall();
     void SendFile(const QSharedPointer<DoTransTask>& task);
+    QFuture<bool> sendFrameAsync(const QSharedPointer<FrameBuffer>& frame);
 
 private:
     QSharedPointer<DoTransTask> downTask_;

ClientCore/RemoteFile.cpp

@@ -14,12 +14,13 @@ void RemoteFile::setClientCore(ClientCore* cliCore)
 bool RemoteFile::GetHome()
 {
     InfoMsg info;
-    return cliCore_->Send<InfoMsg>(info, FBT_CLI_ASK_HOME, cliCore_->GetRemoteID());
+    auto frame = cliCore_->GetBuffer(info, FBT_CLI_ASK_HOME, cliCore_->GetRemoteID());
+    return ClientCore::asyncInvoke(cliCore_, [this, frame]() { return cliCore_->Send(frame); });
 }
 
 bool RemoteFile::GetDirFile(const QString& dir)
 {
     InfoMsg info;
-    info.msg = dir;
+    auto frame = cliCore_->GetBuffer(info, FBT_CLI_ASK_DIRFILE, cliCore_->GetRemoteID());
-    return cliCore_->Send<InfoMsg>(info, FBT_CLI_ASK_DIRFILE, cliCore_->GetRemoteID());
+    return ClientCore::asyncInvoke(cliCore_, [this, frame]() { return cliCore_->Send(frame); });
 }

Gui/Control/ConnectControl.cpp

@@ -7,7 +7,7 @@
 #include "GuiUtil/Public.h"
 #include "ui_ConnectControl.h"
 
-Connecter::Connecter(QWidget* parent) : QWidget(parent), ui(new Ui::Connecter), th_(nullptr)
+Connecter::Connecter(QWidget* parent) : QWidget(parent), ui(new Ui::Connecter)
 {
     ui->setupUi(this);
     InitControl();
@@ -47,29 +47,32 @@ void Connecter::Connect()
         return;
     }
 
-    if (th_) {
+    sockWorker_ = new SocketWorker(clientCore_, nullptr);
-        if (th_->isRunning()) {
+    clientCore_->moveToThread(sockWorker_);
-            th_->quit();
-            th_->wait(1000);
-        }
-        delete th_;
-    }
 
-    auto* worker = new ConnectWorker(clientCore_, nullptr);
+    connect(sockWorker_, &SocketWorker::conSuccess, this, [this]() {
-    th_ = new QThread();
+        setState(ConnectState::CS_CONNECTED);
-    worker->moveToThread(th_);
+        qInfo() << QString(tr("Connected."));
-    clientCore_->moveToThread(th_);
-
-    connect(th_, &QThread::started,
-            [this, worker, ip, port]() { worker->doConnect(ip, port.toInt(), this->parent()->thread()); });
-    connect(worker, &ConnectWorker::connecting, this, [this]() { setState(ConnectState::CS_CONNECTING); });
-    connect(worker, &ConnectWorker::connectResult, this, [this](bool success) {
-        emit sendConnect(success ? ConnectState::CS_CONNECTED : ConnectState::CS_DISCONNECT);
-        th_->quit();
     });
-    connect(th_, &QThread::finished, worker, &QObject::deleteLater);
+
-    connect(th_, &QThread::finished, th_, &QObject::deleteLater);
+    connect(sockWorker_, &SocketWorker::conFailed, this, [this]() {
-    th_->start();
+        setState(ConnectState::CS_DISCONNECT);
+        qInfo() << QString(tr("Connect failed."));
+    });
+
+    connect(sockWorker_, &SocketWorker::connecting, this, [this]() {
+        setState(ConnectState::CS_CONNECTING);
+        qInfo() << QString(tr("Connecting..."));
+    });
+
+    connect(sockWorker_, &SocketWorker::disconnected, this, [this]() {
+        setState(ConnectState::CS_DISCONNECT);
+        qInfo() << QString(tr("Disconnected."));
+    });
+
+    connect(sockWorker_, &QThread::finished, sockWorker_, &QObject::deleteLater);
+    sockWorker_->SetConnectInfo(ip, port.toInt());
+    sockWorker_->start();
 }
 
 void Connecter::setState(ConnectState cs)
@@ -78,7 +81,6 @@ void Connecter::setState(ConnectState cs)
     case CS_CONNECTING:
         ui->btnConnect->setEnabled(false);
         ui->btnDisconnect->setEnabled(false);
-        qInfo() << QString(tr("Connecting..."));
         break;
     case CS_CONNECTED:
         ui->btnConnect->setEnabled(false);
@@ -99,7 +101,8 @@ void Connecter::RefreshClient()
     auto frame = QSharedPointer<FrameBuffer>::create();
     frame->data = infoPack(info);
     frame->type = FBT_SER_MSG_ASKCLIENTS;
-    if (!clientCore_->Send(frame)) {
+    auto sendRet = ClientCore::asyncInvoke(clientCore_, [this, frame]() { return clientCore_->Send(frame); });
+    if (!sendRet) {
         qCritical() << QString(tr("send ask client list failed."));
         return;
     }

Gui/Control/ConnectControl.h

@@ -50,35 +50,8 @@ private:
 
 private:
     QMenu* menu_;
-    QThread* th_;
+    SocketWorker* sockWorker_{};
-    QThread* mainTh_;
     QStandardItemModel* model_;
 };
 
-class ConnectWorker : public QObject
-{
-    Q_OBJECT
-public:
-    explicit ConnectWorker(ClientCore* clientCore, QObject* parent = nullptr)
-        : QObject(parent), clientCore_(clientCore)
-    {
-    }
-
-public slots:
-    void doConnect(const QString& ip, int port, QThread* parentThread)
-    {
-        emit connecting();
-        bool connected = clientCore_->Connect(ip, port);
-        clientCore_->moveToThread(parentThread);
-        emit connectResult(connected);
-    }
-
-signals:
-    void connectResult(bool success);
-    void connecting();
-
-private:
-    ClientCore* clientCore_;
-};
-
 #endif   // CONNECTCONTROL_H

Gui/Control/FileControl.cpp

@@ -42,6 +42,11 @@ void FileManager::SetModeStr(const QString& modeStr, int type, ClientCore* clien
     ui->tableWidget->setBasePathCall([this]() { return curRoot_; });
 }
 
+void FileManager::SetOtherSideCall(const std::function<QString()>& call)
+{
+    ui->tableWidget->setOtherSideCall(call);
+}
+
 void FileManager::InitControl()
 {
     QStringList headers;

Gui/Control/FileControl.h

@@ -22,6 +22,7 @@ public:
 
 public:
     void SetModeStr(const QString& modeStr, int type = 0, ClientCore* clientCore = nullptr);
+    void SetOtherSideCall(const std::function<QString()>& call);
     QString GetCurRoot();
 
 signals:

Gui/Control/cusTableWidget.cpp

@@ -42,6 +42,11 @@ void CustomTableWidget::setRemoteIDCall(const std::function<QString()>& call)
     ridCall_ = call;
 }
 
+void CustomTableWidget::setOtherSideCall(const std::function<QString()>& call)
+{
+    otherSideCall_ = call;
+}
+
 void CustomTableWidget::dropEvent(QDropEvent* event)
 {
     if (!event->mimeData()->hasFormat("application/x-qabstractitemmodeldatalist")) {
@@ -51,12 +56,29 @@ void CustomTableWidget::dropEvent(QDropEvent* event)
     QByteArray encoded = event->mimeData()->data("application/x-qabstractitemmodeldatalist");
     QDataStream stream(&encoded, QIODevice::ReadOnly);
 
+    QVector<TransTask> tasks;
     QList<QTableWidgetItem*> draggedItems;
     while (!stream.atEnd()) {
         int row, col;
         QMap<int, QVariant> roleData;
         stream >> row >> col >> roleData;
+        if (col != 1) {
+            continue;
         }
+        TransTask task;
+        task.isUpload = isRemote_;
+        task.localId = oidCall_();
+        task.remoteId = ridCall_();
+        if (isRemote_) {
+            task.remotePath = basePathCall_();
+            task.localPath = Util::Join(otherSideCall_(), roleData[Qt::DisplayRole].toString());
+        } else {
+            task.localPath = basePathCall_();
+            task.remotePath = Util::Join(otherSideCall_(), roleData[Qt::DisplayRole].toString());
+        }
+        tasks.push_back(task);
+    }
+    emit sigTasks(tasks);
 }
 
 void CustomTableWidget::dragEnterEvent(QDragEnterEvent* event)

Gui/Control/cusTableWidget.h

@@ -20,6 +20,7 @@ public:
     void setBasePathCall(const std::function<QString()>& call);
     void setOwnIDCall(const std::function<QString()>& call);
     void setRemoteIDCall(const std::function<QString()>& call);
+    void setOtherSideCall(const std::function<QString()>& call);
 
 protected:
     void dropEvent(QDropEvent* event) override;

Gui/Form/Transform.cpp

@@ -33,20 +33,20 @@ void TransForm::SetTasks(const QVector<TransTask>& tasks)
 void TransForm::startTask()
 {
     for (auto& task : tasks_) {
-        sigSetUi(task);
+        emit sigSetUi(task);
         if (task.isUpload) {
             fileTrans_->ReqSendFile(task);
             while (true) {
                 auto progress = fileTrans_->GetSendProgress();
                 if (progress < 0) {
-                    sigFailed();
+                    emit sigFailed();
                     break;
                 }
                 if (progress >= 100.0) {
-                    sigDone();
+                    emit sigDone();
                     break;
                 }
-                sigProgress(progress);
+                emit sigProgress(progress);
                 QThread::msleep(10);
             }
         } else {
@@ -54,14 +54,14 @@ void TransForm::startTask()
             while (true) {
                 auto progress = fileTrans_->GetDownProgress();
                 if (progress < 0) {
-                    sigFailed();
+                    emit sigFailed();
                     break;
                 }
                 if (progress >= 100.0) {
-                    sigDone();
+                    emit sigDone();
                     break;
                 }
-                sigProgress(progress);
+                emit sigProgress(progress);
                 QThread::msleep(10);
             }
         }

Gui/frelayGUI.cpp

@@ -5,6 +5,7 @@
 #include <fversion.h>
 
 #include "./ui_frelayGUI.h"
+#include "Control/LogControl.h"
 
 static LogPrint* logPrint = nullptr;
 
@@ -20,7 +21,7 @@ frelayGUI::frelayGUI(QWidget* parent) : QMainWindow(parent), ui(new Ui::frelayGU
 
     QLabel* permanent = new QLabel(this);
     permanent->setFrameStyle(QFrame::Box | QFrame::Sunken);
-    permanent->setText(QString("%1 on %2").arg(VERSION_GIT_COMMIT).arg(VERSION_GIT_BRANCH));
+    permanent->setText(QString("%1 on %2").arg(VERSION_GIT_COMMIT, VERSION_GIT_BRANCH));
     this->statusBar()->addPermanentWidget(permanent);
 }
 
@@ -44,8 +45,11 @@ void frelayGUI::InitControl()
 
     localFile_ = new FileManager(this);
     remoteFile_ = new FileManager(this);
+
     localFile_->SetModeStr(tr("Local:"));
+    localFile_->SetOtherSideCall([this]() { return remoteFile_->GetCurRoot(); });
     remoteFile_->SetModeStr(tr("Remote:"), 1, clientCore_);
+    remoteFile_->SetOtherSideCall([this]() { return localFile_->GetCurRoot(); });
 
     tabWidget_ = new QTabWidget(this);
 

Gui/frelayGUI.h

@@ -5,11 +5,9 @@
 #include <QFile>
 #include <QMainWindow>
 #include <QTabWidget>
-#include <thread>
 
 #include "Control/ConnectControl.h"
 #include "Control/FileControl.h"
-#include "Control/LogControl.h"
 #include "Control/CompareControl.h"
 #include "Form/Transform.h"
 

Protocol/Protocol.h

@@ -38,6 +38,8 @@ struct FrameBuffer {
     QString fid;
     QString tid;
     FrameBufferType type = FBT_NONE;
+    bool sendRet;
+    std::function<void(QSharedPointer<FrameBuffer>)> call{};
 };
 
 class Protocol

Util/Util.h

@@ -1,8 +1,8 @@
 #ifndef UTIL_H
 #define UTIL_H
 
-#include <QObject>
 #include <InfoDirFile.h>
+#include <QObject>
 
 class Util : public QObject
 {
@@ -18,7 +18,8 @@ public:
     static void ConsoleMsgHander(QtMsgType type, const QMessageLogContext& context, const QString& msg);
 };
 
-class DirFileHelper : public QObject {
+class DirFileHelper : public QObject
+{
     Q_OBJECT
 public:
     DirFileHelper() = default;