Impl listening message

rpc/pb_types.h

@@ -33,3 +33,14 @@ typedef struct {
 } DbField_t;
 typedef vector<DbField_t> DbRow_t;
 typedef vector<DbRow_t> DbRows_t;
+
+typedef struct {
+    bool is_self;
+    bool is_group;
+    int32_t type;
+    string id;
+    string xml;
+    string sender;
+    string roomid;
+    string content;
+} WxMsg_t;

spy/receive_msg.cpp

@@ -1,6 +1,26 @@
 #pragma execution_character_set("utf-8")
 
+#include "framework.h"
+#include <queue>
+
+#include "load_calls.h"
 #include "receive_msg.h"
+#include "util.h"
+
+// Defined in rpc_server.cpp
+extern bool gIsListening;
+extern HANDLE g_hEvent;
+extern queue<WxMsg_t> gMsgQueue;
+
+// Defined in spy.cpp
+extern WxCalls_t g_WxCalls;
+extern DWORD g_WeChatWinDllAddr;
+
+static DWORD reg_buffer          = 0;
+static DWORD recvMsgHookAddr     = 0;
+static DWORD recvMsgCallAddr     = 0;
+static DWORD recvMsgJumpBackAddr = 0;
+static CHAR recvMsgBackupCode[5] = { 0 };
 
 MsgTypes_t GetMsgTypes()
 {
@@ -26,60 +46,6 @@ MsgTypes_t GetMsgTypes()
     return m;
 }
 
