前情回顾
前面我们搞定了1，2，3以及5过程
今天主要完成
4，6和7
资源服务器grpc设置
因为资源服务器要通知ChatServer，所以要设置grpc客户端
完善下proto协议，新增消息通知请求
具体代码逻辑
message NotifyChatImgReq{
int32 from_uid = 1;
int32 to_uid = 2;
int32 message_id = 3;
string file_name = 4;
int64 total_size = 5;
int32 thread_id =6;
}
message NotifyChatImgRsp{
int32 error = 1;
int32 from_uid = 2;
int32 to_uid = 3;
int32 message_id = 4;
string file_name = 5;
int64 total_size = 6;
int32 thread_id =7;
}
service ChatService {
rpc NotifyAddFriend(AddFriendReq) returns (AddFriendRsp) {}
rpc RplyAddFriend(RplyFriendReq) returns (RplyFriendRsp) {}
rpc SendChatMsg(SendChatMsgReq) returns (SendChatMsgRsp) {}
rpc NotifyAuthFriend(AuthFriendReq) returns (AuthFriendRsp) {}
rpc NotifyTextChatMsg(TextChatMsgReq) returns (TextChatMsgRsp){}
rpc NotifyKickUser(KickUserReq) returns (KickUserRsp){}
rpc NotifyChatImgMsg(NotifyChatImgReq) returns (NotifyChatImgRsp){}
}
实现grpc客户端逻辑如下:
#pragma once
#include "const.h"
#include "Singleton.h"
#include "ConfigMgr.h"
#include "message.grpc.pb.h"
#include "message.pb.h"
#include <grpcpp/grpcpp.h>
#include <queue>
#include <condition_variable>
using grpc::Channel;
using grpc::Status;
using grpc::ClientContext;
using message::ChatService;
using message::NotifyChatImgReq;
using message::NotifyChatImgRsp;
class ChatServerConPool {
public:
ChatServerConPool(size_t poolSize, std::string host, std::string port)
: poolSize_(poolSize), host_(host), port_(port), b_stop_(false) {
for (size_t i = 0; i < poolSize_; ++i) {
std::shared_ptr<Channel> channel = grpc::CreateChannel(host + ":" + port,
grpc::InsecureChannelCredentials());
connections_.push(ChatService::NewStub(channel));
}
}
~ChatServerConPool() {
std::lock_guard<std::mutex> lock(mutex_);
Close();
while (!connections_.empty()) {
connections_.pop();
}
}
std::unique_ptr<ChatService::Stub> getConnection() {
std::unique_lock<std::mutex> lock(mutex_);
cond_.wait(lock, [this] {
if (b_stop_) {
return true;
}
return !connections_.empty();
});
//如果停止则直接返回空指针
if (b_stop_) {
return nullptr;
}
auto context = std::move(connections_.front());
connections_.pop();
return context;
}
void returnConnection(std::unique_ptr<ChatService::Stub> context) {
std::lock_guard<std::mutex> lock(mutex_);
if (b_stop_) {
return;
}
connections_.push(std::move(context));
cond_.notify_one();
}
void Close() {
b_stop_ = true;
cond_.notify_all();
}
private:
atomic<bool> b_stop_;
size_t poolSize_;
std::string host_;
std::string port_;
std::queue<std::unique_ptr<ChatService::Stub>> connections_;
std::mutex mutex_;
std::condition_variable cond_;
};
class ChatServerGrpcClient :public Singleton<ChatServerGrpcClient>
{
friend class Singleton<ChatServerGrpcClient>;
public:
~ChatServerGrpcClient() {
}
NotifyChatImgRsp NotifyChatImgMsg(int message_id, std::string chatserver);
private:
ChatServerGrpcClient();
//sever_ip到连接池的映射, <chatserver1,std::unique_ptr<ChatServerConPool>>
std::unordered_map<std::string, std::unique_ptr<ChatServerConPool>> _hash_pools;
};
具体实现
#include "ChatServerGrpcClient.h"
#include "MysqlMgr.h"
NotifyChatImgRsp ChatServerGrpcClient::NotifyChatImgMsg(int message_id,std::string chatserver)
{
ClientContext context;
NotifyChatImgRsp reply;
NotifyChatImgReq request;
request.set_message_id(message_id);
if (_hash_pools.find(chatserver) == _hash_pools.end()) {
reply.set_error(ErrorCodes::ServerIpErr);
return reply;
}
auto chat_msg = MysqlMgr::GetInstance()->GetChatMsgById(message_id);
request.set_file_name(chat_msg->content);
request.set_from_uid(chat_msg->sender_id);
request.set_to_uid(chat_msg->recv_id);
request.set_thread_id(chat_msg->thread_id);
// 资源文件路径
auto file_dir = ConfigMgr::Inst().GetFileOutPath();
//该消息是接收方客户端发送过来的,服务器将资源存储在发送方的文件夹中
auto uid_str = std::to_string(chat_msg->sender_id);
auto file_path = (file_dir / uid_str / chat_msg->content);
boost::uintmax_t file_size = boost::filesystem::file_size(file_path);
request.set_total_size(file_size);
auto &pool_ = _hash_pools[chatserver];
auto stub = pool_->getConnection();
Status status = stub->NotifyChatImgMsg(&context, request, &reply);
Defer defer([&stub, &pool_, this]() {
pool_->returnConnection(std::move(stub));
});
if (status.ok()) {
return reply;
}
else {
reply.set_error(ErrorCodes::RPCFailed);
return reply;
}
}
ChatServerGrpcClient::ChatServerGrpcClient()
{
auto& gCfgMgr = ConfigMgr::Inst();
std::string host1 = gCfgMgr["chatserver1"]["Host"];
std::string port1 = gCfgMgr["chatserver1"]["Port"];
_hash_pools["chatserver1"] = std::make_unique<ChatServerConPool>(5, host1, port1);
std::string host2 = gCfgMgr["chatserver2"]["Host"];
std::string port2 = gCfgMgr["chatserver2"]["Port"];
_hash_pools["chatserver2"] = std::make_unique<ChatServerConPool>(5, host2, port2);
}
实现了通知接口，用来通知ChatServer图片消息上传完成，让ChatServer再通知其他客户端.
ChatServer响应资源服务器通知
ChatServer的proto也需要进行同样配置，这里略去
具体在ChatServiceImpl中添加响应消息通知的逻辑
Status ChatServiceImpl::NotifyChatImgMsg(::grpc::ServerContext* context, const ::message::NotifyChatImgReq* request, ::message::NotifyChatImgRsp* response)
{
//查找用户是否在本服务器
auto uid = request->to_uid();
auto session = UserMgr::GetInstance()->GetSession(uid);
Defer defer([request, response]() {
//设置具体的回包信息
response->set_error(ErrorCodes::Success);
response->set_message_id(request->message_id());
});
//用户不在内存中则直接返回
if (session == nullptr) {
//这里只是返回1个状态
return Status::OK;
}
//在内存中则直接发送通知对方
session->NotifyChatImgRecv(request);
//这里只是返回1个状态
return Status::OK;
}
通过Session通知客户端
void CSession::NotifyChatImgRecv(const ::message::NotifyChatImgReq* request) {
Json::Value rtvalue;
rtvalue["error"] = ErrorCodes::Success;
rtvalue["message_id"] = request->message_id();
rtvalue["sender_id"] = request->from_uid();
rtvalue["receiver_id"] = request->to_uid();
rtvalue["img_name"] = request->file_name();
rtvalue["total_size"] = std::to_string(request->total_size());
rtvalue["thread_id"] = request->thread_id();
std::string return_str = rtvalue.toStyledString();
//通知图片聊天信息
Send(return_str, ID_NOTIFY_IMG_CHAT_MSG_REQ);
return;
}
客户端获取通知
客户端收到服务器通知后，会优先查看本地资源是否存在，如果存在则直接加载图片，添加聊天记录到页面。
如果不存在则组织下载，但是也需要将消息添加到聊天界面。
_handlers.insert(ID_NOTIFY_IMG_CHAT_MSG_REQ, [this](ReqId id, int len, QByteArray data) {
Q_UNUSED(len);
qDebug() << "handle id is " << id << " data is " << data;
// 将QByteArray转换为QJsonDocument
QJsonDocument jsonDoc = QJsonDocument::fromJson(data);
// 检查转换是否成功
if (jsonDoc.isNull()) {
qDebug() << "Failed to create QJsonDocument.";
return;
}
QJsonObject jsonObj = jsonDoc.object();
qDebug() << "receive notify img chat msg req success" ;
//收到消息后转发给页面
auto thread_id = jsonObj["thread_id"].toInt();
auto sender_id = jsonObj["sender_id"].toInt();
auto message_id = jsonObj["message_id"].toInt();
auto receiver_id = jsonObj["receiver_id"].toInt();
auto img_name = jsonObj["img_name"].toString();
auto total_size_str = jsonObj["total_size"].toString();
auto total_size = total_size_str.toLongLong();
auto uid = UserMgr::GetInstance()->GetUid();
//客户端存储聊天记录，按照如下格式存储C:\Users\secon\AppData\Roaming\llfcchat\chatimg\uid, uid为对方uid
QString storageDir = QStandardPaths::writableLocation(QStandardPaths::AppDataLocation);
QString img_path_str = storageDir +"/user/"+ QString::number(uid)+ "/chatimg/" + QString::number(sender_id);
auto file_info = UserMgr::GetInstance()->GetTransFileByName(img_name);
//正常情况是找不到的，所以这里初始化一个文件信息放入UserMgr中管理
if (!file_info) {
//预览图先默认空白，md5为空
file_info = std::make_shared<MsgInfo>(MsgType::IMG_MSG, img_path_str, CreateLoadingPlaceholder(200, 200), img_name, total_size, "");
UserMgr::GetInstance()->AddTransFile(img_name, file_info);
}
file_info->_msg_id = message_id;
file_info->_sender = sender_id;
file_info->_receiver = receiver_id;
file_info->_thread_id = thread_id;
//设置文件传输的类型
file_info->_transfer_type = TransferType::Download;
//设置文件传输状态
file_info->_transfer_state = TransferState::Uploading;
auto img_chat_data_ptr = std::make_shared<ImgChatData>(file_info, "",
thread_id, ChatFormType::PRIVATE, ChatMsgType::PIC,
sender_id, MsgStatus::READED);
emit sig_img_chat_msg(img_chat_data_ptr);
//组织请求，准备下载
QJsonObject jsonObj_send;
jsonObj_send["name"] = img_name;
jsonObj_send["seq"] = file_info->_seq;
jsonObj_send["trans_size"] = "0";
jsonObj_send["total_size"] = QString::number(file_info->_total_size);
jsonObj_send["token"] = UserMgr::GetInstance()->GetToken();
jsonObj_send["sender_id"] = sender_id;
jsonObj_send["receiver_id"] = receiver_id;
jsonObj_send["message_id"] = message_id;
jsonObj_send["uid"] = uid;
//客户端存储聊天记录，按照如下格式存储C:\Users\secon\AppData\Roaming\llfcchat\chatimg\uid, uid为对方uid
QDir chatimgDir(img_path_str);
jsonObj["client_path"] = img_path_str;
if (!chatimgDir.exists()) {
chatimgDir.mkpath("."); // 创建当前路径
}
QJsonDocument doc(jsonObj_send);
auto send_data = doc.toJson();
FileTcpMgr::GetInstance()->SendData(ID_IMG_CHAT_DOWN_REQ, send_data);
});
收到服务器通知后，开始构造json数据，发送ID_IMG_CHAT_DOWN_REQ请求
聊天记录添加
客户端在请求服务器资源的时候，因为本地没有资源，可以先在聊天界面生成一个预览的空白图片，同时显示进度条
这部分逻辑是在客户端的tcpmgr中处理服务器通知聊天消息的逻辑里
_handlers.insert(ID_NOTIFY_IMG_CHAT_MSG_REQ, [this](ReqId id, int len, QByteArray data) {
//...
//发送给界面显示
emit sig_img_chat_msg(img_chat_data_ptr);
}
客户端将图片消息发送给界面显示
在ChatDialog的构造函数中添加信号槽链接
connect(TcpMgr::GetInstance().get(), &TcpMgr::sig_img_chat_msg,
this, &ChatDialog::slot_img_chat_msg);
ChatDialog收到该信号后，会触发添加消息的逻辑
void ChatDialog::slot_img_chat_msg(std::shared_ptr<ImgChatData> imgchat) {
//更新数据
auto thread_id = imgchat->GetThreadId();
auto thread_data = UserMgr::GetInstance()->GetChatThreadByThreadId(thread_id);
thread_data->AddMsg(imgchat);
if (_cur_chat_thread_id != thread_id) {
return;
}
ui->chat_page->AppendOtherMsg(imgchat);
}
添加其他消息的逻辑, 此处都是将其他人发送的图片消息添加到聊天界面显示
void ChatPage::AppendOtherMsg(std::shared_ptr<ChatDataBase> msg) {
auto self_info = UserMgr::GetInstance()->GetUserInfo();
ChatRole role;
if (msg->GetSendUid() == self_info->_uid) {
role = ChatRole::Self;
ChatItemBase* pChatItem = new ChatItemBase(role);
pChatItem->setUserName(self_info->_name);
SetSelfIcon(pChatItem, self_info->_icon);
QWidget* pBubble = nullptr;
if (msg->GetMsgType() == ChatMsgType::TEXT) {
pBubble = new TextBubble(role, msg->GetMsgContent());
}
else if (msg->GetMsgType() == ChatMsgType::PIC) {
auto img_msg = dynamic_pointer_cast<ImgChatData>(msg);
auto pic_bubble = new PictureBubble(img_msg->_msg_info->_preview_pix, role, img_msg->_msg_info->_total_size);
pic_bubble->setMsgInfo(img_msg->_msg_info);
pBubble = pic_bubble;
//连接暂停和恢复信号
connect(dynamic_cast<PictureBubble*>(pBubble), &PictureBubble::pauseRequested,
this, &ChatPage::on_clicked_paused);
connect(dynamic_cast<PictureBubble*>(pBubble), &PictureBubble::resumeRequested,
this, &ChatPage::on_clicked_resume);
}
pChatItem->setWidget(pBubble);
auto status = msg->GetStatus();
pChatItem->setStatus(status);
ui->chat_data_list->appendChatItem(pChatItem);
_base_item_map[msg->GetMsgId()] = pChatItem;
}
else {
role = ChatRole::Other;
ChatItemBase* pChatItem = new ChatItemBase(role);
auto friend_info = UserMgr::GetInstance()->GetFriendById(msg->GetSendUid());
if (friend_info == nullptr) {
return;
}
pChatItem->setUserName(friend_info->_name);
// 使用正则表达式检查是否是默认头像
QRegularExpression regex("^:/res/head_(\\d+)\\.jpg$");
QRegularExpressionMatch match = regex.match(friend_info->_icon);
if (match.hasMatch()) {
pChatItem->setUserIcon(QPixmap(friend_info->_icon));
}
else {
// 如果是用户上传的头像，获取存储目录
QString storageDir = QStandardPaths::writableLocation(QStandardPaths::AppDataLocation);
auto uid = UserMgr::GetInstance()->GetUid();
QDir avatarsDir(storageDir + "/user/" + QString::number(uid) + "/avatars");
auto file_name = QFileInfo(self_info->_icon).fileName();
// 确保目录存在
if (avatarsDir.exists()) {
QString avatarPath = avatarsDir.filePath(file_name); // 获取上传头像的完整路径
QPixmap pixmap(avatarPath); // 加载上传的头像图片
if (!pixmap.isNull()) {
pChatItem->setUserIcon(pixmap);
}
else {
qWarning() << "无法加载上传的头像：" << avatarPath;
auto icon_label = pChatItem->getIconLabel();
LoadHeadIcon(avatarPath, icon_label, file_name, "self_icon");
}
}
else {
qWarning() << "头像存储目录不存在：" << avatarsDir.path();
//创建目录
avatarsDir.mkpath(".");
auto icon_label = pChatItem->getIconLabel();
QString avatarPath = avatarsDir.filePath(file_name);
LoadHeadIcon(avatarPath, icon_label, file_name, "self_icon");
}
}
QWidget* pBubble = nullptr;
if (msg->GetMsgType() == ChatMsgType::TEXT) {
pBubble = new TextBubble(role, msg->GetMsgContent());
}
else if (msg->GetMsgType() == ChatMsgType::PIC) {
auto img_msg = dynamic_pointer_cast<ImgChatData>(msg);
auto pic_bubble = new PictureBubble(img_msg->_msg_info->_preview_pix, role, img_msg->_msg_info->_total_size);
pic_bubble->setMsgInfo(img_msg->_msg_info);
pBubble = pic_bubble;
//连接暂停和恢复信号
connect(dynamic_cast<PictureBubble*>(pBubble), &PictureBubble::pauseRequested,
this, &ChatPage::on_clicked_paused);
connect(dynamic_cast<PictureBubble*>(pBubble), &PictureBubble::resumeRequested,
this, &ChatPage::on_clicked_resume);
}
pChatItem->setWidget(pBubble);
auto status = msg->GetStatus();
pChatItem->setStatus(status);
ui->chat_data_list->appendChatItem(pChatItem);
_base_item_map[msg->GetMsgId()] = pChatItem;
}
}
资源服务器响应下载请求
LogicWorker中增加请求的处理
_fun_callbacks[ID_IMG_CHAT_DOWN_REQ] = [this](std::shared_ptr<CSession> session, const short& msg_req_id,
const string& msg_data) {
Json::Reader reader;
Json::Value root;
reader.parse(msg_data, root);
auto seq = root["seq"].asInt();
auto name = root["name"].asString();
auto total_size_str = root["total_size"].asString();
auto trans_size_str = root["trans_size"].asString();
auto file_path = ConfigMgr::Inst().GetFileOutPath();
auto message_id = root["message_id"].asInt();
auto sender = root["sender_id"].asInt();
auto receiver = root["receiver_id"].asInt();
auto token = root["token"].asString();
auto uid = root["uid"].asInt();
auto callback = [=](const Json::Value& result) {
// 在异步任务完成后调用
Json::Value rtvalue = result;
rtvalue["error"] = ErrorCodes::Success;
rtvalue["name"] = name;
rtvalue["sender_id"] = sender;
rtvalue["receiver_id"] = receiver;
std::string return_str = rtvalue.toStyledString();
session->Send(return_str, ID_IMG_CHAT_DOWN_RSP);
};
// 使用 std::hash 对字符串进行哈希
std::hash<std::string> hash_fn;
size_t hash_value = hash_fn(name); // 生成哈希值
int index = hash_value % DOWN_LOAD_WORKER_COUNT;
std::cout << "Hash value: " << hash_value << std::endl;
//第一个包校验一下token是否合理
if (seq == 1) {
//从redis获取用户token是否正确
std::string uid_str = std::to_string(uid);
std::string token_key = USERTOKENPREFIX + uid_str;
std::string token_value = "";
bool success = RedisMgr::GetInstance()->Get(token_key, token_value);
Json::Value rtvalue;
if (!success) {
rtvalue["error"] = ErrorCodes::UidInvalid;
std::string return_str = rtvalue.toStyledString();
session->Send(return_str, ID_IMG_CHAT_DOWN_RSP);
return;
}
if (token_value != token) {
rtvalue["error"] = ErrorCodes::TokenInvalid;
std::string return_str = rtvalue.toStyledString();
session->Send(return_str, ID_IMG_CHAT_DOWN_RSP);
return;
}
}
auto sender_str = std::to_string(sender);
//转化为字符串
auto uid_str = std::to_string(uid);
auto file_path_str = (file_path / sender_str / name).string();
auto down_load_task = std::make_shared<DownloadTask>(session, uid, name, seq, file_path_str, callback);
FileSystem::GetInstance()->PostDownloadTaskToQue(down_load_task,index);
};
LogicWorker将请求投递给FileSystem队列，FileSystem队列排队处理消息，被DownloaderWorker处理
void DownloadWorker::task_callback(std::shared_ptr<DownloadTask> task)
{
// 解码
auto file_path_str = task->_file_path;
//std::cout << "file_path_str is " << file_path_str << std::endl;
boost::filesystem::path file_path(file_path_str);
Json::Value result;
result["error"] = ErrorCodes::Success;
if (!boost::filesystem::exists(file_path)) {
std::cerr << "文件不存在: " << file_path_str << std::endl;
result["error"] = ErrorCodes::FileNotExists;
task->_callback(result);
return;
}
std::ifstream infile(file_path_str, std::ios::binary);
if (!infile) {
std::cerr << "无法打开文件进行读取。" << std::endl;
result["error"] = ErrorCodes::FileReadPermissionFailed;
task->_callback(result);
return;
}
std::shared_ptr<FileInfo> file_info = nullptr;
if (task->_seq == 1) {
// 获取文件大小
infile.seekg(0, std::ios::end);
std::streamsize file_size = infile.tellg();
infile.seekg(0, std::ios::beg);
//如果为空，则创建FileInfo 构造数据存储
file_info = std::make_shared<FileInfo>();
file_info->_file_path_str = file_path_str;
file_info->_name = task->_name;
file_info->_seq = 1;
file_info->_total_size = file_size;
file_info->_trans_size = 0;
// 立即保存到 Redis，覆盖旧数据，设置过期时间
RedisMgr::GetInstance()->SetDownLoadInfo(task->_name, file_info);
std::cout << "[新下载] 文件: " << task->_name
<< ", 大小: " << file_size << " 字节" << std::endl;
}
else {
//断点续传，从 Redis 获取历史信息
file_info = RedisMgr::GetInstance()->GetDownloadInfo(task->_name);
if (file_info == nullptr) {
// Redis 中没有信息（可能过期了）
std::cerr << "断点续传失败，Redis 中无下载信息: " << task->_name << std::endl;
result["error"] = ErrorCodes::RedisReadErr;
task->_callback(result);
infile.close();
return;
}
// 验证序列号是否匹配
if (task->_seq != file_info->_seq) {
std::cerr << "序列号不匹配，期望: " << file_info->_seq
<< ", 实际: " << task->_seq << std::endl;
result["error"] = ErrorCodes::FileSeqInvalid;
task->_callback(result);
infile.close();
return;
}
std::cout << "[续传] 文件: " << task->_name
<< ", seq: " << task->_seq
<< ", 进度: " << file_info->_trans_size
<< "/" << file_info->_total_size << std::endl;
}
// 计算当前偏移量
std::streamsize offset = ((std::streamsize)task->_seq - 1) * MAX_FILE_LEN;
if (offset >= file_info->_total_size) {
std::cerr << "偏移量超出文件大小。" << std::endl;
result["error"] = ErrorCodes::FileOffsetInvalid;
task->_callback(result);
infile.close();
return;
}
// 定位到指定偏移量
infile.seekg(offset);
// 读取最多MAX_FILE_LEN字节
char buffer[MAX_FILE_LEN];
infile.read(buffer, MAX_FILE_LEN);
//获取read实际读取多少字节
std::streamsize bytes_read = infile.gcount();
if (bytes_read <= 0) {
std::cerr << "读取文件失败。" << std::endl;
result["error"] = ErrorCodes::FileReadFailed;
task->_callback(result);
infile.close();
return;
}
// 将读取的数据进行base64编码
std::string data_to_encode(buffer, bytes_read);
std::string encoded_data = base64_encode(data_to_encode);
// 检查是否是最后一个包
std::streamsize current_pos = offset + bytes_read;
bool is_last = (current_pos >= file_info->_total_size);
// 设置返回结果
result["data"] = encoded_data;
result["seq"] = task->_seq;
result["total_size"] = std::to_string(file_info->_total_size);
result["current_size"] = std::to_string(current_pos);
result["is_last"] = is_last;
infile.close();
if (is_last) {
std::cout << "文件读取完成: " << file_path_str << std::endl;
RedisMgr::GetInstance()->DelDownLoadInfo(task->_name);
}
else {
//更新信息
file_info->_seq++;
file_info->_trans_size = offset + bytes_read;
//更新redis
RedisMgr::GetInstance()->SetDownLoadInfo(task->_name, file_info);
}
if (task->_callback) {
task->_callback(result);
}
}
资源服务器每次收到请求后，由DownloadWorker从队列中获取请求，查询服务器资源，将资源按照seq计算偏移量最后读取数据发送给客户端。
客户端存储下载的资源
客户端需要存储服务器传输的资源
_handlers.insert(ID_IMG_CHAT_DOWN_RSP, [this](ReqId id, int len, QByteArray data) {
Q_UNUSED(len);
qDebug() << "handle id is " << id << " data is " << data;
// 将QByteArray转换为QJsonDocument
QJsonDocument jsonDoc = QJsonDocument::fromJson(data);
// 检查转换是否成功
if (jsonDoc.isNull()) {
qDebug() << "Failed to create QJsonDocument.";
return;
}
QJsonObject jsonObj = jsonDoc.object();
if (!jsonObj.contains("error")) {
int err = ErrorCodes::ERR_JSON;
qDebug() << "parse create private chat json parse failed " << err;
return;
}
int err = jsonObj["error"].toInt();
if (err != ErrorCodes::SUCCESS) {
qDebug() << "get create private chat failed, error is " << err;
return;
}
qDebug() << "Receive download file info rsp success";
QString base64Data = jsonObj["data"].toString();
int seq = jsonObj["seq"].toInt();
bool is_last = jsonObj["is_last"].toBool();
QString total_size_str = jsonObj["total_size"].toString();
qint64 total_size = total_size_str.toLongLong(nullptr);
QString current_size_str = jsonObj["current_size"].toString();
qint64 current_size = current_size_str.toLongLong(nullptr);
QString name = jsonObj["name"].toString();
auto file_info = UserMgr::GetInstance()->GetTransFileByName(name);
if (file_info == nullptr) {
qDebug() << "file: " << name << " not found";
return;
}
file_info->_current_size = current_size;
file_info->_rsp_size = current_size;
file_info->_total_size = total_size;
auto clientPath = file_info->_text_or_url;
//Base64解码
QByteArray decodedData = QByteArray::fromBase64(base64Data.toUtf8());
auto file_path = clientPath + "/" + name;
QFile file(file_path);
// 根据 seq 决定打开模式
QIODevice::OpenMode mode;
if (seq == 1) {
// 第一个包，覆盖写入
mode = QIODevice::WriteOnly;
}
else {
// 后续包，追加写入
mode = QIODevice::WriteOnly | QIODevice::Append;
}
if (!file.open(mode)) {
qDebug() << "Failed to open file for writing:" << clientPath;
qDebug() << "Error:" << file.errorString();
return;
}
qint64 bytesWritten = file.write(decodedData);
if (bytesWritten != decodedData.size()) {
qDebug() << "Failed to write all data. Written:" << bytesWritten
<< "Expected:" << decodedData.size();
}
file.close();
qDebug() << "Successfully wrote" << bytesWritten << "bytes to file";
qDebug() << "Progress:" << current_size << "/" << total_size
<< "(" << (current_size * 100 / total_size) << "%)";
if (is_last) {
qDebug() << "File download completed:" << clientPath;
UserMgr::GetInstance()->RmvTransFileByName(name);
//通知界面下载完成
emit sig_download_finish(file_info, file_path);
}
else {
//继续请求
file_info->_seq = seq + 1;
file_info->_last_confirmed_seq = seq;
if (file_info->_transfer_state == TransferState::Paused) {
//暂停状态，则直接返回
return;
}
//组织请求，准备下载
QJsonObject jsonObj_send;
jsonObj_send["name"] = name;
jsonObj_send["seq"] = file_info->_seq;
jsonObj_send["trans_size"] = QString::number(file_info->_current_size);
jsonObj_send["total_size"] = QString::number(file_info->_total_size);
jsonObj_send["token"] = UserMgr::GetInstance()->GetToken();
jsonObj_send["sender_id"] = file_info->_sender;
jsonObj_send["receiver_id"] = file_info->_receiver;
jsonObj_send["message_id"] = file_info->_msg_id;
auto uid = UserMgr::GetInstance()->GetUid();
jsonObj_send["uid"] = uid;
QJsonDocument doc(jsonObj_send);
auto send_data = doc.toJson();
FileTcpMgr::GetInstance()->SendData(ID_IMG_CHAT_DOWN_REQ, send_data);
//todo...通知界面更新进度
emit sig_update_download_progress(file_info);
}
});
通过QFile类实现文件写入。
客户端进度显示
为了让客户端更为直观的显示下载进度，可以在收到服务器消息后，将文件下载进度同步给界面，同时显示支持暂停和继续
进度通知在上述逻辑中
_handlers.insert(ID_IMG_CHAT_DOWN_RSP, [this](ReqId id, int len, QByteArray data) {
//...
emit sig_update_download_progress(file_info);
});
同样是在ChatDialog构造函数中添加消息链接
//接收tcp返回的下载进度信息
connect(FileTcpMgr::GetInstance().get(), &FileTcpMgr::sig_update_download_progress,
this, &ChatDialog::slot_update_download_progress);
进度处理槽函数
void ChatDialog::slot_update_download_progress(std::shared_ptr<MsgInfo> msg_info) {
auto chat_data = UserMgr::GetInstance()->GetChatThreadByThreadId(msg_info->_thread_id);
if (chat_data == nullptr) {
return;
}
//更新消息，其实不用更新，都是共享msg_info的一块内存，这里为了安全还是再次更新下
chat_data->UpdateProgress(msg_info);
if (_cur_chat_thread_id != msg_info->_thread_id) {
return;
}
//更新聊天界面信息
ui->chat_page->UpdateFileProgress(msg_info);
}
在ChatPage中详细处理更新
void ChatPage::UpdateFileProgress(std::shared_ptr<MsgInfo> msg_info) {
auto iter = _base_item_map.find(msg_info->_msg_id);
if (iter == _base_item_map.end()) {
return;
}
if (msg_info->_msg_type == MsgType::IMG_MSG) {
auto bubble = iter.value()->getBubble();
PictureBubble* pic_bubble = dynamic_cast<PictureBubble*>(bubble);
pic_bubble->setProgress(msg_info->_rsp_size, msg_info->_total_size);
}
}
PicBubble中完成状态显示
void PictureBubble::setProgress(int value, int total_value)
{
if (m_total_size != total_value) {
m_total_size = total_value;
}
float percent = (value / (m_total_size*1.0))*100;
m_progressBar->setValue(percent);
if (percent >= 100) {
setState(TransferState::Completed);
}
}
断点续传
因为在客户端收到服务器通知的图片聊天信息的时候，已经通过sig_img_chat_msg将消息发送给ChatDialog添加到页面上了。同时传输了图片的状态为下载中。
点击继续和暂停的逻辑可以复用PicBubble的逻辑
void PictureBubble::onPictureClicked()
{
switch (m_state) {
case TransferState::Downloading:
case TransferState::Uploading:
// 暂停
setState(TransferState::Paused);
emit pauseRequested(_msg_info->_unique_name, _msg_info->_transfer_type);
break;
case TransferState::Paused:
// 继续
resumeState(); //
emit resumeRequested(_msg_info->_unique_name, _msg_info->_transfer_type);
break;
case TransferState::Failed:
// 重试
emit resumeRequested(_msg_info->_unique_name, _msg_info->_transfer_type);
break;
default:
// 其他状态可以实现查看大图等功能
break;
}
}
接下来我们响应暂停和继续，这部分逻辑也已经复用之前的逻辑即可
暂停逻辑
void ChatPage::on_clicked_paused(QString unique_name, TransferType transfer_type)
{
UserMgr::GetInstance()->PauseTransFileByName(unique_name);
}
void UserMgr::PauseTransFileByName(QString name) {
std::lock_guard<std::mutex> mtx(_trans_mtx);
auto iter = _name_to_msg_info.find(name);
if (iter == _name_to_msg_info.end()) {
return;
}
iter.value()->_transfer_state = TransferState::Paused;
}
恢复逻辑
void ChatPage::on_clicked_resume(QString unique_name, TransferType transfer_type)
{
UserMgr::GetInstance()->ResumeTransFileByName(unique_name);
//继续发送或者下载
if (transfer_type == TransferType::Upload) {
FileTcpMgr::GetInstance()->ContinueUploadFile(unique_name);
return;
}
if (transfer_type == TransferType::Download) {
FileTcpMgr::GetInstance()->ContinueDownloadFile(unique_name);
return;
}
}
void UserMgr::ResumeTransFileByName(QString name)
{
std::lock_guard<std::mutex> mtx(_trans_mtx);
auto iter = _name_to_msg_info.find(name);
if (iter == _name_to_msg_info.end()) {
return;
}
if (iter.value()->_transfer_type == TransferType::Download) {
iter.value()->_transfer_state = TransferState::Downloading;
return;
}
if (iter.value()->_transfer_type == TransferType::Upload) {
iter.value()->_transfer_state = TransferState::Uploading;
return;
}
}
发送继续下载信号通知FileTcpMgr继续下载
void FileTcpMgr::ContinueDownloadFile(QString unique_name) {
emit sig_continue_download_file(unique_name);
}
FileTcpMgr响应下载逻辑
void FileTcpMgr::slot_continue_download_file(QString unique_name) {
auto file_info = UserMgr::GetInstance()->GetTransFileByName(unique_name);
if (file_info == nullptr) {
return;
}
if (file_info->_current_size >= file_info->_total_size) {
qDebug() << "file has received finished";
return;
}
//组织请求，准备下载
QJsonObject jsonObj_send;
jsonObj_send["name"] = unique_name;
jsonObj_send["seq"] = file_info->_seq;
jsonObj_send["trans_size"] = QString::number(file_info->_current_size);
jsonObj_send["total_size"] = QString::number(file_info->_total_size);
jsonObj_send["token"] = UserMgr::GetInstance()->GetToken();
jsonObj_send["sender_id"] = file_info->_sender;
jsonObj_send["receiver_id"] = file_info->_receiver;
jsonObj_send["message_id"] = file_info->_msg_id;
auto uid = UserMgr::GetInstance()->GetUid();
jsonObj_send["uid"] = uid;
QJsonDocument doc(jsonObj_send);
auto send_data = doc.toJson();
FileTcpMgr::GetInstance()->SendData(ID_IMG_CHAT_DOWN_REQ, send_data);
}
通过上述逻辑可以实现客户端的断点下载和暂停。
效果演示
源码链接
https://gitee.com/secondtonone1/llfcchat
注意第二季分支为Season_2