Shamrock: 调整推送器逻辑

Signed-off-by: WhiteChi <whitechi73@outlook.com>

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/HttpService.kt

@@ -3,333 +3,52 @@ package moe.fuqiuluo.shamrock.remote.service
 
 import com.tencent.qqnt.kernel.nativeinterface.MsgConstant
 import moe.fuqiuluo.shamrock.helper.MessageHelper
-import com.tencent.qqnt.kernel.nativeinterface.MsgElement
 import com.tencent.qqnt.kernel.nativeinterface.MsgRecord
 import moe.fuqiuluo.qqinterface.servlet.GroupSvc
 import moe.fuqiuluo.qqinterface.servlet.MsgSvc
 import io.ktor.client.statement.bodyAsText
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonArray
 import kotlinx.serialization.json.JsonElement
 import kotlinx.serialization.json.JsonPrimitive
-import moe.fuqiuluo.qqinterface.servlet.TicketSvc
 import moe.fuqiuluo.qqinterface.servlet.msg.*
-import moe.fuqiuluo.qqinterface.servlet.msg.convert.toSegments
+import moe.fuqiuluo.shamrock.remote.service.api.HttpTransmitServlet
-import moe.fuqiuluo.shamrock.remote.service.api.HttpPushServlet
-import moe.fuqiuluo.shamrock.remote.service.config.ShamrockConfig
 import moe.fuqiuluo.shamrock.remote.service.data.push.*
 import moe.fuqiuluo.shamrock.tools.*
 import moe.fuqiuluo.shamrock.helper.Level
 import moe.fuqiuluo.shamrock.helper.LogCenter
+import moe.fuqiuluo.shamrock.remote.service.api.GlobalEventTransmitter
 