-#if 0
-#include <queue>
-
-#include "framework.h"
-
-#include "load_calls.h"
-#include "receive_msg.h"
-#include "spy_types.h"
-#include "util.h"
-
-using namespace std;
-
-using wcf::MsgTypes;
-using wcf::WxMsg;
-
-extern bool gIsListening;
-extern mutex gMutex;
-extern queue<WxMsg> gMsgQueue;
-extern condition_variable gCv;
-extern WxCalls_t g_WxCalls;
-extern DWORD g_WeChatWinDllAddr;
-
-static DWORD reg_buffer          = 0;
-static DWORD recvMsgHookAddr     = 0;
-static DWORD recvMsgCallAddr     = 0;
-static DWORD recvMsgJumpBackAddr = 0;
-static CHAR recvMsgBackupCode[5] = { 0 };
-
-void GetMsgTypes(MsgTypes *types)
-{
-    const map<int32_t, string> m = { { 0x01, "文字" },
-                                     { 0x03, "图片" },
-                                     { 0x22, "语音" },
-                                     { 0x25, "好友确认" },
-                                     { 0x28, "POSSIBLEFRIEND_MSG" },
-                                     { 0x2A, "名片" },
-                                     { 0x2B, "视频" },
-                                     { 0x2F, "石头剪刀布 | 表情图片" },
-                                     { 0x30, "位置" },
-                                     { 0x31, "共享实时位置、文件、转账、链接" },
-                                     { 0x32, "VOIPMSG" },
-                                     { 0x33, "微信初始化" },
-                                     { 0x34, "VOIPNOTIFY" },
-                                     { 0x35, "VOIPINVITE" },
-                                     { 0x3E, "小视频" },
-                                     { 0x270F, "SYSNOTICE" },
-                                     { 0x2710, "红包、系统消息" },
-                                     { 0x2712, "撤回消息" } };
-
-    for (auto &[k, v] : m) {
-        (*(types->mutable_types()))[k] = v;
-    }
-}
-
 void HookAddress(DWORD hookAddr, LPVOID funcAddr, CHAR recvMsgBackupCode[5])
 {
     // 组装跳转数据
@@ -102,32 +68,30 @@ void UnHookAddress(DWORD hookAddr, CHAR restoreCode[5])
 
 void DispatchMsg(DWORD reg)
 {
-    WxMsg wxMsg;
+    WxMsg_t wxMsg;
     DWORD *p = (DWORD *)reg; // 消息结构基址
 
-    wxMsg.set_type(GET_DWORD(*p + g_WxCalls.recvMsg.type));
+    wxMsg.type    = GET_DWORD(*p + g_WxCalls.recvMsg.type);
-    wxMsg.set_is_self(GET_DWORD(*p + g_WxCalls.recvMsg.isSelf));
+    wxMsg.is_self = GET_DWORD(*p + g_WxCalls.recvMsg.isSelf);
-    wxMsg.set_id(GetStringByAddress(*p + g_WxCalls.recvMsg.msgId));
+    wxMsg.id      = GetStringByAddress(*p + g_WxCalls.recvMsg.msgId);
-    wxMsg.set_xml(GetStringByAddress(*p + g_WxCalls.recvMsg.msgXml));
+    wxMsg.xml     = GetStringByAddress(*p + g_WxCalls.recvMsg.msgXml);
 
-    // 群里的系统消息，xml 为空
+    // 群里的系统消息，xml 为空；或者包含 <membercount>
-    if ((wxMsg.xml().empty()) || (strstr(wxMsg.xml().c_str(), "<membercount>") != NULL)) {
+    if ((wxMsg.xml.empty()) || (wxMsg.xml.find("<membercount>") != string::npos)) {
-        wxMsg.set_is_group(true);
+        wxMsg.is_group = true;
-        wxMsg.set_sender(GetStringByAddress(*p + g_WxCalls.recvMsg.wxId));
+        wxMsg.sender   = GetStringByAddress(*p + g_WxCalls.recvMsg.wxId);
-        wxMsg.set_roomid(GetStringByAddress(*p + g_WxCalls.recvMsg.roomId));
+        wxMsg.roomid   = GetStringByAddress(*p + g_WxCalls.recvMsg.roomId);
     } else {
-        wxMsg.set_is_group(false);
+        wxMsg.is_group = false;
-        wxMsg.set_sender(GetStringByAddress(*p + g_WxCalls.recvMsg.roomId));
+        wxMsg.sender   = GetStringByAddress(*p + g_WxCalls.recvMsg.roomId);
     }
-    wxMsg.set_content(GetStringByAddress(*p + g_WxCalls.recvMsg.content));
+    wxMsg.content = GetStringByAddress(*p + g_WxCalls.recvMsg.content);
 
     // 推送到队列
-    unique_lock<std::mutex> lock(gMutex);
     gMsgQueue.push(wxMsg);
-    lock.unlock();
 
     // 通知各方消息就绪
-    gCv.notify_all();
+    SetEvent(g_hEvent);
 }
 
 __declspec(naked) void RecieveMsgFunc()
@@ -147,6 +111,7 @@ __declspec(naked) void RecieveMsgFunc()
 
 void ListenMessage()
 {
+    // OutputDebugString(L"ListenMessage\n");
     // MessageBox(NULL, L"ListenMessage", L"ListenMessage", 0);
     if (gIsListening || (g_WeChatWinDllAddr == 0)) {
         return;
@@ -168,4 +133,3 @@ void UnListenMessage()
     UnHookAddress(recvMsgHookAddr, recvMsgBackupCode);
     gIsListening = false;
 }
-#endif

spy/receive_msg.h

@@ -2,14 +2,6 @@
 
 #include "pb_types.h"
 
-typedef std::map<int, std::string> MsgTypes_t;
-
-MsgTypes_t GetMsgTypes();
-
-#if 0
-#include "../proto/wcf.grpc.pb.h"
-
 void ListenMessage();
 void UnListenMessage();
-void GetMsgTypes(wcf::MsgTypes *types);
+MsgTypes_t GetMsgTypes();
-#endif

spy/rpc_server.cpp

@@ -17,6 +17,7 @@
 #include "exec_sql.h"
 #include "get_contacts.h"
 #include "log.h"
+#include "pb_types.h"
 #include "pb_util.h"
 #include "receive_msg.h"
 #include "rpc_server.h"
@@ -27,15 +28,12 @@
 
 #define G_BUF_SIZE (16 * 1024 * 1024)
 
-extern int IsLogin(void);         // Defined in spy.cpp
+extern int IsLogin(void);    // Defined in spy.cpp
-extern std::string GetSelfWxid(); // Defined in spy.cpp
+extern string GetSelfWxid(); // Defined in spy.cpp
 
-using namespace std;
-
-mutex gMutex;
-queue<WxMsg> gMsgQueue;
-condition_variable gCv;
 bool gIsListening;
+HANDLE g_hEvent; // New message signal
+queue<WxMsg_t> gMsgQueue;
 
 static DWORD lThreadId = 0;
 static bool lIsRunning = false;
@@ -206,6 +204,105 @@ bool func_send_img(char *path, char *receiver, uint8_t *out, size_t *len)
     return true;
 }
 
+static void PushMessage()
+{
+    static uint8_t buffer[1024] = { 0 };
+
+    int rv;
+    Response rsp  = Response_init_default;
+    rsp.func      = Functions_FUNC_ENABLE_RECV_TXT;
+    rsp.which_msg = Response_wxmsg_tag;
+
+    pb_ostream_t stream = pb_ostream_from_buffer(buffer, 1024);
+
+    while (gIsListening) {
+        // 中断式，兼顾及时性和CPU使用率
+        rv = WaitForSingleObject(g_hEvent, 1000); // 等待消息，每秒检查一次条件
+        if (rv == WAIT_TIMEOUT) {
+            continue;
+        } else if (rv != WAIT_OBJECT_0) {
+            LOG_INFO("WaitForSingleObject ERRIR[{}]: {}", rv, GetLastError());
+            continue;
+        }
+
+        while (!gMsgQueue.empty()) {
+            auto wxmgs = gMsgQueue.front();
+            rsp.msg.wxmsg.is_self  = wxmgs.is_self;
+            rsp.msg.wxmsg.is_group = wxmgs.is_group;
+            rsp.msg.wxmsg.type     = wxmgs.type;
+            rsp.msg.wxmsg.id       = (char *)wxmgs.id.c_str();
+            rsp.msg.wxmsg.xml      = (char *)wxmgs.xml.c_str();
+            rsp.msg.wxmsg.sender   = (char *)wxmgs.sender.c_str();
+            rsp.msg.wxmsg.roomid   = (char *)wxmgs.roomid.c_str();
+            rsp.msg.wxmsg.content  = (char *)wxmgs.content.c_str();
+            gMsgQueue.pop();
+
+            if (!pb_encode(&stream, Response_fields, &rsp)) {
+                LOG_ERROR("Encoding failed: {}", PB_GET_ERROR(&stream));
+                continue;
+            }
+            rv = nng_send(sock, buffer, stream.bytes_written, 0);
+            if (rv != 0) {
+                LOG_ERROR("nng_send: {}", rv);
+            }
+        }
+        ResetEvent(g_hEvent);
+    }
+}
+
+bool func_enable_recv_txt(uint8_t *out, size_t *len)
+{
+    Response rsp   = Response_init_default;
+    rsp.func       = Functions_FUNC_ENABLE_RECV_TXT;
+    rsp.which_msg  = Response_status_tag;
+    rsp.msg.status = -1;
+
+    g_hEvent = CreateEvent(NULL,  // default security attributes
+                           TRUE,  // manual-reset event
+                           FALSE, // initial state is nonsignaled
+                           NULL   // unnamed
+    );
+    if (g_hEvent == NULL) {
+        LOG_ERROR("CreateMutex error: {}", GetLastError());
+        return false;
+    }
+
+    ListenMessage();
+    HANDLE msgThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)PushMessage, NULL, NULL, NULL);
+    if (msgThread != 0) {
+        CloseHandle(msgThread);
+        rsp.msg.status = 0;
+    }
+
+    pb_ostream_t stream = pb_ostream_from_buffer(out, *len);
+    if (!pb_encode(&stream, Response_fields, &rsp)) {
+        LOG_ERROR("Encoding failed: {}", PB_GET_ERROR(&stream));
+        return false;
+    }
+    *len = stream.bytes_written;
+
+    return true;
+}
+
+bool func_disable_recv_txt(uint8_t *out, size_t *len)
+{
+    Response rsp   = Response_init_default;
+    rsp.func       = Functions_FUNC_DISABLE_RECV_TXT;
+    rsp.which_msg  = Response_status_tag;
+    rsp.msg.status = 0;
+
+    UnListenMessage(); // 可能需要1秒之后才能退出，见 PushMessage
+
+    pb_ostream_t stream = pb_ostream_from_buffer(out, *len);
+    if (!pb_encode(&stream, Response_fields, &rsp)) {
+        LOG_ERROR("Encoding failed: {}", PB_GET_ERROR(&stream));
+        return false;
+    }
+    *len = stream.bytes_written;
+
+    return true;
+}
+
 bool func_exec_db_query(char *db, char *sql, uint8_t *out, size_t *len)
 {
     Response rsp  = Response_init_default;
@@ -279,6 +376,16 @@ static bool dispatcher(uint8_t *in, size_t in_len, uint8_t *out, size_t *out_len
             ret = func_send_img(req.msg.img.path, req.msg.txt.receiver, out, out_len);
             break;
         }
+        case Functions_FUNC_ENABLE_RECV_TXT: {
+            LOG_INFO("[Functions_FUNC_ENABLE_RECV_TXT]");
+            ret = func_enable_recv_txt(out, out_len);
+            break;
+        }
+        case Functions_FUNC_DISABLE_RECV_TXT: {
+            LOG_INFO("[Functions_FUNC_DISABLE_RECV_TXT]");
+            ret = func_disable_recv_txt(out, out_len);
+            break;
+        }
         case Functions_FUNC_EXEC_DB_QUERY: {
             LOG_INFO("[Functions_FUNC_EXEC_DB_QUERY]");
             ret = func_exec_db_query(req.msg.query.db, req.msg.query.sql, out, out_len);
@@ -361,9 +468,10 @@ int RpcStopServer()
 {
     if (lIsRunning) {
         nng_close(sock);
-        // UnListenMessage();  // Do it in RpcDisableRecvMsg
+        UnListenMessage();
-        LOG_INFO("Server stoped.");
         lIsRunning = false;
+        Sleep(1000);
+        LOG_INFO("Server stoped.");
     }
     return 0;
 }