diff --git a/src/api/iptux-core/CoreThread.h b/src/api/iptux-core/CoreThread.h index 8a670dce..8cb410f3 100644 --- a/src/api/iptux-core/CoreThread.h +++ b/src/api/iptux-core/CoreThread.h @@ -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); - void AsyncSendMsgPara(std::shared_ptr 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, GError** error = nullptr); + void sendMsgParaAsync(std::shared_ptr 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); diff --git a/src/iptux-core/CoreThread.cpp b/src/iptux-core/CoreThread.cpp index 23da3a97..0296650e 100644 --- a/src/iptux-core/CoreThread.cpp +++ b/src/iptux-core/CoreThread.cpp @@ -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; @@ -1129,9 +1127,9 @@ bool CoreThread::SendMessage(CPPalInfo pal, const ChipData& chipData) { } } -bool CoreThread::SendMsgPara(shared_ptr para) { +bool CoreThread::SendMsgPara(shared_ptr 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; } @@ -1139,9 +1137,40 @@ bool CoreThread::SendMsgPara(shared_ptr para) { return true; } -void CoreThread::AsyncSendMsgPara(std::shared_ptr msgPara) { - thread t(&CoreThread::SendMsgPara, this, msgPara); - t.detach(); +void CoreThread::sendMsgParaAsync(std::shared_ptr 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), + [](gpointer data) { + delete static_cast*>(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(source_object); + std::shared_ptr msgPara = + *static_cast*>(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) { diff --git a/src/iptux-core/internal/Command.cpp b/src/iptux-core/internal/Command.cpp index fa0a0fb9..ee91455b 100644 --- a/src/iptux-core/internal/Command.cpp +++ b/src/iptux-core/internal/Command.cpp @@ -12,6 +12,7 @@ #include "config.h" #include "Command.h" +#include "glib.h" #include #include #include @@ -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; // 此数据包需要检验回复 @@ -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; } /** @@ -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; @@ -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; } @@ -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; diff --git a/src/iptux-core/internal/Command.h b/src/iptux-core/internal/Command.h index ff074e22..e7b04454 100644 --- a/src/iptux-core/internal/Command.h +++ b/src/iptux-core/internal/Command.h @@ -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); @@ -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 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); diff --git a/src/iptux-gi/iptux-service.cpp b/src/iptux-gi/iptux-service.cpp index 6bb6aa03..1e86e51a 100644 --- a/src/iptux-gi/iptux-service.cpp +++ b/src/iptux-gi/iptux-service.cpp @@ -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(*pal->pal_info); @@ -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) { diff --git a/src/iptux/DialogPeer.cpp b/src/iptux/DialogPeer.cpp index df1cb35f..f8115f45 100644 --- a/src/iptux/DialogPeer.cpp +++ b/src/iptux/DialogPeer.cpp @@ -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(static_cast(source_object)); + GError* error = nullptr; + DialogPeer* self = static_cast(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; }