-internal object HttpService: HttpPushServlet() {
+internal object HttpService: HttpTransmitServlet() {
     private val actionMsgTypes = arrayOf(
         "record", "voice", "video", "markdown"
     )
 
-    override fun pushSelfPrivateSentMsg(
+    override fun submitFlowJob(job: Job) {
-        record: MsgRecord,
+        // HTTP 回调不会触发断连，无需释放之前的JOB
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    ) {
-        pushMsg(
-            record,
-            elements,
-            raw,
-            msgHash,
-            MsgType.Private,
-            MsgSubType.Friend,
-            postType = PostType.MsgSent
-        )
     }
 
-    override fun pushSelfGroupSentMsg(
+    override fun cancelFlowJobs() {
-        record: MsgRecord,
+
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    ) {
-        pushMsg(
-            record,
-            elements,
-            raw,
-            msgHash,
-            MsgType.Group,
-            MsgSubType.NORMAL,
-            postType = PostType.MsgSent
-        )
     }
 
-    override fun pushPrivateMsg(
+    override fun initTransmitter() {
-        record: MsgRecord,
+        submitFlowJob(GlobalScope.launch {
-        elements: List<MsgElement>,
+            GlobalEventTransmitter.onMessageEvent { (record, event) ->
-        raw: String,
+                val respond = pushTo(event) ?: return@onMessageEvent
-        msgHash: Int
+                handleQuicklyReply(record, event.messageId, respond.bodyAsText())
-    ) {
-        pushMsg(
-            record,
-            elements,
-            raw,
-            msgHash,
-            MsgType.Private,
-            MsgSubType.Friend
-        )
-    }
-
-    override fun pushGroupMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    ) {
-        pushMsg(
-            record, elements, raw, msgHash, MsgType.Group, MsgSubType.NORMAL,
-            role = when (record.senderUin) {
-                GroupSvc.getOwner(record.peerUin.toString()) -> MemberRole.Owner
-                in GroupSvc.getAdminList(record.peerUin.toString()) -> MemberRole.Admin
-                else -> MemberRole.Member
             }
-        )
+        })
-    }
+        submitFlowJob(GlobalScope.launch {
-
+            GlobalEventTransmitter.onNoticeEvent { event ->
-    override fun pushGroupPoke(time: Long, operation: Long, userId: Long, groupId: Long) {
+                pushTo(event)
-        pushNotice(
+            }
-            time = time,
+        })
-            type = NoticeType.Notify,
+        LogCenter.log("HttpService: 初始化服务", Level.WARN)
-            subType = NoticeSubType.Poke,
-            operation = operation,
-            userId = operation,
-            groupId = groupId,
-            target = userId,
-        )
-    }
-
-    override fun pushPrivateMsgRecall(time: Long, operation: Long, msgHash: Int, tip: String) {
-        pushNotice(
-            time = time,
-            type = NoticeType.FriendRecall,
-            operation = operation,
-            userId = operation,
-            msgHash = msgHash,
-            tip = tip
-        )
-    }
-
-    override fun pushGroupMsgRecall(
-        time: Long,
-        operation: Long,
-        userId: Long,
-        groupId: Long,
-        msgHash: Int,
-        tip: String
-    ) {
-        pushNotice(
-            time = time,
-            type = NoticeType.GroupRecall,
-            operation = operation,
-            userId = userId,
-            groupId = groupId,
-            msgHash = msgHash,
-            tip = tip
-        )
-    }
-
-    override fun pushGroupBan(
-        time: Long,
-        operation: Long,
-        userId: Long,
-        groupId: Long,
-        duration: Int
-    ) {
-        pushNotice(
-            time,
-            NoticeType.GroupBan,
-            if (duration == 0) NoticeSubType.LiftBan else NoticeSubType.Ban,
-            operation,
-            userId,
-            groupId,
-            duration
-        )
-    }
-
-    override fun pushGroupMemberDecreased(
-        time: Long,
-        target: Long,
-        groupId: Long,
-        operation: Long,
-        type: NoticeType,
-        subType: NoticeSubType
-    ) {
-        pushNotice(
-            time,
-            type,
-            subType,
-            operation,
-            target,
-            groupId
-        )
-    }
-
-    override fun pushGroupAdminChange(time: Long, target: Long, groupId: Long, setAdmin: Boolean) {
-        pushNotice(
-            time,
-            NoticeType.GroupAdminChange,
-            if (setAdmin) NoticeSubType.Set else NoticeSubType.UnSet,
-            0,
-            target,
-            groupId
-        )
-    }
-
-    override fun pushGroupFileCome(
-        time: Long,
-        userId: Long,
-        groupId: Long,
-        fileId: String,
-        fileName: String,
-        fileSize: Long,
-        bizId: Int,
-        url: String
-    ) {
-        pushNotice(
-            time = time,
-            type = NoticeType.GroupUpload,
-            groupId = groupId,
-            operation = userId,
-            userId = userId,
-            groupFileMsg = GroupFileMsg(
-                id = fileId,
-                name = fileName,
-                size = fileSize,
-                busid = bizId.toLong(),
-                url = url
-            )
-        )
-    }
-
-    override fun pushC2CPoke(time: Long, userId: Long, targetId: Long) {
-        pushNotice(
-            time = time,
-            type = NoticeType.Notify,
-            subType = NoticeSubType.Poke,
-            operation = userId,
-            userId = userId,
-            target = targetId,
-            sender = userId
-        )
-    }
-
-    override fun pushC2CFileCome(
-        time: Long,
-        sender: Long,
-        fileId: String,
-        fileSubId: String,
-        fileName: String,
-        fileSize: Long,
-        expireTime: Long,
-        url: String
-    ) {
-        pushNotice(
-            time = time,
-            type = NoticeType.PrivateUpload,
-            operation = sender,
-            userId = sender,
-            sender = sender,
-            privateFileMsg = PrivateFileMsg(
-                id = fileId,
-                name = fileName,
-                size = fileSize,
-                url = url,
-                subId = fileSubId,
-                expire = expireTime
-            )
-        )
-    }
-
-    private fun pushNotice(
-        time: Long,
-        type: NoticeType,
-        subType: NoticeSubType = NoticeSubType.None,
-        operation: Long,
-        userId: Long,
-        groupId: Long = 0,
-        duration: Int = 0,
-        msgHash: Int = 0,
-        target: Long = 0,
-        sender: Long = 0,
-        tip: String = "",
-        groupFileMsg: GroupFileMsg? = null,
-        privateFileMsg: PrivateFileMsg? = null
-    ) {
-        GlobalScope.launch {
-            pushTo(PushNotice(
-                time = time,
-                selfId = app.longAccountUin,
-                postType = PostType.Notice,
-                type = type,
-                subType = subType,
-                operatorId = operation,
-                userId = userId,
-                groupId = groupId,
-                duration = duration,
-                target = target,
-                msgId = msgHash,
-                tip = tip,
-                file = groupFileMsg,
-                senderId = sender,
-                privateFile = privateFileMsg
-            ))
-        }
-    }
-
-    private fun pushMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int,
-        msgType: MsgType,
-        subType: MsgSubType,
-        role: MemberRole = MemberRole.Member,
-        postType: PostType = PostType.Msg
-    ) {
-        val uin = TicketSvc.getUin().toLong()
-        GlobalScope.launch {
-            val respond = pushTo(PushMessage(
-                time = record.msgTime,
-                selfId = app.longAccountUin,
-                postType = postType,
-                messageType = msgType,
-                subType = subType,
-                messageId = msgHash,
-                groupId = if(msgType == MsgType.Private) 0 else record.peerUin,
-                targetId = if(msgType != MsgType.Private) 0 else record.peerUin,
-                peerId = if (record.senderUin == uin) record.peerUin else uin,
-                userId = record.senderUin,
-                message = if(ShamrockConfig.useCQ()) raw.json
-                else elements.toSegments(record.chatType, record.peerUin.toString()).map {
-                    it.toJson()
-                }.json,
-                rawMessage = raw,
-                font = 0,
-                sender = Sender(
-                    userId = record.senderUin,
-                    nickname = record.sendNickName,
-                    card = record.sendMemberName,
-                    role = role,
-                    title = "",
-                    level = "",
-                )
-            )) ?: return@launch
-            handleQuicklyReply(
-                record,
-                msgHash,
-                respond.bodyAsText()
-            )
-        }
     }
 
     private suspend fun handleQuicklyReply(record: MsgRecord, msgHash: Int, jsonText: String) {

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/WebSocketClientService.kt

@@ -6,6 +6,7 @@ import com.tencent.qqnt.kernel.nativeinterface.MsgElement
 import com.tencent.qqnt.kernel.nativeinterface.MsgRecord
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
 import moe.fuqiuluo.shamrock.remote.service.api.WebSocketClientServlet
 import moe.fuqiuluo.shamrock.remote.service.config.ShamrockConfig
@@ -14,295 +15,39 @@ import moe.fuqiuluo.shamrock.tools.json
 import moe.fuqiuluo.qqinterface.servlet.GroupSvc
 import moe.fuqiuluo.qqinterface.servlet.TicketSvc
 import moe.fuqiuluo.qqinterface.servlet.msg.convert.toSegments
+import moe.fuqiuluo.shamrock.helper.Level
+import moe.fuqiuluo.shamrock.helper.LogCenter
+import moe.fuqiuluo.shamrock.remote.service.api.GlobalEventTransmitter
 
 internal class WebSocketClientService(
     override val address: String,
     wsHeaders: Map<String, String>
 ) : WebSocketClientServlet(address, wsHeaders) {
+    private val eventJobList = mutableSetOf<Job>()
 
-    override fun pushSelfPrivateSentMsg(
+    override fun submitFlowJob(job: Job) {
-        record: MsgRecord,
+        eventJobList.add(job)
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    ) {
-        pushMsg(
-            record,
-            elements,
-            raw,
-            msgHash,
-            MsgType.Private,
-            MsgSubType.Friend,
-            postType = PostType.MsgSent
-        )
     }
 
-    override fun pushSelfGroupSentMsg(
+    override fun initTransmitter() {
-        record: MsgRecord,
+        HttpService.submitFlowJob(GlobalScope.launch {
-        elements: List<MsgElement>,
+            GlobalEventTransmitter.onMessageEvent { (_, event) ->
-        raw: String,
+                pushTo(event)
-        msgHash: Int
-    ) {
-        pushMsg(
-            record,
-            elements,
-            raw,
-            msgHash,
-            MsgType.Group,
-            MsgSubType.NORMAL,
-            postType = PostType.MsgSent
-        )
-    }
-
-    override fun pushPrivateMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    ) {
-        pushMsg(record, elements, raw, msgHash, MsgType.Private, MsgSubType.Friend)
-    }
-
-    override fun pushGroupMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    ) {
-        pushMsg(
-            record, elements, raw, msgHash, MsgType.Group, MsgSubType.NORMAL,
-            role = when (record.senderUin) {
-                GroupSvc.getOwner(record.peerUin.toString()) -> MemberRole.Owner
-                in GroupSvc.getAdminList(record.peerUin.toString()) -> MemberRole.Admin
-                else -> MemberRole.Member
             }
-        )
+        })
+        HttpService.submitFlowJob(GlobalScope.launch {
+            GlobalEventTransmitter.onNoticeEvent { event ->
+                pushTo(event)
+            }
+        })
+        LogCenter.log("WebSocketClientService: 初始化服务", Level.WARN)
     }
 
-    override fun pushGroupPoke(time: Long, operation: Long, userId: Long, groupId: Long) {
+    override fun cancelFlowJobs() {
-        pushNotice(
+        eventJobList.removeIf { job ->
-            time = time,
+            job.cancel()
-            type = NoticeType.Notify,
+            return@removeIf true
-            subType = NoticeSubType.Poke,
-            operation = operation,
-            userId = operation,
-            groupId = groupId,
-            target = userId
-        )
-    }
-
-    override fun pushPrivateMsgRecall(time: Long, operation: Long, msgHash: Int, tip: String) {
-        pushNotice(
-            time = time,
-            type = NoticeType.FriendRecall,
-            operation = operation,
-            userId = operation,
-            msgHash = msgHash,
-            tip = tip
-        )
-    }
-
-    override fun pushGroupMsgRecall(
-        time: Long,
-        operation: Long,
-        userId: Long,
-        groupId: Long,
-        msgHash: Int,
-        tip: String
-    ) {
-        pushNotice(
-            time = time,
-            type = NoticeType.GroupRecall,
-            operation = operation,
-            userId = userId,
-            groupId = groupId,
-            msgHash = msgHash,
-            tip = tip
-        )
-    }
-
-    override fun pushGroupBan(
-        time: Long,
-        operation: Long,
-        userId: Long,
-        groupId: Long,
-        duration: Int
-    ) {
-        pushNotice(
-            time,
-            NoticeType.GroupBan,
-            if (duration == 0) NoticeSubType.LiftBan else NoticeSubType.Ban,
-            operation,
-            userId,
-            groupId,
-            duration
-        )
-    }
-
-    override fun pushGroupMemberDecreased(
-        time: Long,
-        target: Long,
-        groupId: Long,
-        operation: Long,
-        type: NoticeType,
-        subType: NoticeSubType
-    ) {
-        pushNotice(time, type, subType, operation, target, groupId)
-    }
-
-    override fun pushGroupAdminChange(time: Long, target: Long, groupId: Long, setAdmin: Boolean) {
-        pushNotice(
-            time,
-            NoticeType.GroupAdminChange,
-            if (setAdmin) NoticeSubType.Set else NoticeSubType.UnSet,
-            0,
-            target,
-            groupId
-        )
-    }
-
-    override fun pushGroupFileCome(
-        time: Long,
-        userId: Long,
-        groupId: Long,
-        fileId: String,
-        fileName: String,
-        fileSize: Long,
-        bizId: Int,
-        url: String
-    ) {
-        pushNotice(
-            time = time,
-            type = NoticeType.GroupUpload,
-            groupId = groupId,
-            operation = userId,
-            userId = userId,
-            groupFileMsg = GroupFileMsg(
-                id = fileId,
-                name = fileName,
-                size = fileSize,
-                busid = bizId.toLong(),
-                url = url
-            )
-        )
-    }
-
-    override fun pushC2CPoke(time: Long, userId: Long, targetId: Long) {
-        pushNotice(
-            time = time,
-            type = NoticeType.Notify,
-            subType = NoticeSubType.Poke,
-            operation = userId,
-            userId = userId,
-            target = targetId,
-            sender = userId
-        )
-    }
-
-    override fun pushC2CFileCome(
-        time: Long,
-        sender: Long,
-        fileId: String,
-        fileSubId: String,
-        fileName: String,
-        fileSize: Long,
-        expireTime: Long,
-        url: String
-    ) {
-        pushNotice(
-            time = time,
-            type = NoticeType.PrivateUpload,
-            operation = sender,
-            userId = sender,
-            sender = sender,
-            privateFileMsg = PrivateFileMsg(
-                id = fileId,
-                name = fileName,
-                size = fileSize,
-                url = url,
-                subId = fileSubId,
-                expire = expireTime
-            )
-        )
-    }
-
-    private fun pushNotice(
-        time: Long,
-        type: NoticeType,
-        subType: NoticeSubType = NoticeSubType.None,
-        operation: Long,
-        userId: Long,
-        groupId: Long = 0,
-        duration: Int = 0,
-        msgHash: Int = 0,
-        target: Long = 0,
-        sender: Long = 0,
-        tip: String = "",
-        groupFileMsg: GroupFileMsg? = null,
-        privateFileMsg: PrivateFileMsg? = null
-    ) {
-        GlobalScope.launch {
-            pushTo(
-                PushNotice(
-                    time = time,
-                    selfId = app.longAccountUin,
-                    postType = PostType.Notice,
-                    type = type,
-                    subType = subType,
-                    operatorId = operation,
-                    userId = userId,
-                    groupId = groupId,
-                    duration = duration,
-                    target = target,
-                    msgId = msgHash,
-                    tip = tip,
-                    file = groupFileMsg,
-                    senderId = sender,
-                    privateFile = privateFileMsg
-                )
-            )
-        }
-    }
-
-    private fun pushMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int,
-        msgType: MsgType,
-        subType: MsgSubType,
-        role: MemberRole = MemberRole.Member,
-        postType: PostType = PostType.Msg
-    ) {
-        val uin = TicketSvc.getUin().toLong()
-        GlobalScope.launch {
-            pushTo(
-                PushMessage(
-                    time = record.msgTime,
-                    selfId = app.longAccountUin,
-                    postType = postType,
-                    messageType = msgType,
-                    subType = subType,
-                    messageId = msgHash,
-                    groupId = if (msgType == MsgType.Private) 0 else record.peerUin,
-                    targetId = if (msgType != MsgType.Private) 0 else record.peerUin,
-                    peerId = if (record.senderUin == uin) record.peerUin else uin,
-                    userId = record.senderUin,
-                    message = if (ShamrockConfig.useCQ()) raw.json else elements.toSegments(record.chatType, record.peerUin.toString()).map {
-                        it.toJson()
-                    }.json,
-                    rawMessage = raw,
-                    font = 0,
-                    sender = Sender(
-                        userId = record.senderUin,
-                        nickname = record.sendNickName,
-                        card = record.sendMemberName,
-                        role = role,
-                        title = "",
-                        level = "",
-                    )
-                )
-            )
         }
+        LogCenter.log("WebSocketClientService: 释放服务", Level.WARN)
     }
 }

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/WebSocketService.kt

@@ -2,316 +2,52 @@
 
 package moe.fuqiuluo.shamrock.remote.service
 
-import com.tencent.qqnt.kernel.nativeinterface.MsgElement
-import com.tencent.qqnt.kernel.nativeinterface.MsgRecord
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
-import moe.fuqiuluo.qqinterface.servlet.GroupSvc
-import moe.fuqiuluo.qqinterface.servlet.TicketSvc
-import moe.fuqiuluo.qqinterface.servlet.msg.convert.toSegments
 import moe.fuqiuluo.shamrock.helper.ErrorTokenException
-import moe.fuqiuluo.shamrock.remote.service.api.WebSocketPushServlet
+import moe.fuqiuluo.shamrock.remote.service.api.WebSocketTransmitServlet
 import moe.fuqiuluo.shamrock.remote.service.config.ShamrockConfig
 import moe.fuqiuluo.shamrock.remote.service.data.BotStatus
 import moe.fuqiuluo.shamrock.remote.service.data.Self
 import moe.fuqiuluo.shamrock.remote.service.data.push.*
 import moe.fuqiuluo.shamrock.tools.ifNullOrEmpty
-import moe.fuqiuluo.shamrock.tools.json
 import moe.fuqiuluo.shamrock.helper.Level
 import moe.fuqiuluo.shamrock.helper.LogCenter
+import moe.fuqiuluo.shamrock.remote.service.api.GlobalEventTransmitter
 import moe.fuqiuluo.shamrock.xposed.helper.AppRuntimeFetcher
-import mqq.app.MobileQQ
 import org.java_websocket.WebSocket
 import org.java_websocket.handshake.ClientHandshake
 import java.net.URI
 
-internal class WebSocketService(port: Int): WebSocketPushServlet(port) {
+internal class WebSocketService(port: Int): WebSocketTransmitServlet(port) {
-    fun pushMetaLifecycle() {
+    private val eventJobList = mutableSetOf<Job>()
-        GlobalScope.launch {
+
-            val runtime = AppRuntimeFetcher.appRuntime
+    override fun submitFlowJob(job: Job) {
-            val curUin = runtime.currentAccountUin
+        eventJobList.add(job)
-            pushTo(PushMetaEvent(
-                time = System.currentTimeMillis() / 1000,
-                selfId = app.longAccountUin,
-                postType = PostType.Meta,
-                type = MetaEventType.LifeCycle,
-                subType = MetaSubType.Connect,
-                status = BotStatus(
-                    Self("qq", curUin.toLong()), runtime.isLogin, status = "正常", good = true
-                ),
-                interval = 15000
-            ))
-        }
     }
 
-    override fun pushSelfPrivateSentMsg(
+    override fun initTransmitter() {
-        record: MsgRecord,
+        HttpService.submitFlowJob(GlobalScope.launch {
-        elements: List<MsgElement>,
+            GlobalEventTransmitter.onMessageEvent { (_, event) ->
-        raw: String,
+                pushTo(event)
-        msgHash: Int
-    ) {
-        pushMsg(
-            record,
-            elements,
-            raw,
-            msgHash,
-            MsgType.Private,
-            MsgSubType.Friend,
-            postType = PostType.MsgSent
-        )
-    }
-
-    override fun pushSelfGroupSentMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    ) {
-        pushMsg(
-            record,
-            elements,
-            raw,
-            msgHash,
-            MsgType.Group,
-            MsgSubType.NORMAL,
-            postType = PostType.MsgSent
-        )
-    }
-
-    override fun pushPrivateMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    ) {
-        pushMsg(record, elements, raw, msgHash, MsgType.Private, MsgSubType.Friend)
-    }
-
-    override fun pushGroupMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    ) {
-        pushMsg(
-            record, elements, raw, msgHash, MsgType.Group, MsgSubType.NORMAL,
-            role = when (record.senderUin) {
-                GroupSvc.getOwner(record.peerUin.toString()) -> MemberRole.Owner
-                in GroupSvc.getAdminList(record.peerUin.toString()) -> MemberRole.Admin
-                else -> MemberRole.Member
             }
-        )
+        })
+        HttpService.submitFlowJob(GlobalScope.launch {
+            GlobalEventTransmitter.onNoticeEvent { event ->
+                pushTo(event)
+            }
+        })
+        LogCenter.log("WebSocketService: 初始化服务", Level.WARN)
     }
 
-    override fun pushGroupPoke(time: Long, operation: Long, userId: Long, groupId: Long) {
+    override fun cancelFlowJobs() {
-        pushNotice(
+        eventJobList.removeIf { job ->
-            time = time,
+            job.cancel()
-            type = NoticeType.Notify,
+            return@removeIf true
-            subType = NoticeSubType.Poke,
-            operation = operation,
-            userId = operation,
-            groupId = groupId,
-            target = userId
-        )
-    }
-
-    override fun pushPrivateMsgRecall(time: Long, operation: Long, msgHash: Int, tip: String) {
-        pushNotice(
-            time = time,
-            type = NoticeType.FriendRecall,
-            operation = operation,
-            userId = operation,
-            msgHash = msgHash,
-            tip = tip
-        )
-    }
-
-    override fun pushGroupMsgRecall(
-        time: Long,
-        operation: Long,
-        userId: Long,
-        groupId: Long,
-        msgHash: Int,
-        tip: String
-    ) {
-        pushNotice(
-            time = time,
-            type = NoticeType.GroupRecall,
-            operation = operation,
-            userId = userId,
-            groupId =  groupId,
-            msgHash = msgHash,
-            tip = tip
-        )
-    }
-
-    override fun pushGroupBan(
-        time: Long,
-        operation: Long,
-        userId: Long,
-        groupId: Long,
-        duration: Int
-    ) {
-        pushNotice(time, NoticeType.GroupBan, if (duration == 0) NoticeSubType.LiftBan else NoticeSubType.Ban, operation, userId, groupId, duration)
-    }
-
-    override fun pushGroupMemberDecreased(
-        time: Long,
-        target: Long,
-        groupId: Long,
-        operation: Long,
-        type: NoticeType,
-        subType: NoticeSubType
-    ) {
-        pushNotice(time, type, subType, operation, target, groupId)
-    }
-
-    override fun pushGroupAdminChange(time: Long, target: Long, groupId: Long, setAdmin: Boolean) {
-        pushNotice(time, NoticeType.GroupAdminChange, if (setAdmin) NoticeSubType.Set else NoticeSubType.UnSet, 0, target, groupId)
-    }
-
-    override fun pushGroupFileCome(
-        time: Long,
-        userId: Long,
-        groupId: Long,
-        fileId: String,
-        fileName: String,
-        fileSize: Long,
-        bizId: Int,
-        url: String
-    ) {
-        pushNotice(
-            time = time,
-            type = NoticeType.GroupUpload,
-            groupId = groupId,
-            operation = userId,
-            userId = userId,
-            groupFileMsg = GroupFileMsg(
-                id = fileId,
-                name = fileName,
-                size = fileSize,
-                busid = bizId.toLong(),
-                url = url
-            )
-        )
-    }
-
-    override fun pushC2CPoke(time: Long, userId: Long, targetId: Long) {
-        pushNotice(
-            time = time,
-            type = NoticeType.Notify,
-            subType = NoticeSubType.Poke,
-            operation = userId,
-            userId = userId,
-            target = targetId,
-            sender = userId
-        )
-    }
-
-    override fun pushC2CFileCome(
-        time: Long,
-        sender: Long,
-        fileId: String,
-        fileSubId: String,
-        fileName: String,
-        fileSize: Long,
-        expireTime: Long,
-        url: String
-    ) {
-        pushNotice(
-            time = time,
-            type = NoticeType.PrivateUpload,
-            operation = sender,
-            userId = sender,
-            sender = sender,
-            privateFileMsg = PrivateFileMsg(
-                id = fileId,
-                name = fileName,
-                size = fileSize,
-                url = url,
-                subId = fileSubId,
-                expire = expireTime
-            )
-        )
-    }
-
-    private fun pushNotice(
-        time: Long,
-        type: NoticeType,
-        subType: NoticeSubType = NoticeSubType.None,
-        operation: Long,
-        userId: Long,
-        groupId: Long = 0,
-        duration: Int = 0,
-        msgHash: Int = 0,
-        target: Long = 0,
-        sender: Long = 0,
-        tip: String = "",
-        groupFileMsg: GroupFileMsg? = null,
-        privateFileMsg: PrivateFileMsg? = null
-    ) {
-        GlobalScope.launch {
-            pushTo(
-                PushNotice(
-                    time = time,
-                    selfId = app.longAccountUin,
-                    postType = PostType.Notice,
-                    type = type,
-                    subType = subType,
-                    operatorId = operation,
-                    userId = userId,
-                    groupId = groupId,
-                    duration = duration,
-                    target = target,
-                    msgId = msgHash,
-                    tip = tip,
-                    file = groupFileMsg,
-                    senderId = sender,
-                    privateFile = privateFileMsg
-                )
-            )
-        }
-    }
-
-    private fun pushMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int,
-        msgType: MsgType,
-        subType: MsgSubType,
-        role: MemberRole = MemberRole.Member,
-        postType: PostType = PostType.Msg
-    ) {
-        val uin = TicketSvc.getUin().toLong()
-        GlobalScope.launch {
-            pushTo(PushMessage(
-                time = record.msgTime,
-                selfId = app.longAccountUin,
-                postType = postType,
-                messageType = msgType,
-                subType = subType,
-                messageId = msgHash,
-                groupId = if(msgType == MsgType.Private) 0 else record.peerUin,
-                targetId = if(msgType != MsgType.Private) 0 else record.peerUin,
-                peerId = if (record.senderUin == uin) record.peerUin else uin,
-                userId = record.senderUin,
-                message = if (ShamrockConfig.useCQ()) raw.json else elements.toSegments(record.chatType, record.peerUin.toString()).map {
-                    it.toJson()
-                }.json,
-                rawMessage = raw,
-                font = 0,
-                sender = Sender(
-                    userId = record.senderUin,
-                    nickname = record.sendNickName,
-                    card = record.sendMemberName,
-                    role = role,
-                    title = "",
-                    level = "",
-                )
-            ))
         }
+        LogCenter.log("WebSocketService: 释放服务", Level.WARN)
     }
 
     override fun onOpen(conn: WebSocket, handshake: ClientHandshake) {
@@ -337,4 +73,22 @@ internal class WebSocketService(port: Int): WebSocketPushServlet(port) {
         }
         LogCenter.log({ "WSServer连接(${conn.remoteSocketAddress.address.hostAddress}:${conn.remoteSocketAddress.port}$path)" }, Level.DEBUG)
     }
+
+    private fun pushMetaLifecycle() {
+        GlobalScope.launch {
+            val runtime = AppRuntimeFetcher.appRuntime
+            val curUin = runtime.currentAccountUin
+            pushTo(PushMetaEvent(
+                time = System.currentTimeMillis() / 1000,
+                selfId = app.longAccountUin,
+                postType = PostType.Meta,
+                type = MetaEventType.LifeCycle,
+                subType = MetaSubType.Connect,
+                status = BotStatus(
+                    Self("qq", curUin.toLong()), runtime.isLogin, status = "正常", good = true
+                ),
+                interval = 15000
+            ))
+        }
+    }
 }

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/api/BasePushServlet.kt

@@ -1,108 +0,0 @@
-package moe.fuqiuluo.shamrock.remote.service.api
-
-import com.tencent.mobileqq.app.QQAppInterface
-import com.tencent.qqnt.kernel.nativeinterface.MsgElement
-import com.tencent.qqnt.kernel.nativeinterface.MsgRecord
-import moe.fuqiuluo.shamrock.remote.service.data.push.NoticeSubType
-import moe.fuqiuluo.shamrock.remote.service.data.push.NoticeType
-import moe.fuqiuluo.shamrock.xposed.helper.AppRuntimeFetcher
-import mqq.app.MobileQQ
-import oicq.wlogin_sdk.tools.MD5
-
-internal interface BasePushServlet {
-    val address: String
-
-    fun allowPush(): Boolean
-
-    fun pushSelfPrivateSentMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    )
-
-    fun pushSelfGroupSentMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    )
-
-    fun pushPrivateMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    )
-
-    fun pushGroupMsg(
-        record: MsgRecord,
-        elements: List<MsgElement>,
-        raw: String,
-        msgHash: Int
-    )
-
-    fun pushGroupPoke(time: Long, operation: Long, userId: Long, groupId: Long)
-
-    //fun pushPrivatePoke(time: Long, operation: Long, userId: Long, sender: Long)
-
-    fun pushPrivateMsgRecall(time: Long, operation: Long, msgHash: Int, tip: String)
-
-    fun pushGroupMsgRecall(
-        time: Long,
-        operation: Long,
-        userId: Long,
-        groupId: Long,
-        msgHash: Int,
-        tip: String
-    )
-
-    fun pushGroupBan(
-        time: Long,
-        operation: Long,
-        userId: Long,
-        groupId: Long,
-        duration: Int
-    )
-
-    fun pushGroupMemberDecreased(
-        time: Long,
-        target: Long,
-        groupId: Long,
-        operation: Long = 0,
-        type: NoticeType,
-        subType: NoticeSubType
-    )
-
-    fun pushGroupAdminChange(time: Long, target: Long, groupId: Long, setAdmin: Boolean)
-
-    fun pushGroupFileCome(
-        time: Long,
-        userId: Long,
-        groupId: Long,
-        fileId: String,
-        fileName: String,
-        fileSize: Long,
-        bizId: Int,
-        url: String
-    )
-
-    fun pushC2CPoke(time: Long, userId: Long, targetId: Long)
-
-    fun pushC2CFileCome(
-        msgTime: Long,
-        sender: Long,
-        fileId: String,
-        fileSubId: String,
-        fileName: String,
-        fileSize: Long,
-        expireTime: Long,
-        url: String
-    )
-
-    val app: QQAppInterface
-        get() = AppRuntimeFetcher.appRuntime as QQAppInterface
-
-    val id: String
-        get() = MD5.getMD5String(address.toByteArray())
-}

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/api/BaseTransmitServlet.kt

@@ -0,0 +1,28 @@
+package moe.fuqiuluo.shamrock.remote.service.api
+
+import com.tencent.mobileqq.app.QQAppInterface
+import com.tencent.qqnt.kernel.nativeinterface.MsgElement
+import com.tencent.qqnt.kernel.nativeinterface.MsgRecord
+import kotlinx.coroutines.Job
+import moe.fuqiuluo.shamrock.remote.service.data.push.NoticeSubType
+import moe.fuqiuluo.shamrock.remote.service.data.push.NoticeType
+import moe.fuqiuluo.shamrock.xposed.helper.AppRuntimeFetcher
+import oicq.wlogin_sdk.tools.MD5
+
+internal interface BaseTransmitServlet {
+    val address: String
+
+    fun allowTransmit(): Boolean
+
+    fun submitFlowJob(job: Job)
+
+    fun cancelFlowJobs()
+
+    fun initTransmitter()
+
+    val app: QQAppInterface
+        get() = AppRuntimeFetcher.appRuntime as QQAppInterface
+
+    val id: String
+        get() = MD5.getMD5String(address.toByteArray())
+}

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/api/GlobalEventTransmitter.kt

@@ -0,0 +1,347 @@
+package moe.fuqiuluo.shamrock.remote.service.api
+
+import com.tencent.qqnt.kernel.nativeinterface.MsgElement
+import com.tencent.qqnt.kernel.nativeinterface.MsgRecord
+import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.flow.MutableSharedFlow
+import moe.fuqiuluo.qqinterface.servlet.BaseSvc
+import moe.fuqiuluo.qqinterface.servlet.GroupSvc
+import moe.fuqiuluo.qqinterface.servlet.msg.convert.toSegments
+import moe.fuqiuluo.shamrock.remote.service.HttpService
+import moe.fuqiuluo.shamrock.remote.service.config.ShamrockConfig
+import moe.fuqiuluo.shamrock.remote.service.data.push.GroupFileMsg
+import moe.fuqiuluo.shamrock.remote.service.data.push.MemberRole
+import moe.fuqiuluo.shamrock.remote.service.data.push.MsgSubType
+import moe.fuqiuluo.shamrock.remote.service.data.push.MsgType
+import moe.fuqiuluo.shamrock.remote.service.data.push.PostType
+import moe.fuqiuluo.shamrock.remote.service.data.push.MessageEvent
+import moe.fuqiuluo.shamrock.remote.service.data.push.NoticeEvent
+import moe.fuqiuluo.shamrock.remote.service.data.push.NoticeSubType
+import moe.fuqiuluo.shamrock.remote.service.data.push.NoticeType
+import moe.fuqiuluo.shamrock.remote.service.data.push.PrivateFileMsg
+import moe.fuqiuluo.shamrock.remote.service.data.push.Sender
+import moe.fuqiuluo.shamrock.tools.ShamrockDsl
+import moe.fuqiuluo.shamrock.tools.json
+import java.util.ArrayList
+
+internal object GlobalEventTransmitter: BaseSvc() {
+    private val messageEventFlow by lazy {
+        MutableSharedFlow<Pair<MsgRecord, MessageEvent>>()
+    }
+    private val noticeEventFlow by lazy {
+        MutableSharedFlow<NoticeEvent>()
+    }
+
+    private fun pushNotice(noticeEvent: NoticeEvent) = noticeEventFlow.tryEmit(noticeEvent)
+
+    private fun transMessageEvent(record: MsgRecord, message: MessageEvent) = messageEventFlow.tryEmit(record to message)
+
+    /**
+     * 消息 手淫器
+     */
+    object MessageTransmitter {
+        /**
+         * 推送群聊消息
+         */
+        suspend fun transGroupMessage(
+            record: MsgRecord,
+            elements: ArrayList<MsgElement>,
+            rawMsg: String,
+            msgHash: Int,
+            postType: PostType = PostType.Msg
+        ): Boolean {
+            val uin = app.longAccountUin
+            return transMessageEvent(record,
+                MessageEvent(
+                    time = record.msgTime,
+                    selfId = uin,
+                    postType = postType,
+                    messageType = MsgType.Group,
+                    subType = MsgSubType.NORMAL,
+                    messageId = msgHash,
+                    groupId = record.peerUin,
+                    peerId = uin,
+                    userId = record.senderUin,
+                    message = if(ShamrockConfig.useCQ()) rawMsg.json
+                    else elements.toSegments(record.chatType, record.peerUin.toString()).map {
+                        it.toJson()
+                    }.json,
+                    rawMessage = rawMsg,
+                    font = 0,
+                    sender = Sender(
+                        userId = record.senderUin,
+                        nickname = record.sendNickName,
+                        card = record.sendMemberName,
+                        role = when (record.senderUin) {
+                            GroupSvc.getOwner(record.peerUin.toString()) -> MemberRole.Owner
+                            in GroupSvc.getAdminList(record.peerUin.toString()) -> MemberRole.Admin
+                            else -> MemberRole.Member
+                        },
+                        title = "",
+                        level = "",
+                    )
+                )
+            )
+        }
+
+        /**
+         * 推送私聊消息
+         */
+        suspend fun transPrivateMessage(
+            record: MsgRecord,
+            elements: ArrayList<MsgElement>,
+            rawMsg: String,
+            msgHash: Int,
+            postType: PostType = PostType.Msg
+        ): Boolean {
+            val uin = app.longAccountUin
+            return transMessageEvent(record,
+                MessageEvent(
+                    time = record.msgTime,
+                    selfId = uin,
+                    postType = postType,
+                    messageType = MsgType.Private,
+                    subType = MsgSubType.Friend,
+                    messageId = msgHash,
+                    targetId = record.peerUin,
+                    peerId = uin,
+                    userId = record.senderUin,
+                    message = if(ShamrockConfig.useCQ()) rawMsg.json
+                    else elements.toSegments(record.chatType, record.peerUin.toString()).map {
+                        it.toJson()
+                    }.json,
+                    rawMessage = rawMsg,
+                    font = 0,
+                    sender = Sender(
+                        userId = record.senderUin,
+                        nickname = record.sendNickName,
+                        card = record.sendMemberName,
+                        role = when (record.senderUin) {
+                            GroupSvc.getOwner(record.peerUin.toString()) -> MemberRole.Owner
+                            in GroupSvc.getAdminList(record.peerUin.toString()) -> MemberRole.Admin
+                            else -> MemberRole.Member
+                        },
+                        title = "",
+                        level = "",
+                    )
+                )
+            )
+        }
+    }
+
+    /**
+     * 文件通知 通知器
+     */
+    object FileNoticeTransmitter {
+        /**
+         * 推送私聊文件事件
+         */
+        fun transPrivateFileEvent(
+            msgTime: Long,
+            userId: Long,
+            fileId: String,
+            fileSubId: String,
+            fileName: String,
+            fileSize: Long,
+            expireTime: Long,
+            url: String
+        ): Boolean {
+            return pushNotice(NoticeEvent(
+                time = msgTime,
+                selfId = app.longAccountUin,
+                postType = PostType.Notice,
+                type = NoticeType.PrivateUpload,
+                operatorId = userId,
+                userId = userId,
+                senderId = userId,
+                privateFile = PrivateFileMsg(
+                    id = fileId,
+                    name = fileName,
+                    size = fileSize,
+                    url = url,
+                    subId = fileSubId,
+                    expire = expireTime
+                )
+            ))
+        }
+
+        /**
+         * 推送私聊文件事件
+         */
+        fun transGroupFileEvent(
+            msgTime: Long,
+            userId: Long,
+            groupId: Long,
+            uuid: String,
+            fileName: String,
+            fileSize: Long,
+            bizId: Int,
+            url: String
+        ): Boolean {
+            return pushNotice(NoticeEvent(
+                time = msgTime,
+                selfId = app.longAccountUin,
+                postType = PostType.Notice,
+                type = NoticeType.GroupUpload,
+                operatorId = userId,
+                userId = userId,
+                groupId = groupId,
+                file = GroupFileMsg(
+                    id = uuid,
+                    name = fileName,
+                    size = fileSize,
+                    busid = bizId.toLong(),
+                    url = url
+                )
+            ))
+        }
+    }
+
+    /**
+     * 群聊通知 通知器
+     */
+    object GroupNoticeTransmitter {
+        fun transGroupPoke(time: Long, operation: Long, target: Long, groupCode: Long): Boolean {
+            return pushNotice(NoticeEvent(
+                time = time,
+                selfId = app.longAccountUin,
+                postType = PostType.Notice,
+                type = NoticeType.Notify,
+                subType = NoticeSubType.Poke,
+                operatorId = operation,
+                userId = operation,
+                groupId = groupCode,
+                target = target
+            ))
+        }
+
+        fun transGroupMemberNumChanged(
+            time: Long,
+            target: Long,
+            groupCode: Long,
+            operation: Long,
+            noticeType: NoticeType,
+            noticeSubType: NoticeSubType
+        ): Boolean {
+            return pushNotice(NoticeEvent(
+                time = time,
+                selfId = app.longAccountUin,
+                postType = PostType.Notice,
+                type = noticeType,
+                subType = noticeSubType,
+                operatorId = operation,
+                userId = operation,
+                senderId = operation,
+                target = target,
+                groupId = groupCode
+            ))
+        }
+
+        fun transGroupAdminChanged(
+            msgTime: Long,
+            target: Long,
+            groupCode: Long,
+            setAdmin: Boolean
+        ): Boolean {
+            return pushNotice(NoticeEvent(
+                time = msgTime,
+                selfId = app.longAccountUin,
+                postType = PostType.Notice,
+                type = NoticeType.GroupAdminChange,
+                subType = if (setAdmin) NoticeSubType.Set else NoticeSubType.UnSet,
+                operatorId = 0,
+                target = target,
+                groupId = groupCode
+            ))
+        }
+
+        fun transGroupBan(
+            msgTime: Long,
+            operation: Long,
+            target: Long,
+            groupCode: Long,
+            duration: Int
+        ): Boolean {
+            return pushNotice(NoticeEvent(
+                time = msgTime,
+                selfId = app.longAccountUin,
+                postType = PostType.Notice,
+                type = NoticeType.GroupBan,
+                subType = if (duration == 0) NoticeSubType.LiftBan else NoticeSubType.Ban,
+                operatorId = operation,
+                userId = operation,
+                senderId = operation,
+                target = target,
+                groupId = groupCode,
+                duration = duration
+            ))
+        }
+
+        fun transGroupMsgRecall(
+            time: Long,
+            operator: Long,
+            target: Long,
+            groupCode: Long,
+            msgHash: Int,
+            tipText: String
+        ): Boolean {
+            return pushNotice(NoticeEvent(
+                time = time,
+                selfId = app.longAccountUin,
+                postType = PostType.Notice,
+                type = NoticeType.GroupRecall,
+                operatorId = operator,
+                userId = target,
+                msgId = msgHash,
+                tip = tipText,
+                groupId = groupCode
+            ))
+        }
+    }
+
+    /**
+     * 私聊通知 通知器
+     */
+    object PrivateNoticeTransmitter {
+        fun transPrivatePoke(msgTime: Long, operation: Long, target: Long): Boolean {
+            return pushNotice(NoticeEvent(
+                time = msgTime,
+                selfId = app.longAccountUin,
+                postType = PostType.Notice,
+                type = NoticeType.Notify,
+                subType = NoticeSubType.Poke,
+                operatorId = operation,
+                userId = operation,
+                senderId = operation,
+                target = target
+            ))
+        }
+
+        fun transPrivateRecall(time: Long, operation: Long, msgHashId: Int, tipText: String): Boolean {
+            return pushNotice(NoticeEvent(
+                time = time,
+                selfId = app.longAccountUin,
+                postType = PostType.Notice,
+                type = NoticeType.FriendRecall,
+                subType = NoticeSubType.Poke,
+                operatorId = operation,
+                userId = operation,
+                msgId = msgHashId,
+                tip = tipText
+            ))
+        }
+    }
+
+    @ShamrockDsl
+    suspend fun onMessageEvent(collector: FlowCollector<Pair<MsgRecord, MessageEvent>>) {
+        messageEventFlow.collect(collector)
+    }
+
+    @ShamrockDsl
+    suspend fun onNoticeEvent(collector: FlowCollector<NoticeEvent>) {
+        noticeEventFlow.collect(collector)
+    }
+}
+
+
+

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/api/GlobalPusher.kt

@@ -1,34 +0,0 @@
-package moe.fuqiuluo.shamrock.remote.service.api
-
-import moe.fuqiuluo.shamrock.helper.Level
-import moe.fuqiuluo.shamrock.helper.LogCenter
-import moe.fuqiuluo.shamrock.remote.service.config.ShamrockConfig
-import java.util.Collections
-
-internal object GlobalPusher {
-    private val cacheConn = Collections.synchronizedMap(mutableMapOf<String, BasePushServlet>())
-
-    fun register(servlet: BasePushServlet){
-        if (ShamrockConfig.isIgnoreAllEvent()) {
-            return
-        }
-        LogCenter.log("推送器注册(id = ${servlet.id}): ${servlet.toString().split(".").last()}", Level.WARN)
-        if (!cacheConn.containsKey(servlet.id) && !cacheConn.containsValue(servlet)) {
-            cacheConn[servlet.id] = servlet
-        }
-    }
-
-    fun unregister(servlet: BasePushServlet){
-        LogCenter.log("推送器注销(id = ${servlet.id}): ${servlet.toString().split(".").last()}", Level.WARN)
-        if (cacheConn.containsKey(servlet.id) || cacheConn.containsValue(servlet)) {
-            cacheConn.remove(servlet.id)
-        }
-    }
-
-    operator fun invoke(): List<BasePushServlet> {
-        return cacheConn.map { it.value }
-    }
-}
-
-
-

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/api/HttpTransmitServlet.kt

@@ -1,6 +1,5 @@
 package moe.fuqiuluo.shamrock.remote.service.api
 
-import com.arthenica.ffmpegkit.BuildConfig
 import io.ktor.client.network.sockets.ConnectTimeoutException
 import io.ktor.client.plugins.HttpRequestTimeoutException
 import io.ktor.client.request.header
@@ -9,6 +8,7 @@ import io.ktor.client.request.setBody
 import io.ktor.client.statement.HttpResponse
 import io.ktor.http.ContentType
 import io.ktor.http.contentType
+import kotlinx.coroutines.Job
 import moe.fuqiuluo.shamrock.remote.service.config.ShamrockConfig
 import moe.fuqiuluo.shamrock.tools.GlobalClient
 import moe.fuqiuluo.shamrock.helper.Level
@@ -19,16 +19,15 @@ import moe.fuqiuluo.shamrock.xposed.helper.AppRuntimeFetcher
 import mqq.app.MobileQQ
 import java.net.SocketException
 
-internal abstract class HttpPushServlet : BasePushServlet {
+internal abstract class HttpTransmitServlet : BaseTransmitServlet {
-    override val address: String
+    override val address: String  by lazy { ShamrockConfig.getWebHookAddress() }
-        get() = ShamrockConfig.getWebHookAddress()
 
-    override fun allowPush(): Boolean {
+    override fun allowTransmit(): Boolean {
         return ShamrockConfig.allowWebHook()
     }
 
     protected suspend inline fun <reified T> pushTo(body: T): HttpResponse? {
-        if (!allowPush()) return null
+        if (!allowTransmit()) return null
         try {
             if (address.startsWith("http://") || address.startsWith("https://")) {
                 return GlobalClient.post(address) {

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/api/WebSocketClientServlet.kt

@@ -2,12 +2,13 @@
 
 package moe.fuqiuluo.shamrock.remote.service.api
 
+import io.ktor.client.statement.bodyAsText
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.launch
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
-import kotlinx.serialization.json.JsonObject
 import moe.fuqiuluo.shamrock.remote.action.ActionManager
 import moe.fuqiuluo.shamrock.remote.action.ActionSession
 import moe.fuqiuluo.shamrock.remote.entries.EmptyObject
@@ -17,6 +18,7 @@ import moe.fuqiuluo.shamrock.remote.service.config.ShamrockConfig
 import moe.fuqiuluo.shamrock.tools.*
 import moe.fuqiuluo.shamrock.helper.Level
 import moe.fuqiuluo.shamrock.helper.LogCenter
+import moe.fuqiuluo.shamrock.remote.service.HttpService
 import moe.fuqiuluo.shamrock.remote.service.data.BotStatus
 import moe.fuqiuluo.shamrock.remote.service.data.Self
 import moe.fuqiuluo.shamrock.remote.service.data.push.MetaEventType
@@ -24,7 +26,6 @@ import moe.fuqiuluo.shamrock.remote.service.data.push.MetaSubType
 import moe.fuqiuluo.shamrock.remote.service.data.push.PostType
 import moe.fuqiuluo.shamrock.remote.service.data.push.PushMetaEvent
 import moe.fuqiuluo.shamrock.xposed.helper.AppRuntimeFetcher
-import mqq.app.MobileQQ
 import org.java_websocket.client.WebSocketClient
 import org.java_websocket.handshake.ServerHandshake
 import java.lang.Exception
@@ -34,16 +35,17 @@ import kotlin.concurrent.timer
 internal abstract class WebSocketClientServlet(
     url: String,
     wsHeaders: Map<String, String>
-) : BasePushServlet, WebSocketClient(URI(url), wsHeaders) {
+) : BaseTransmitServlet, WebSocketClient(URI(url), wsHeaders) {
-    override fun allowPush(): Boolean {
+    override fun allowTransmit(): Boolean {
         return ShamrockConfig.openWebSocketClient()
     }
 
     override fun onOpen(handshakedata: ServerHandshake?) {
         LogCenter.log("WebSocketClient onOpen: ${handshakedata?.httpStatus}, ${handshakedata?.httpStatusMessage}")
+
         startHeartbeatTimer()
         pushMetaLifecycle()
-        GlobalPusher.register(this)
+        initTransmitter()
     }
 
     override fun onMessage(message: String) {
@@ -79,16 +81,16 @@ internal abstract class WebSocketClientServlet(
 
     override fun onClose(code: Int, reason: String?, remote: Boolean) {
         LogCenter.log("WebSocketClient onClose: $code, $reason, $remote")
-        GlobalPusher.unregister(this)
+        cancelFlowJobs()
     }
 
     override fun onError(ex: Exception?) {
         LogCenter.log("WebSocketClient onError: ${ex?.message}")
-        GlobalPusher.unregister(this)
+        cancelFlowJobs()
     }
 
     protected inline fun <reified T> pushTo(body: T) {
-        if (!allowPush() || isClosed || isClosing) return
+        if (!allowTransmit() || isClosed || isClosing) return
         try {
             send(GlobalJson.encodeToString(body))
         } catch (e: Throwable) {

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/api/WebSocketTransmitServlet.kt

@@ -21,7 +21,6 @@ import moe.fuqiuluo.shamrock.tools.*
 import moe.fuqiuluo.shamrock.helper.Level
 import moe.fuqiuluo.shamrock.helper.LogCenter
 import moe.fuqiuluo.shamrock.xposed.helper.AppRuntimeFetcher
-import mqq.app.MobileQQ
 import org.java_websocket.WebSocket
 import org.java_websocket.server.WebSocketServer
 import java.net.InetSocketAddress
@@ -29,15 +28,15 @@ import java.net.URI
 import java.util.Collections
 import kotlin.concurrent.timer
 
-internal abstract class WebSocketPushServlet(
+internal abstract class WebSocketTransmitServlet(
     port: Int
-) : BasePushServlet, WebSocketServer(InetSocketAddress(port)) {
+) : BaseTransmitServlet, WebSocketServer(InetSocketAddress(port)) {
     protected val eventReceivers: MutableList<WebSocket> = Collections.synchronizedList(mutableListOf<WebSocket>())
 
     override val address: String
         get() = "-"
 
-     override fun allowPush(): Boolean {
+     override fun allowTransmit(): Boolean {
          return ShamrockConfig.openWebSocket()
      }
 
@@ -102,16 +101,16 @@ internal abstract class WebSocketPushServlet(
 
     override fun onError(conn: WebSocket, ex: Exception?) {
         LogCenter.log("WSServer Error: " + ex?.stackTraceToString(), Level.ERROR)
-        GlobalPusher.unregister(this)
+        cancelFlowJobs()
     }
 
     override fun onStart() {
-        GlobalPusher.register(this)
+        initTransmitter()
         LogCenter.log("WSServer start running on ws://0.0.0.0:$port!")
     }
 
     protected inline fun <reified T> pushTo(body: T) {
-        if(!allowPush()) return
+        if(!allowTransmit()) return
         try {
             broadcastTextEvent(GlobalJson.encodeToString(body))
         } catch (e: Throwable) {

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/data/push/MessageEvent.kt

@@ -41,7 +41,7 @@ internal enum class PostType {
  * 不要使用继承的方式实现通用字段，那样会很难维护！
  */
 @Serializable
-internal data class PushMessage (
+internal data class MessageEvent (
     @SerialName("time") val time: Long,
     @SerialName("self_id") val selfId: Long,
     @SerialName("post_type") val postType: PostType,

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/data/push/NoticeEvent.kt

@@ -39,7 +39,7 @@ internal enum class NoticeSubType {
  * 不要使用继承的方式实现通用字段，那样会很难维护！
  */
 @Serializable
-internal data class PushNotice(
+internal data class NoticeEvent(
     @SerialName("time") val time: Long,
     @SerialName("self_id") val selfId: Long,
     @SerialName("post_type") val postType: PostType,
@@ -50,7 +50,7 @@ internal data class PushNotice(
     @SerialName("user_id") val userId: Long = 0,
     @SerialName("sender_id") val senderId: Long = 0,
     @SerialName("duration") val duration: Int = 0,
-    @SerialName("message_id") val msgId: Int,
+    @SerialName("message_id") val msgId: Int = 0,
     @SerialName("tip_text") val tip: String = "",
     @SerialName("target_id") val target: Long = 0,
     @SerialName("file") val file: GroupFileMsg? = null,

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/listener/AioListener.kt

@@ -1,7 +1,6 @@
 @file:OptIn(DelicateCoroutinesApi::class)
 package moe.fuqiuluo.shamrock.remote.service.listener
 
-import android.util.Log
 import moe.fuqiuluo.shamrock.helper.MessageHelper
 import com.tencent.qqnt.kernel.nativeinterface.*
 import kotlinx.coroutines.DelicateCoroutinesApi
@@ -9,10 +8,11 @@ import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.launch
 import moe.fuqiuluo.qqinterface.servlet.msg.convert.toCQCode
 import moe.fuqiuluo.qqinterface.servlet.transfile.RichProtoSvc
-import moe.fuqiuluo.shamrock.remote.service.api.GlobalPusher
 import moe.fuqiuluo.shamrock.remote.service.config.ShamrockConfig
 import moe.fuqiuluo.shamrock.helper.Level
 import moe.fuqiuluo.shamrock.helper.LogCenter
+import moe.fuqiuluo.shamrock.remote.service.api.GlobalEventTransmitter
+import moe.fuqiuluo.shamrock.remote.service.data.push.PostType
 import java.util.ArrayList
 import java.util.HashMap
 
@@ -52,8 +52,10 @@ internal object AioListener: IKernelMsgListener {
                         if (rule.white?.contains(record.peerUin) == false) return
                     }
 
-                    GlobalPusher().forEach {
+                    if(!GlobalEventTransmitter.MessageTransmitter.transGroupMessage(
-                        it.pushGroupMsg(record, record.elements, rawMsg, msgHash)
+                        record, record.elements, rawMsg, msgHash
+                    )) {
+                        LogCenter.log("群消息推送失败 -> MessageTransmitter", Level.WARN)
                     }
                 }
                 MsgConstant.KCHATTYPEC2C -> {
@@ -63,8 +65,10 @@ internal object AioListener: IKernelMsgListener {
                         if (rule.white?.contains(record.peerUin) == false) return
                     }
 
-                    GlobalPusher().forEach {
+                    if(!GlobalEventTransmitter.MessageTransmitter.transPrivateMessage(
-                        it.pushPrivateMsg(record, record.elements, rawMsg, msgHash)
+                            record, record.elements, rawMsg, msgHash
+                    )) {
+                        LogCenter.log("私聊消息推送失败 -> MessageTransmitter", Level.WARN)
                     }
                 }
                 else -> LogCenter.log("不支持PUSH事件: ${record.chatType}")
@@ -98,14 +102,12 @@ internal object AioListener: IKernelMsgListener {
 
                 when (record.chatType) {
                     MsgConstant.KCHATTYPEGROUP -> {
-                        GlobalPusher().forEach {
+                        GlobalEventTransmitter.MessageTransmitter
-                            it.pushSelfGroupSentMsg(record, record.elements, rawMsg, msgHash)
+                            .transGroupMessage(record, record.elements, rawMsg, msgHash, PostType.MsgSent)
-                        }
                     }
                     MsgConstant.KCHATTYPEC2C -> {
-                        GlobalPusher().forEach {
+                        GlobalEventTransmitter.MessageTransmitter
-                            it.pushSelfPrivateSentMsg(record, record.elements, rawMsg, msgHash)
+                            .transPrivateMessage(record, record.elements, rawMsg, msgHash, PostType.MsgSent)
-                        }
                     }
                     else -> LogCenter.log("不支持SELF PUSH事件: ${record.chatType}")
                 }
@@ -220,8 +222,9 @@ internal object AioListener: IKernelMsgListener {
         val fileSubId = fileMsg.fileSubId ?: ""
         val url = RichProtoSvc.getC2CFileDownUrl(fileId, fileSubId)
 
-        GlobalPusher().forEach {
+        if(!GlobalEventTransmitter.FileNoticeTransmitter
-            it.pushC2CFileCome(record.msgTime, userId, fileId, fileSubId, fileName, fileSize, expireTime, url)
+            .transPrivateFileEvent(record.msgTime, userId, fileId, fileSubId, fileName, fileSize, expireTime, url)) {
+            LogCenter.log("私聊文件消息推送失败 -> FileNoticeTransmitter", Level.WARN)
         }
     }
 
@@ -242,8 +245,9 @@ internal object AioListener: IKernelMsgListener {
 
         val url = RichProtoSvc.getGroupFileDownUrl(record.peerUin, uuid, bizId)
 
-        GlobalPusher().forEach {
+        if(!GlobalEventTransmitter.FileNoticeTransmitter
-            it.pushGroupFileCome(record.msgTime, userId, groupId, uuid, fileName, fileSize, bizId, url)
+            .transGroupFileEvent(record.msgTime, userId, groupId, uuid, fileName, fileSize, bizId, url)) {
+            LogCenter.log("群聊文件消息推送失败 -> FileNoticeTransmitter", Level.WARN)
         }
     }
 

xposed/src/main/java/moe/fuqiuluo/shamrock/remote/service/listener/PrimitiveListener.kt

@@ -18,15 +18,14 @@ import moe.fuqiuluo.proto.asUtf8String
 import moe.fuqiuluo.proto.ProtoUtils
 import moe.fuqiuluo.proto.asByteArray
 import moe.fuqiuluo.proto.asList
-import moe.fuqiuluo.proto.asMap
 import moe.fuqiuluo.proto.asULong
 import moe.fuqiuluo.shamrock.helper.MessageHelper
-import moe.fuqiuluo.shamrock.remote.service.api.GlobalPusher
 import moe.fuqiuluo.shamrock.remote.service.data.push.NoticeSubType
 import moe.fuqiuluo.shamrock.remote.service.data.push.NoticeType
 import moe.fuqiuluo.shamrock.tools.slice
 import moe.fuqiuluo.shamrock.helper.Level
 import moe.fuqiuluo.shamrock.helper.LogCenter
+import moe.fuqiuluo.shamrock.remote.service.api.GlobalEventTransmitter
 import moe.fuqiuluo.shamrock.tools.readBuf32Long
 import moe.fuqiuluo.shamrock.tools.toHexString
 import moe.fuqiuluo.shamrock.xposed.helper.PacketHandler
@@ -90,7 +89,10 @@ internal object PrimitiveListener {
             }
         LogCenter.log("私聊戳一戳: $operation -> $target")
 
-        GlobalPusher().forEach { it.pushC2CPoke(msgTime, operation.toLong(), target.toLong()) }
+        if(!GlobalEventTransmitter.PrivateNoticeTransmitter
+            .transPrivatePoke(msgTime, operation.toLong(), target.toLong())) {
+            LogCenter.log("私聊戳一戳推送失败！", Level.WARN)
+        }
     }
 
     private fun onGroupPoke(time: Long, pb: ProtoMap) {
@@ -127,7 +129,10 @@ internal object PrimitiveListener {
         }
         LogCenter.log("群戳一戳($groupCode): $operation -> $target")
 
-        GlobalPusher().forEach { it.pushGroupPoke(time, operation.toLong(), target.toLong(), groupCode) }
+        if(!GlobalEventTransmitter.GroupNoticeTransmitter
+            .transGroupPoke(time, operation.toLong(), target.toLong(), groupCode)) {
+            LogCenter.log("群戳一戳推送失败！", Level.WARN)
+        }
     }
 
     private suspend fun onC2CRecall(time: Long, pb: ProtoMap) {
@@ -143,8 +148,9 @@ internal object PrimitiveListener {
 
         LogCenter.log("私聊消息撤回: $operation, seq = $msgSeq, hash = ${mapping.msgHashId}, tip = $tipText")
 
-        GlobalPusher().forEach {
+        if(!GlobalEventTransmitter.PrivateNoticeTransmitter
-            it.pushPrivateMsgRecall(time, operation, mapping.msgHashId, tipText)
+            .transPrivateRecall(time, operation, mapping.msgHashId, tipText)) {
+            LogCenter.log("私聊消息撤回推送失败！", Level.WARN)
         }
     }
 
@@ -157,12 +163,13 @@ internal object PrimitiveListener {
 
         LogCenter.log("群成员增加($groupCode): $target, type = $type")
 
-        GlobalPusher().forEach {
+        if(!GlobalEventTransmitter.GroupNoticeTransmitter
-            it.pushGroupMemberDecreased(time, target, groupCode, operation, NoticeType.GroupMemIncrease, when(type) {
+            .transGroupMemberNumChanged(time, target, groupCode, operation, NoticeType.GroupMemIncrease, when(type) {
                 130 -> NoticeSubType.Approve
                 131 -> NoticeSubType.Invite
                 else -> NoticeSubType.Approve
-            })
+            })) {
+            LogCenter.log("群成员增加推送失败！", Level.WARN)
         }
     }
 
@@ -176,13 +183,14 @@ internal object PrimitiveListener {
         val target = ContactHelper.getUinByUidAsync(targetUid).toLong()
         LogCenter.log("群成员减少($groupCode): $target, type = $type")
 
-        GlobalPusher().forEach {
+        if(!GlobalEventTransmitter.GroupNoticeTransmitter
-            it.pushGroupMemberDecreased(time, target, groupCode, operation, NoticeType.GroupMemDecrease, when(type) {
+                .transGroupMemberNumChanged(time, target, groupCode, operation, NoticeType.GroupMemDecrease, when(type) {
-                130 -> NoticeSubType.Kick
+                    130 -> NoticeSubType.Kick
-                131 -> NoticeSubType.Leave
+                    131 -> NoticeSubType.Leave
-                3 -> NoticeSubType.KickMe
+                    3 -> NoticeSubType.KickMe
-                else -> NoticeSubType.Kick
+                    else -> NoticeSubType.Kick
-            })
+                })) {
+            LogCenter.log("群成员减少推送失败！", Level.WARN)
         }
     }
 
@@ -200,8 +208,9 @@ internal object PrimitiveListener {
         val target = ContactHelper.getUinByUidAsync(targetUid).toLong()
         LogCenter.log("群管理员变动($groupCode): $target, isSetAdmin = $isSetAdmin")
 
-        GlobalPusher().forEach {
+        if(!GlobalEventTransmitter.GroupNoticeTransmitter
-            it.pushGroupAdminChange(msgTime, target, groupCode, isSetAdmin)
+            .transGroupAdminChanged(msgTime, target, groupCode, isSetAdmin)) {
+            LogCenter.log("群管理员变动推送失败！", Level.WARN)
         }
     }
 
@@ -214,8 +223,9 @@ internal object PrimitiveListener {
         val target = ContactHelper.getUinByUidAsync(targetUid).toLong()
         LogCenter.log("群禁言($groupCode): $operation -> $target, 时长 = ${duration}s")
 
-        GlobalPusher().forEach {
+        if(!GlobalEventTransmitter.GroupNoticeTransmitter
-            it.pushGroupBan(msgTime, operation, target, groupCode, duration)
+            .transGroupBan(msgTime, operation, target, groupCode, duration)) {
+            LogCenter.log("群禁言推送失败！", Level.WARN)
         }
     }
 
@@ -246,8 +256,9 @@ internal object PrimitiveListener {
 
             LogCenter.log("群消息撤回($groupCode): $operator -> $target, seq = $msgSeq, hash = $msgHash, tip = $tipText")
 
-            GlobalPusher().forEach {
+            if(!GlobalEventTransmitter.GroupNoticeTransmitter
-                it.pushGroupMsgRecall(time, operator, target, groupCode, msgHash, tipText)
+                .transGroupMsgRecall(time, operator, target, groupCode, msgHash, tipText)) {
+                LogCenter.log("群消息撤回推送失败！", Level.WARN)
             }
         } finally {
             readPacket.release()

xposed/src/main/java/moe/fuqiuluo/shamrock/xposed/actions/InitRemoteService.kt

@@ -8,12 +8,13 @@ import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.launch
 import moe.fuqiuluo.shamrock.remote.service.WebSocketClientService
 import moe.fuqiuluo.shamrock.remote.service.WebSocketService
-import moe.fuqiuluo.shamrock.remote.service.api.GlobalPusher
+import moe.fuqiuluo.shamrock.remote.service.api.GlobalEventTransmitter
 import moe.fuqiuluo.shamrock.remote.service.config.ShamrockConfig
 import moe.fuqiuluo.shamrock.utils.PlatformUtils
 import moe.fuqiuluo.shamrock.helper.Level
 import moe.fuqiuluo.shamrock.helper.LogCenter
 import moe.fuqiuluo.shamrock.remote.HTTPServer
+import moe.fuqiuluo.shamrock.remote.service.HttpService
 import moe.fuqiuluo.shamrock.tools.ShamrockVersion
 import moe.fuqiuluo.shamrock.xposed.helper.AppRuntimeFetcher
 import mqq.app.MobileQQ
@@ -34,7 +35,7 @@ internal class InitRemoteService : IAction {
         if (!PlatformUtils.isMqqPackage()) return
 
         if (ShamrockConfig.allowWebHook()) {
-            GlobalPusher.register(moe.fuqiuluo.shamrock.remote.service.HttpService)
+            HttpService.initTransmitter()
         }
 
         if (ShamrockConfig.openWebSocket()) {
@@ -84,7 +85,7 @@ internal class InitRemoteService : IAction {
                     wsClient.connect()
                     timer(initialDelay = 5000L, period = 5000L) {
                         if (wsClient.isClosed || wsClient.isClosing) {
-                            GlobalPusher.unregister(wsClient)
+                            wsClient.cancelFlowJobs()
                             wsClient = WebSocketClientService(url, wsHeaders)
                             wsClient.connect()
                         }

xposed/src/main/java/moe/fuqiuluo/shamrock/xposed/helper/NTServiceFetcher.kt

@@ -73,6 +73,7 @@ internal object NTServiceFetcher {
             LogCenter.log("Register MSG listener successfully.")
             msgService.addMsgListener(AioListener)
 
+            // 接口缺失 暂不使用
             //groupService.addKernelGroupListener(GroupEventListener)
             //LogCenter.log("Register Group listener successfully.")