|
|
@@ -15,15 +15,20 @@ import net.mamoe.mirai.internal.contact.CommonGroupImpl
|
|
|
import net.mamoe.mirai.internal.message.getMessageSourceKindFromC2cCmdOrNull
|
|
|
import net.mamoe.mirai.internal.message.toMessageChainOnline
|
|
|
import net.mamoe.mirai.internal.network.protocol.data.proto.MsgComm
|
|
|
-import net.mamoe.mirai.internal.network.protocol.packet.chat.TroopManagement
|
|
|
import net.mamoe.mirai.internal.network.protocol.packet.chat.receive.MessageSvcPbGetGroupMsg
|
|
|
import net.mamoe.mirai.message.data.MessageChain
|
|
|
+import net.mamoe.mirai.message.data.MessageSource
|
|
|
+import net.mamoe.mirai.message.data.OnlineMessageSource
|
|
|
|
|
|
internal class RoamingMessagesImplGroup(
|
|
|
override val contact: CommonGroupImpl
|
|
|
-) : AbstractRoamingMessages() {
|
|
|
- private val bot get() = contact.bot
|
|
|
+) : SequenceBasedRoamingMessagesImpl() {
|
|
|
|
|
|
+ /**
|
|
|
+ * time-based roaming without extending [TimeBasedRoamingMessagesImpl]
|
|
|
+ * because protocol MessageSvc.PbGetGroupMsg doesn't support querying via time.
|
|
|
+ * so this is actually sequence-based roaming.
|
|
|
+ */
|
|
|
override suspend fun getMessagesIn(
|
|
|
timeStart: Long,
|
|
|
timeEnd: Long,
|
|
|
@@ -33,25 +38,11 @@ internal class RoamingMessagesImplGroup(
|
|
|
|
|
|
return flow {
|
|
|
while (true) {
|
|
|
- val resp = contact.bot.network.sendAndExpect(
|
|
|
- MessageSvcPbGetGroupMsg(
|
|
|
- client = contact.bot.client,
|
|
|
- groupUin = contact.uin,
|
|
|
- messageSequence = currentSeq.toLong(),
|
|
|
- count = 20 // maximum 20
|
|
|
- )
|
|
|
- )
|
|
|
-
|
|
|
- if (resp is MessageSvcPbGetGroupMsg.Failed) break
|
|
|
- resp as MessageSvcPbGetGroupMsg.Success // stupid smart cast
|
|
|
- if (resp.msgElem.isEmpty()) break
|
|
|
+ val resp = getGroupMsg(currentSeq.toLong()) ?: break
|
|
|
|
|
|
// the message may be sorted increasing by message time,
|
|
|
// if so, additional sortBy will not take cost.
|
|
|
- val messageTimeSequence = resp.msgElem.asSequence().map { it.time }
|
|
|
-
|
|
|
- val maxTime = messageTimeSequence.max()
|
|
|
-
|
|
|
+ val maxTime = resp.msgElem.maxOf { it.time }
|
|
|
if (maxTime < timeStart) break // we have fetched all messages
|
|
|
|
|
|
emitAll(
|
|
|
@@ -69,24 +60,55 @@ internal class RoamingMessagesImplGroup(
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- private val MsgComm.Msg.time get() = msgHead.msgTime
|
|
|
+ override suspend fun getMessagesBeforeFlow(
|
|
|
+ source: MessageSource?,
|
|
|
+ filter: RoamingMessageFilter?
|
|
|
+ ): Flow<MessageChain> {
|
|
|
+ var currentSeq = if (source != null) {
|
|
|
+ (source as? OnlineMessageSource) ?: error("source is not OnlineMessageSource")
|
|
|
+ source.ids.firstOrNull()?.minus(1) ?: return emptyFlow()
|
|
|
+ } else {
|
|
|
+ getLastMsgSeq() ?: return emptyFlow()
|
|
|
+ }
|
|
|
|
|
|
- private fun RoamingMessageFilter?.apply(
|
|
|
- it: MsgComm.Msg
|
|
|
- ) = this?.invoke(createRoamingMessage(it, listOf())) != false
|
|
|
+ return flow {
|
|
|
+ while (true) {
|
|
|
+ val resp = getGroupMsg(currentSeq.toLong()) ?: break
|
|
|
|
|
|
- private suspend fun getLastMsgSeq(): Int? {
|
|
|
- // Iterate from the newest message to find messages within [timeStart] and [timeEnd]
|
|
|
- val lastMsgSeqResp = bot.network.sendAndExpect(
|
|
|
- TroopManagement.GetGroupLastMsgSeq(
|
|
|
- client = bot.client,
|
|
|
- groupUin = contact.uin
|
|
|
+ emitAll(
|
|
|
+ resp.msgElem.asSequence()
|
|
|
+ .filter { getMessageSourceKindFromC2cCmdOrNull(it.msgHead.c2cCmd) != null } // ignore unsupported messages
|
|
|
+ .sortedByDescending { it.time } // Ensure caller receiver newer messages first
|
|
|
+ .filter { filter.apply(it) } // Call filter after sort
|
|
|
+ .asFlow()
|
|
|
+ .map { it.toMessageChainOnline(bot) }
|
|
|
+ )
|
|
|
+
|
|
|
+ currentSeq = resp.msgElem.minBy { it.time }.msgHead.msgSeq
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private suspend fun getGroupMsg(seq: Long): MessageSvcPbGetGroupMsg.Success? {
|
|
|
+ val resp = contact.bot.network.sendAndExpect(
|
|
|
+ MessageSvcPbGetGroupMsg(
|
|
|
+ client = contact.bot.client,
|
|
|
+ groupUin = contact.uin,
|
|
|
+ messageSequence = seq,
|
|
|
+ count = 20 // maximum 20
|
|
|
)
|
|
|
)
|
|
|
|
|
|
- return when (lastMsgSeqResp) {
|
|
|
- TroopManagement.GetGroupLastMsgSeq.Response.Failed -> null
|
|
|
- is TroopManagement.GetGroupLastMsgSeq.Response.Success -> lastMsgSeqResp.seq
|
|
|
- }
|
|
|
+ if (resp is MessageSvcPbGetGroupMsg.Failed) return null
|
|
|
+ resp as MessageSvcPbGetGroupMsg.Success
|
|
|
+ if (resp.msgElem.isEmpty()) return null
|
|
|
+
|
|
|
+ return resp
|
|
|
}
|
|
|
+
|
|
|
+ private val MsgComm.Msg.time get() = msgHead.msgTime
|
|
|
+
|
|
|
+ private fun RoamingMessageFilter?.apply(
|
|
|
+ it: MsgComm.Msg
|
|
|
+ ) = this?.invoke(createRoamingMessage(it, listOf())) != false
|
|
|
}
|
StageGuard