taynpg/frelay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

尝试改成asio通信

Browse Source
This commit is contained in:
taynpg
2026-03-24 16:53:06 +08:00
parent 5843332980
commit 5bb1836bd1
640 changed files with 169314 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

162
ClientCore/ClientCore.cpp
View File

@@ -2,25 +2,20 @@
#include <QDebug>
ClientCore::ClientCore(QObject* parent) : QObject(parent)
ClientCore::ClientCore(QObject* parent, asio::io_context& ioContext)
: QObject(parent), ioContext_(ioContext), mSocket_(ioContext_)
{
}
void ClientCore::Instance()
{
// qDebug() << "Instance() thread:" << QThread::currentThread();
socket_ = new QTcpSocket(this);
connect(socket_, &QTcpSocket::readyRead, this, &ClientCore::onReadyRead);
connect(socket_, &QTcpSocket::disconnected, this, &ClientCore::onDisconnected);
}
ClientCore::~ClientCore()
{
}
bool ClientCore::SendFrame(QSharedPointer<FrameBuffer> frame)
void ClientCore::Instance()
{
return Send(frame);
// qDebug() << "Instance() thread:" << QThread::currentThread();
// connect(socket_, &QTcpSocket::readyRead, this, &ClientCore::onReadyRead);
// connect(socket_, &QTcpSocket::disconnected, this, &ClientCore::onDisconnected);
}
void ClientCore::DoConnect(const QString& ip, quint16 port)
@@ -40,9 +35,13 @@ bool ClientCore::Connect(const QString& ip, quint16 port)
qInfo() << QString(tr("已连接。"));
return true;
}
socket_->connectToHost(ip, port);
if (!socket_->waitForConnected(3000)) {
qCritical() << QString(tr("%1:%2 连接失败。")).arg(ip).arg(port);
try {
asio::ip::tcp::resolver resolver(ioContext_);
asio::ip::tcp::resolver::results_type endpoints = resolver.resolve(ip.toStdString(), std::to_string(port));
asio::connect(mSocket_, endpoints);
return true;
} catch (const std::exception& ex) {
qCritical() << QString(tr("%1:%2 连接失败。%3")).arg(ip).arg(port).arg(ex.what());
return false;
}
qInfo() << QString(tr("%1:%2 连接成功。")).arg(ip).arg(port);
@@ -53,26 +52,58 @@ bool ClientCore::Connect(const QString& ip, quint16 port)
void ClientCore::Disconnect()
{
QMutexLocker locker(&conMutex_);
if (socket_ && socket_->state() != QAbstractSocket::UnconnectedState) {
socket_->disconnectFromHost();
if (socket_->state() != QAbstractSocket::UnconnectedState) {
socket_->waitForDisconnected(1000);
if (mSocket_.is_open()) {
try {
mSocket_.shutdown(asio::ip::tcp::socket::shutdown_both);
mSocket_.close();
} catch (const std::exception& ex) {
qCritical() << QString(tr("Error during disconnection: %1")).arg(ex.what());
}
}
// if (socket_ && socket_->state() != QAbstractSocket::UnconnectedState) {
// socket_->disconnectFromHost();
// if (socket_->state() != QAbstractSocket::UnconnectedState) {
// socket_->waitForDisconnected(1000);
// }
// }
connected_ = false;
}
asio::awaitable<void> ClientCore::recvData()
{
for (; mSocket_.is_open();) {
try {
auto bytes_transferred = co_await mSocket_.async_read_some(asio::buffer(mBuffer_), asio::use_awaitable);
recvBuffer_.append(mBuffer_.data(), bytes_transferred);
while (true) {
auto frame = Protocol::ParseBuffer(recvBuffer_);
if (frame == nullptr) {
break;
}
std::unique_lock lock(frameMutex_);
// 暂时这样，后面优化成控制信息优先发出。
frameQueue_.push(frame);
}
} catch (const std::exception& ex) {
qCritical() << QString(tr("Error during receiving header: %1")).arg(ex.what());
co_return;
}
}
}
void ClientCore::onReadyRead()
{
QByteArray data = socket_->readAll();
recvBuffer_.append(data);
while (true) {
auto frame = Protocol::ParseBuffer(recvBuffer_);
if (frame == nullptr) {
break;
}
UseFrame(frame);
}
// QByteArray data = socket_->readAll();
// recvBuffer_.append(data);
// while (true) {
// auto frame = Protocol::ParseBuffer(recvBuffer_);
// if (frame == nullptr) {
// break;
// }
// UseFrame(frame);
// }
}
void ClientCore::onDisconnected()
@@ -82,7 +113,7 @@ void ClientCore::onDisconnected()
emit sigDisconnect();
}
void ClientCore::handleAsk(QSharedPointer<FrameBuffer> frame)
asio::awaitable<void> ClientCore::handleAsk(QSharedPointer<FrameBuffer> frame)
{
InfoMsg msg = infoUnpack<InfoMsg>(frame->data);
// TODO: 处理询问请求
@@ -105,48 +136,49 @@ void ClientCore::handleAsk(QSharedPointer<FrameBuffer> frame)
}
}
}
if (!Send<InfoMsg>(msg, FBT_MSGINFO_ANSWER, frame->fid)) {
bool sr = co_await Send<InfoMsg>(msg, FBT_MSGINFO_ANSWER, frame->fid);
if (!sr) {
auto logMsg = tr("给") + frame->fid + tr("返回检查文件存在性消息失败。");
qCritical() << logMsg;
return;
co_return;
}
return;
co_return;
}
if (msg.command == STRMSG_AC_RENAME_FILEDIR) {
msg.command = STRMSG_AC_ANSWER_RENAME_FILEDIR;
msg.msg = Util::Rename(msg.fromPath, msg.toPath, msg.type == STR_DIR);
if (!Send<InfoMsg>(msg, FBT_MSGINFO_ANSWER, frame->fid)) {
if (!co_await Send<InfoMsg>(msg, FBT_MSGINFO_ANSWER, frame->fid)) {
auto logMsg = tr("给") + frame->fid + tr("返回重命名结果消息失败。");
qCritical() << logMsg;
return;
co_return;
}
return;
co_return;
}
if (msg.command == STRMSG_AC_DEL_FILEDIR) {
msg.command = STRMSG_AC_ANSWER_DEL_FILEDIR;
msg.msg = Util::Delete(msg.fromPath);
if (!Send<InfoMsg>(msg, FBT_MSGINFO_ANSWER, frame->fid)) {
if (!co_await Send<InfoMsg>(msg, FBT_MSGINFO_ANSWER, frame->fid)) {
auto logMsg = tr("给") + frame->fid + tr("返回删除结果消息失败。");
qCritical() << logMsg;
return;
co_return;
}
return;
co_return;
}
if (msg.command == STRMSG_AC_NEW_DIR) {
msg.command = STRMSG_AC_ANSWER_NEW_DIR;
msg.msg = Util::NewDir(msg.fromPath);
if (!Send<InfoMsg>(msg, FBT_MSGINFO_ANSWER, frame->fid)) {
if (!co_await Send<InfoMsg>(msg, FBT_MSGINFO_ANSWER, frame->fid)) {
auto logMsg = tr("给") + frame->fid + tr("返回新建结果消息失败。");
qCritical() << logMsg;
return;
co_return;
}
return;
co_return;
}
// 未知信息
qWarning() << QString(tr("未知询问信息类型：%1")).arg(msg.command);
}
void ClientCore::UseFrame(QSharedPointer<FrameBuffer> frame)
asio::awaitable<void> ClientCore::UseFrame(QSharedPointer<FrameBuffer> frame)
{
switch (frame->type) {
case FrameBufferType::FBT_SER_MSG_ASKCLIENTS: {
@@ -171,11 +203,11 @@ void ClientCore::UseFrame(QSharedPointer<FrameBuffer> frame)
InfoMsg info = infoUnpack<InfoMsg>(frame->data);
if (!localFile_.GetDirFile(info.msg, vec)) {
qWarning() << QString(tr("访问文件失败： %1")).arg(info.msg);
return;
co_return;
}
if (!Send<DirFileInfoVec>(vec, FBT_CLI_ANS_DIRFILE, frame->fid)) {
if (!co_await Send<DirFileInfoVec>(vec, FBT_CLI_ANS_DIRFILE, frame->fid)) {
qCritical() << QString(tr("发送文件列表结果失败。"));
return;
co_return;
}
break;
}
@@ -183,9 +215,9 @@ void ClientCore::UseFrame(QSharedPointer<FrameBuffer> frame)
InfoMsg info;
info.msg = Util::GetUserHome();
info.list = Util::GetLocalDrivers();
if (!Send<InfoMsg>(info, FBT_CLI_ANS_HOME, frame->fid)) {
if (!co_await Send<InfoMsg>(info, FBT_CLI_ANS_HOME, frame->fid)) {
qCritical() << QString(tr("send home failed."));
return;
co_return;
}
break;
}
@@ -248,7 +280,7 @@ void ClientCore::UseFrame(QSharedPointer<FrameBuffer> frame)
break;
}
case FBT_MSGINFO_ASK: {
handleAsk(frame);
co_await handleAsk(frame);
break;
}
case FBT_MSGINFO_ANSWER: {
@@ -269,35 +301,28 @@ void ClientCore::UseFrame(QSharedPointer<FrameBuffer> frame)
}
}
bool ClientCore::Send(QSharedPointer<FrameBuffer> frame)
asio::awaitable<bool> ClientCore::Send(QSharedPointer<FrameBuffer> frame)
{
if (frame == nullptr) {
return false;
co_return false;
}
auto data = Protocol::PackBuffer(frame);
if (data.size() == 0) {
return false;
co_return false;
}
return Send(data.constData(), data.size());
co_return co_await Send(data.constData(), data.size());
}
bool ClientCore::Send(const char* data, qint64 len)
asio::awaitable<bool> ClientCore::Send(const char* data, size_t size)
{
if (socket_->state() != QAbstractSocket::ConnectedState) {
qCritical() << QString("客户端 %1 未连接...").arg(remoteID_);
return false;
try {
//auto c = mSocket_.get_executor();
auto bytes_transferred = co_await asio::async_write(mSocket_, asio::buffer(data, size), asio::use_awaitable);
co_return bytes_transferred == size;
} catch (std::exception& e) {
qCritical() << QString("向服务器发送数据失败： %1").arg(e.what());
co_return false;
}
qint64 bytesWritten = -1;
{
QMutexLocker locker(&sockMut_);
bytesWritten = socket_->write(data, len);
}
if (bytesWritten == -1 || !socket_->waitForBytesWritten(5000)) {
qCritical() << QString("向服务器发送数据失败： %1").arg(socket_->errorString());
return false;
}
return true;
}
bool ClientCore::IsConnect()
@@ -305,6 +330,11 @@ bool ClientCore::IsConnect()
return connected_;
}
const asio::any_io_executor& ClientCore::get_executor()
{
return mSocket_.get_executor();
}
void ClientCore::SetRemoteID(const QString& id)
{
GlobalData::Ins()->SetRemoteID(id);