Skip to content

1 #734

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
lidaobing wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

1 #734

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 13 additions & 4 deletions src/api/iptux-core/CoreThread.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,10 +155,19 @@ class CoreThread {
* @return true if send success
* @return false if send failed
*/
bool SendMessage(CPPalInfo pal, const std::string& message);
bool SendMessage(CPPalInfo pal, const ChipData& chipData);
bool SendMsgPara(std::shared_ptr<MsgPara> msgPara);
void AsyncSendMsgPara(std::shared_ptr<MsgPara> msgPara);
bool SendMessage(CPPalInfo pal,
const std::string& message,
GError** error = nullptr);
bool SendMessage(CPPalInfo pal,
const ChipData& chipData,
GError** error = nullptr);
bool SendMsgPara(std::shared_ptr<MsgPara> msgPara, GError** error = nullptr);
void sendMsgParaAsync(std::shared_ptr<MsgPara> msgPara,
GCancellable* cancellable,
GAsyncReadyCallback callback,
gpointer user_data);
gboolean sendMsgParaFinish(GAsyncResult* result, GError** error);

void SendUnitMessage(const PalKey& palKey,
uint32_t opttype,
const std::string& message);
Expand Down
63 changes: 46 additions & 17 deletions src/iptux-core/CoreThread.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1096,30 +1096,28 @@ void CoreThread::AddBlockIp(in_addr ipv4) {
g_slist_append(pImpl->blacklist, GUINT_TO_POINTER(ipv4.s_addr));
}

bool CoreThread::SendMessage(CPPalInfo palInfo, const string& message) {
bool CoreThread::SendMessage(CPPalInfo palInfo,
const string& message,
GError** error) {
Command cmd(*this);
cmd.SendMessage(getUdpSock(), palInfo, message.c_str());
return true;
return cmd.SendMessage(getUdpSock(), palInfo, message.c_str(), error);
}

bool CoreThread::SendMessage(CPPalInfo pal, const ChipData& chipData) {
bool CoreThread::SendMessage(CPPalInfo pal,
const ChipData& chipData,
GError** error) {
auto ptr = chipData.data.c_str();
bool ret = true;

switch (chipData.type) {
case MessageContentType::STRING:
/* 文本类型 */
return SendMessage(pal, chipData.data);
return SendMessage(pal, chipData.data, error);
case MESSAGE_CONTENT_TYPE_PICTURE: {
GError* error = nullptr;
GSocket* sock = g_socket_new(G_SOCKET_FAMILY_IPV4, G_SOCKET_TYPE_STREAM,
G_SOCKET_PROTOCOL_TCP, &error);
if (error != nullptr) {
LOG_ERROR(_("Fatal Error!!\nFailed to create new socket!\n%s"),
error->message);
g_error_free(error);
G_SOCKET_PROTOCOL_TCP, error);
if (!sock)
return false;
}
ret = Command(*this).SendSublayer(sock, pal, IPTUX_MSGPICOPT, ptr);
g_object_unref(sock);
return ret;
Expand All @@ -1129,19 +1127,50 @@ bool CoreThread::SendMessage(CPPalInfo pal, const ChipData& chipData) {
}
}

bool CoreThread::SendMsgPara(shared_ptr<MsgPara> para) {
bool CoreThread::SendMsgPara(shared_ptr<MsgPara> para, GError** error) {
for (int i = 0; i < int(para->dtlist.size()); ++i) {
if (!SendMessage(para->getPal(), para->dtlist[i])) {
if (!SendMessage(para->getPal(), para->dtlist[i], error)) {
LOG_ERROR("send message failed: %s", para->dtlist[i].ToString().c_str());
return false;
}
}
return true;
}

void CoreThread::AsyncSendMsgPara(std::shared_ptr<MsgPara> msgPara) {
thread t(&CoreThread::SendMsgPara, this, msgPara);
t.detach();
void CoreThread::sendMsgParaAsync(std::shared_ptr<MsgPara> msgPara,
GCancellable* cancellable,
GAsyncReadyCallback callback,
gpointer user_data) {
GTask* task = g_task_new(this, cancellable, callback, user_data);
g_task_set_task_data(task, new std::shared_ptr<MsgPara>(msgPara),
[](gpointer data) {
delete static_cast<std::shared_ptr<MsgPara>*>(data);
});
g_task_run_in_thread(task, [](GTask* task, gpointer source_object,
gpointer task_data, GCancellable*) {
if (g_task_return_error_if_cancelled(task)) {
return;
}
CoreThread* self = static_cast<CoreThread*>(source_object);
std::shared_ptr<MsgPara> msgPara =
*static_cast<std::shared_ptr<MsgPara>*>(task_data);
GError* error = nullptr;

bool success = self->SendMsgPara(msgPara, &error);
if (!success) {
if (!error) {
error =
g_error_new(G_IO_ERROR, G_IO_ERROR_FAILED, "SendMsgPara failed");
}
g_task_return_error(task, error);
} else {
g_task_return_boolean(task, success);
}
});
}

gboolean CoreThread::sendMsgParaFinish(GAsyncResult* result, GError** error) {
return g_task_propagate_boolean(G_TASK(result), error);
}

void CoreThread::InsertMessage(const MsgPara& para) {
Expand Down
43 changes: 23 additions & 20 deletions src/iptux-core/internal/Command.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
#include "config.h"
#include "Command.h"

#include "glib.h"
#include <cinttypes>
#include <fcntl.h>
#include <sys/types.h>
Expand Down Expand Up @@ -235,13 +236,18 @@ void Command::SendDetectPacket(int sock, in_addr ipv4, uint16_t port) {
* @param pal class PalInfo
* @param msg 消息数据
*/
void Command::SendMessage(int sock, CPPalInfo pal, const char* msg) {
bool Command::SendMessage(int sock,
CPPalInfo pal,
const char* msg,
GError** error) {
uint32_t packetno;
uint8_t count;

auto pal2 = coreThread.GetPal(pal->GetKey());
if (!pal2) {
throw Exception(PAL_KEY_NOT_EXIST);
g_set_error_literal(error, G_IO_ERROR, G_IO_ERROR_FAILED, "pal not exist");
LOG_WARN("pal not exist: %s", pal->GetKey().ToString().c_str());
return false;
}

pal2->rpacketn = packetno = packetn; // 此数据包需要检验回复
Expand All @@ -255,10 +261,12 @@ void Command::SendMessage(int sock, CPPalInfo pal, const char* msg) {
count++;
} while (pal->rpacketn == packetno && count < MAX_RETRYTIMES);
if (pal->rpacketn == packetno) {
FeedbackError(
pal, GROUP_BELONG_TYPE_REGULAR,
_("Your pal didn't receive the packet. He or she is offline maybe."));
g_set_error_literal(error, G_IO_ERROR, G_IO_ERROR_FAILED,
"send message failed");
LOG_WARN("send message failed: %s", pal->GetKey().ToString().c_str());
return false;
}
return true;
}

/**
Expand Down Expand Up @@ -505,10 +513,10 @@ void Command::FeedbackError(CPPalInfo pal,
bool Command::SendSublayer(GSocket* sock,
CPPalInfo pal,
uint32_t opttype,
const char* path) {
const char* path,
GError** error) {
LOG_DEBUG("send tcp message to %s, op %d, file %s",
pal->GetKey().ToString().c_str(), int(opttype), path);
GError* error = nullptr;
int fd;
bool ret;

Expand All @@ -521,18 +529,16 @@ bool Command::SendSublayer(GSocket* sock,
GSocketAddress* sockAddr = g_inet_socket_address_new(addr, pal->port());
g_object_unref(addr);

if (!g_socket_connect(sock, sockAddr, nullptr, &error)) {
LOG_WARN("g_socket_connect failed: %s", error->message);
g_error_free(error);
if (!g_socket_connect(sock, sockAddr, nullptr, error)) {
LOG_WARN("g_socket_connect failed: %s", (*error)->message);
g_object_unref(sockAddr);
return false;
}
g_object_unref(sockAddr);

gssize sent = g_socket_send(sock, buf, size, nullptr, &error);
gssize sent = g_socket_send(sock, buf, size, nullptr, error);
if (sent == -1) {
LOG_WARN("g_socket_send failed: %s", error->message);
g_error_free(error);
LOG_WARN("g_socket_send failed: %s", (*error)->message);
return false;
}

Expand All @@ -551,22 +557,19 @@ bool Command::SendSublayer(GSocket* sock,
* @param sock GSocket tcp socket
* @param fd file descriptor
*/
bool Command::SendSublayerData(GSocket* sock, int fd) {
bool Command::SendSublayerData(GSocket* sock, int fd, GError** error) {
ssize_t len;
bool ret = true;
GError* error = nullptr;

do {
if ((len = xread(fd, buf, MAX_UDPLEN)) <= 0)
break;
gssize sent = g_socket_send(sock, buf, len, nullptr, &error);
gssize sent = g_socket_send(sock, buf, len, nullptr, error);
if (sent <= 0) {
if (error) {
LOG_WARN("g_socket_send failed: %s", error->message);
g_error_free(error);
LOG_WARN("g_socket_send failed: %s", (*error)->message);
}
ret = false;
break;
return false;
}
} while (1);
return ret;
Expand Down
10 changes: 7 additions & 3 deletions src/iptux-core/internal/Command.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,10 @@ class Command {
void SendExit(int sock, CPPalInfo pal);
void SendAbsence(int sock, CPPalInfo pal);
void SendDetectPacket(int sock, in_addr ipv4, uint16_t port);
void SendMessage(int sock, CPPalInfo pal, const char* msg);
bool SendMessage(int sock,
CPPalInfo pal,
const char* msg,
GError** error = nullptr);
void SendReply(int sock, CPPalInfo pal, uint32_t packetno);
void SendReply(int sock, const PalKey& pal, uint32_t packetno);
void SendGroupMsg(int sock, CPPalInfo pal, const char* msg);
Expand Down Expand Up @@ -75,14 +78,15 @@ class Command {
bool SendSublayer(GSocket* sock,
CPPalInfo pal,
uint32_t opttype,
const char* path);
const char* path,
GError** error = nullptr);

static std::string encodeFileInfo(const FileInfo& fileInfo);
static std::vector<FileInfo> decodeFileInfos(const std::string& s);

private:
void FeedbackError(CPPalInfo pal, GroupBelongType btype, const char* error);
bool SendSublayerData(GSocket* sock, int fd);
bool SendSublayerData(GSocket* sock, int fd, GError** error = nullptr);
void ConvertEncode(const std::string& encode);
void CreateCommand(uint32_t command, const char* attach);
void CreateIpmsgExtra(const char* extra, const char* encode);
Expand Down
19 changes: 10 additions & 9 deletions src/iptux-gi/iptux-service.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,7 +151,7 @@ gboolean iptux_service_send_message(IptuxService* self,
void iptux_service_send_message_async(IptuxService* self,
IptuxPal* pal,
const gchar* message,
GCancellable*,
GCancellable* cancellable,
GAsyncReadyCallback callback,
gpointer user_data) {
MsgPara::Ptr msgPara = std::make_shared<MsgPara>(*pal->pal_info);
Expand All @@ -162,17 +162,18 @@ void iptux_service_send_message_async(IptuxService* self,
auto core_thread = *(self->core_thread);

msgPara->dtlist.emplace_back(ChipData(std::string(message)));
core_thread->AsyncSendMsgPara(msgPara);

callback(G_OBJECT(self), NULL,
user_data); // Notify that the operation is complete
core_thread->sendMsgParaAsync(msgPara, cancellable, callback, user_data);
return;
}

gboolean iptux_service_send_message_finish(IptuxService*,
GAsyncResult*,
GError**) {
return TRUE;
gboolean iptux_service_send_message_finish(IptuxService* self,
GAsyncResult* result,
GError** error) {
g_return_val_if_fail(
self != nullptr && self->core_thread != nullptr && result != nullptr,
FALSE);
auto& core_thread = *(self->core_thread);
return core_thread->sendMsgParaFinish(result, error);
}

void iptux_service_set_log_level(IptuxService* self, GLogLevelFlags level) {
Expand Down
16 changes: 15 additions & 1 deletion src/iptux/DialogPeer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -396,7 +396,21 @@ bool DialogPeer::SendTextMsg() {

/* 清空缓冲区并发送数据 */
FeedbackMsg(para);
app->getCoreThread()->AsyncSendMsgPara(para);
app->getCoreThread()->sendMsgParaAsync(
para, nullptr,
[](GObject* source_object, GAsyncResult* res, gpointer user_data) {
CoreThread* coreThread =
static_cast<CoreThread*>(static_cast<void*>(source_object));
GError* error = nullptr;
DialogPeer* self = static_cast<DialogPeer*>(user_data);

if (!coreThread->sendMsgParaFinish(res, &error)) {
pop_warning(GTK_WIDGET(self->window), _("Failed to send message: %s"),
error->message);
g_error_free(error);
}
},
this);
return true;
}

Expand Down
Loading