Implement Group OutgoingMessagePipeline

mirai-core/src/commonMain/kotlin/QQAndroidBot.kt

@@ -223,6 +223,8 @@ internal open class QQAndroidBot constructor(
             AccountSecretsManager,
             configuration.createAccountsSecretsManager(bot.logger.subLogger("AccountSecretsManager")),
         )
+
+        set(OutgoingMessagePipelineFactory, OutgoingMessagePipelineFactoryImpl())
     }
 
     /**

mirai-core/src/commonMain/kotlin/contact/GroupImpl.kt

@@ -19,12 +19,16 @@ import net.mamoe.mirai.contact.announcement.Announcements
 import net.mamoe.mirai.data.GroupInfo
 import net.mamoe.mirai.data.MemberInfo
 import net.mamoe.mirai.event.broadcast
-import net.mamoe.mirai.event.events.*
+import net.mamoe.mirai.event.events.BeforeImageUploadEvent
+import net.mamoe.mirai.event.events.BotLeaveEvent
+import net.mamoe.mirai.event.events.EventCancelledException
+import net.mamoe.mirai.event.events.ImageUploadEvent
 import net.mamoe.mirai.internal.QQAndroidBot
 import net.mamoe.mirai.internal.contact.announcement.AnnouncementsImpl
 import net.mamoe.mirai.internal.contact.info.MemberInfoImpl
 import net.mamoe.mirai.internal.message.*
 import net.mamoe.mirai.internal.network.components.BdhSession
+import net.mamoe.mirai.internal.network.components.OutgoingMessagePipelineFactory
 import net.mamoe.mirai.internal.network.handler.NetworkHandler
 import net.mamoe.mirai.internal.network.handler.logger
 import net.mamoe.mirai.internal.network.highway.ChannelKind
@@ -35,8 +39,6 @@ import net.mamoe.mirai.internal.network.highway.postPtt
 import net.mamoe.mirai.internal.network.highway.tryServersUpload
 import net.mamoe.mirai.internal.network.message.MessagePipelineConfiguration
 import net.mamoe.mirai.internal.network.message.MessagePipelineContextImpl
-import net.mamoe.mirai.internal.network.message.OutgoingMessagePhasesGroup
-import net.mamoe.mirai.internal.network.message.buildPhaseConfiguration
 import net.mamoe.mirai.internal.network.protocol.data.proto.Cmd0x388
 import net.mamoe.mirai.internal.network.protocol.packet.chat.TroopEssenceMsgManager
 import net.mamoe.mirai.internal.network.protocol.packet.chat.image.ImgStore
@@ -151,36 +153,12 @@ internal class GroupImpl constructor(
         return bot.id == id || members.firstOrNull { it.id == id } != null
     }
 
-    val sendMessagePipeline: MessagePipelineConfiguration<GroupImpl> = OutgoingMessagePhasesGroup.run {
-        buildPhaseConfiguration {
-            Begin then
-                    Preconditions then
-                    MessageToMessageChain then
-                    BroadcastPreSendEvent(::GroupMessagePreSendEvent) then
-                    CheckLength then
-                    EnsureSequenceIdAvailable then
-                    UploadForwardMessages then
-                    FixGroupImages then
-
-                    Savepoint(1) then
-
-                    ConvertToLongMessage onFailureJumpTo 1 then
-                    StartCreatePackets then
-                    CreatePacketsForMusicShare(specialMessageSourceStrategy) then
-                    CreatePacketsForFileMessage(specialMessageSourceStrategy) then
-                    CreatePacketsNormal() then
-                    LogMessageSent() then
-                    SendPacketsAndCreateReceipt() onFailureJumpTo 1 then
-
-                    Finish finally
-
-                    BroadcastPostSendEvent(::GroupMessagePostSendEvent) finally
-                    CloseContext() finally
-                    ThrowExceptions()
-        }
+    val sendMessagePipeline: MessagePipelineConfiguration<GroupImpl> by lazy {
+        bot.components[OutgoingMessagePipelineFactory].createForGroup(
+            this
+        )
     }
 
-
     override suspend fun sendMessage(message: Message): MessageReceipt<Group> {
         require(!message.isContentEmpty()) { "message is empty" }
         check(!isBotMuted) { throw BotIsBeingMutedException(this) }

mirai-core/src/commonMain/kotlin/message/ForceAsLongMessage.kt

@@ -17,13 +17,28 @@ import net.mamoe.mirai.utils.safeCast
 /**
  * 内部 flag, 放入 chain 强制作为 long 发送
  */
-internal object ForceAsLongMessage : MessageMetadata, ConstrainSingle, InternalFlagOnlyMessage,
+internal object ForceAsLongMessage : MessageMetadata, ConstrainSingle, InternalFlagOnlyMessage, ForceAsSomeMessage,
     AbstractMessageKey<ForceAsLongMessage>({ it.safeCast() }) {
     override val key: MessageKey<ForceAsLongMessage> get() = this
 
     override fun toString(): String = ""
 }
 
+internal sealed interface ForceAsSomeMessage : MessageMetadata {
+    companion object Key : AbstractMessageKey<ForceAsSomeMessage>({ it.safeCast() })
+}
+
+/**
+ * 内部 flag, 放入 chain 强制作为 long 发送
+ */
+internal object ForceAsFragmentedMessage : MessageMetadata, ConstrainSingle, InternalFlagOnlyMessage,
+    ForceAsSomeMessage,
+    AbstractMessageKey<ForceAsFragmentedMessage>({ it.safeCast() }) {
+    override val key: MessageKey<ForceAsFragmentedMessage> get() = this
+
+    override fun toString(): String = ""
+}
+
 /**
  * 强制不发 long
  */

mirai-core/src/commonMain/kotlin/network/components/OutgoingMessagePipelineFactory.kt

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2019-2021 Mamoe Technologies and contributors.
+ *
+ * 此源代码的使用受 GNU AFFERO GENERAL PUBLIC LICENSE version 3 许可证的约束, 可以在以下链接找到该许可证.
+ * Use of this source code is governed by the GNU AGPLv3 license that can be found through the following link.
+ *
+ * https://github.com/mamoe/mirai/blob/dev/LICENSE
+ */
+
+package net.mamoe.mirai.internal.network.components
+
+import net.mamoe.mirai.event.events.GroupMessagePostSendEvent
+import net.mamoe.mirai.event.events.GroupMessagePreSendEvent
+import net.mamoe.mirai.internal.contact.GroupImpl
+import net.mamoe.mirai.internal.network.component.ComponentKey
+import net.mamoe.mirai.internal.network.message.MessagePipelineConfiguration
+import net.mamoe.mirai.internal.network.message.OutgoingMessagePhasesGroup
+import net.mamoe.mirai.internal.network.message.buildPhaseConfiguration
+
+/**
+ * @since 2.8.0-M1
+ */
+internal interface OutgoingMessagePipelineFactory {
+    fun createForGroup(group: GroupImpl): MessagePipelineConfiguration<GroupImpl>
+
+    companion object : ComponentKey<OutgoingMessagePipelineFactory>
+}
+
+internal open class OutgoingMessagePipelineFactoryImpl : OutgoingMessagePipelineFactory {
+    override fun createForGroup(group: GroupImpl): MessagePipelineConfiguration<GroupImpl> {
+        return OutgoingMessagePhasesGroup.run {
+            buildPhaseConfiguration {
+                Begin then
+                        Preconditions then
+                        MessageToMessageChain then
+                        BroadcastPreSendEvent(::GroupMessagePreSendEvent) then
+                        CheckLength then
+                        EnsureSequenceIdAvailable then
+                        UploadForwardMessages then
+                        FixGroupImages then
+
+                        Savepoint(1) then
+
+                        ConvertToLongMessage onFailureJumpTo 1 then
+                        StartCreatePackets then
+                        CreatePacketsForMusicShare(specialMessageSourceStrategy) then
+                        CreatePacketsForFileMessage(specialMessageSourceStrategy) then
+                        CreatePacketsFallback() then
+                        LogMessageSent() then
+                        SendPacketsAndCreateReceipt() onFailureJumpTo 1 then
+
+                        Finish finally
+
+                        BroadcastPostSendEvent(::GroupMessagePostSendEvent) finally
+                        CloseContext() finally
+                        ThrowExceptions()
+            }
+        }
+    }
+
+}

mirai-core/src/commonMain/kotlin/network/message/OutgoingMessagePhasesCommon.kt

@@ -23,14 +23,11 @@ import net.mamoe.mirai.event.events.MessagePreSendEvent
 import net.mamoe.mirai.internal.contact.*
 import net.mamoe.mirai.internal.getMiraiImpl
 import net.mamoe.mirai.internal.message.*
-import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_CAN_SEND_AS_FRAGMENTED
-import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_CAN_SEND_AS_LONG
-import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_CAN_SEND_AS_SIMPLE
 import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_FINAL_MESSAGE_CHAIN
 import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_MESSAGE_SOURCE_RESULT
 import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_ORIGINAL_MESSAGE
 import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_PACKET_TRACE
-import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_SENDING_AS_FRAGMENTED
+import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_STATE_CONTROLLER
 import net.mamoe.mirai.internal.network.pipeline.*
 import net.mamoe.mirai.internal.network.protocol.packet.OutgoingPacket
 import net.mamoe.mirai.internal.network.protocol.packet.chat.FileManagement
@@ -43,7 +40,8 @@ import net.mamoe.mirai.message.MessageReceipt
 import net.mamoe.mirai.message.data.*
 import net.mamoe.mirai.utils.Either
 import net.mamoe.mirai.utils.Either.Companion.fold
-import net.mamoe.mirai.utils.cast
+import net.mamoe.mirai.utils.assertUnreachable
+import net.mamoe.mirai.utils.castOrNull
 
 
 internal typealias MessagePipelineConfigurationBuilder<C> = PipelineConfigurationBuilder<MessagePipelineContext<C>, Message, MessageReceipt<C>>
@@ -155,11 +153,16 @@ internal abstract class OutgoingMessagePhasesCommon {
     @PhaseMarker
     fun <C : AbstractContact> BroadcastPostSendEvent(
         constructor: (C, MessageChain, Throwable?, MessageReceipt<C>?) -> MessagePostSendEvent<in C>
-    ) = object : Node.Finally<MessagePipelineContext<C>>("BroadcastPreSendEvent") {
+    ) = object : Node.Finally<MessagePipelineContext<C>>("BroadcastPostSendEvent") {
         override suspend fun MessagePipelineContext<C>.doFinally() {
             val result = executionResult
             val chain = attributes[KEY_FINAL_MESSAGE_CHAIN]
-            constructor(contact, chain, result.exceptionOrNull(), result.getOrNull()?.cast()).broadcast()
+            constructor(
+                contact,
+                chain,
+                result.exceptionOrNull(),
+                result.getOrNull()?.castOrNull() ?: return
+            ).broadcast() // if cast failed, execution was failed.
         }
     }
 
@@ -271,30 +274,23 @@ internal abstract class OutgoingMessagePhasesCommon {
         object :
             AbstractPhase<MessagePipelineContext<AbstractContact>, MessageChain, MessageChain>("ConvertToLongMessage") {
             override suspend fun MessagePipelineContextRaw.doPhase(input: MessageChain): MessageChain {
-                if (ForceAsLongMessage in input) {
-                    return convertToLongMessageImpl(input)
-                }
+                val controller = attributes[KEY_STATE_CONTROLLER]
 
                 when {
-                    attributes[KEY_CAN_SEND_AS_SIMPLE] -> { // fastest
-                        attributes[KEY_CAN_SEND_AS_SIMPLE] = false
-                        attributes[KEY_SENDING_AS_FRAGMENTED] = false
-                        return input
-                    }
-                    attributes[KEY_CAN_SEND_AS_LONG] && DontAsLongMessage !in input -> {
-                        attributes[KEY_CAN_SEND_AS_LONG] = false
-                        attributes[KEY_SENDING_AS_FRAGMENTED] = false
-                        return convertToLongMessageImpl(input)
-                    }
-                    attributes[KEY_CAN_SEND_AS_FRAGMENTED] -> { // slowest
-                        attributes[KEY_CAN_SEND_AS_FRAGMENTED] = false
-                        attributes[KEY_SENDING_AS_FRAGMENTED] = true
-                        return input
-                    }
-                    else -> {
-                        error("Failed to send message: all strategies tried out.")
+                    ForceAsLongMessage in input -> return convertToLongMessageImpl(input)
+                    ForceAsFragmentedMessage in input -> return input
+                    DontAsLongMessage in input -> {
+                        controller.stateAvailability[SendMessageState.LONG] = false
                     }
                 }
+
+                controller.nextState()
+                return when (controller.state) {
+                    SendMessageState.UNINITIALIZED -> assertUnreachable()
+                    SendMessageState.ORIGIN -> input
+                    SendMessageState.LONG -> convertToLongMessageImpl(input)
+                    SendMessageState.FRAGMENTED -> input
+                }
             }
 
             suspend fun MessagePipelineContextRaw.convertToLongMessageImpl(chain: MessageChain): MessageChain {

mirai-core/src/commonMain/kotlin/network/message/OutgoingMessagePhasesGroup.kt

@@ -13,7 +13,7 @@ import kotlinx.coroutines.CompletableDeferred
 import net.mamoe.mirai.event.nextEventOrNull
 import net.mamoe.mirai.internal.contact.GroupImpl
 import net.mamoe.mirai.internal.message.OnlineMessageSourceToGroupImpl
-import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_SENDING_AS_FRAGMENTED
+import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_STATE_CONTROLLER
 import net.mamoe.mirai.internal.network.notice.group.GroupMessageProcessor
 import net.mamoe.mirai.internal.network.protocol.packet.OutgoingPacket
 import net.mamoe.mirai.internal.network.protocol.packet.chat.receive.MessageSvcPbSendMsg
@@ -47,13 +47,13 @@ internal object OutgoingMessagePhasesGroup : OutgoingMessagePhasesCommon(), Outg
 
     @Suppress("FunctionName")
     @PhaseMarker
-    fun CreatePacketsNormal() = object : CreatePacketsFallback<GroupImpl>() {
+    fun CreatePacketsFallback() = object : CreatePacketsFallback<GroupImpl>() {
         override suspend fun MessagePipelineContext<GroupImpl>.createPacketsImpl(chain: MessageChain): List<OutgoingPacket> {
             return MessageSvcPbSendMsg.createToGroupImpl(
                 bot.client,
                 contact,
                 chain,
-                fragmented = attributes[KEY_SENDING_AS_FRAGMENTED]
+                fragmented = attributes[KEY_STATE_CONTROLLER].state.isFragmented
             ) {
                 attributes[MessagePipelineContext.KEY_MESSAGE_SOURCE_RESULT] = CompletableDeferred(it)
             }

mirai-core/src/commonMain/kotlin/network/message/OutgoingMessagePipeline.kt

@@ -14,10 +14,7 @@ import kotlinx.coroutines.Deferred
 import net.mamoe.mirai.contact.Contact
 import net.mamoe.mirai.internal.contact.AbstractContact
 import net.mamoe.mirai.internal.contact.broadcastMessagePreSendEvent
-import net.mamoe.mirai.internal.network.component.ComponentKey
-import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_CAN_SEND_AS_FRAGMENTED
-import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_CAN_SEND_AS_LONG
-import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_CAN_SEND_AS_SIMPLE
+import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_STATE_CONTROLLER
 import net.mamoe.mirai.internal.network.notice.BotAware
 import net.mamoe.mirai.internal.network.pipeline.AbstractPipelineContext
 import net.mamoe.mirai.internal.network.pipeline.PipelineConfiguration
@@ -29,7 +26,6 @@ import net.mamoe.mirai.message.data.MessageChain
 import net.mamoe.mirai.message.data.OnlineMessageSource
 import net.mamoe.mirai.utils.*
 import kotlin.coroutines.CoroutineContext
-import kotlin.reflect.KClass
 
 /**
  * Steps:
@@ -45,24 +41,8 @@ import kotlin.reflect.KClass
  * 9. Post transformation
  * 10. Send packet
  *
- * @since 2.8-M1
+ * @since 2.8.0-M1
  */
-internal interface OutgoingMessagePipeline {
-    suspend fun <C : AbstractContact> sendMessage(contact: C, message: MessageChain): MessageReceipt<C>
-
-    companion object : ComponentKey<OutgoingMessagePipeline>
-}
-
-internal class OutgoingMessagePipelineImpl(
-    private val pipelineConfigurations: Map<KClass<out AbstractContact>, MessagePipelineConfiguration<out AbstractContact>>, // must be exhaustive ---- covering all AbstractContact
-) : OutgoingMessagePipeline {
-    override suspend fun <C : AbstractContact> sendMessage(contact: C, message: MessageChain): MessageReceipt<C> {
-        val context = MessagePipelineContextImpl<AbstractContact>(contact)
-        return pipelineConfigurations[contact::class]?.execute(context, message)?.cast()
-            ?: error("Internal error: Could")
-    }
-}
-
 internal typealias MessagePipelineConfiguration<T> = PipelineConfiguration<MessagePipelineContext<T>, Message, MessageReceipt<T>>
 
 internal interface MessagePipelineContext<out C : AbstractContact> : PipelineContext, BotAware, CoroutineScope {
@@ -74,16 +54,7 @@ internal interface MessagePipelineContext<out C : AbstractContact> : PipelineCon
 
     companion object {
         @JvmField
-        val KEY_CAN_SEND_AS_SIMPLE = TypeKey<Boolean>("canSendAsSimple")
-
-        @JvmField
-        val KEY_CAN_SEND_AS_LONG = TypeKey<Boolean>("canSendAsLong")
-
-        @JvmField
-        val KEY_CAN_SEND_AS_FRAGMENTED = TypeKey<Boolean>("canSendAsFragmented")
-
-        @JvmField
-        val KEY_SENDING_AS_FRAGMENTED = TypeKey<Boolean>("sendingAsFragmented")
+        val KEY_STATE_CONTROLLER = TypeKey<SendMessageStateController>("stateController")
 
         @JvmField
         val KEY_ORIGINAL_MESSAGE = TypeKey<Message>("originalMessage")
@@ -110,9 +81,7 @@ internal class MessagePipelineContextImpl<out C : AbstractContact>(
         .addNameHierarchically("MessagePipelineContext"),
     override val logger: MiraiLogger = contact.bot.logger.subLogger("MessagePipelineContext"), // TODO: 2021/8/15 use contact's logger
     override val attributes: MutableTypeSafeMap = buildTypeSafeMap {
-        set(KEY_CAN_SEND_AS_FRAGMENTED, true)
-        set(KEY_CAN_SEND_AS_LONG, true)
-        set(KEY_CAN_SEND_AS_SIMPLE, true)
+        set(KEY_STATE_CONTROLLER, SendMessageStateController())
     },
     override val time: TimeSource = TimeSource.System
 ) : MessagePipelineContext<C>, AbstractPipelineContext(attributes)

mirai-core/src/commonMain/kotlin/network/message/SendMessageStrategyController.kt

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2019-2021 Mamoe Technologies and contributors.
+ *
+ * 此源代码的使用受 GNU AFFERO GENERAL PUBLIC LICENSE version 3 许可证的约束, 可以在以下链接找到该许可证.
+ * Use of this source code is governed by the GNU AGPLv3 license that can be found through the following link.
+ *
+ * https://github.com/mamoe/mirai/blob/dev/LICENSE
+ */
+
+package net.mamoe.mirai.internal.network.message
+
+internal class SendMessageStateController(
+    initialState: SendMessageState = SendMessageState.UNINITIALIZED,
+) {
+    var state: SendMessageState = initialState
+        private set
+
+    val stateAvailability: MutableMap<SendMessageState, Boolean> = mutableMapOf()
+
+    fun nextState() {
+        state = when (state) {
+            SendMessageState.UNINITIALIZED -> SendMessageState.ORIGIN
+            SendMessageState.ORIGIN -> SendMessageState.LONG
+            SendMessageState.LONG -> SendMessageState.FRAGMENTED
+            SendMessageState.FRAGMENTED -> throw IllegalStateException("Failed to send message: all strategies tried out.")
+        }
+        if (stateAvailability[state] == false) {
+            nextState()
+        }
+    }
+}
+
+internal enum class SendMessageState {
+    UNINITIALIZED,
+    ORIGIN,
+    LONG,
+    FRAGMENTED
+}
+
+internal val SendMessageState.isFragmented: Boolean get() = this == SendMessageState.FRAGMENTED
+internal val SendMessageState.isLong: Boolean get() = this == SendMessageState.LONG

mirai-core/src/commonMain/kotlin/network/pipeline/PipelineConfiguration.kt

@@ -54,8 +54,9 @@ internal class PipelineConfiguration<C : PipelineContext, InitialIn, FinalOut> {
          * Run [Node.Finally]s and throw [e] with [PipelineContext.exceptionCollector].
          */
         suspend fun failAndExit(e: Throwable): Nothing {
+            context.exceptionCollector.collect(e)
             doAllFinally()
-            context.exceptionCollector.collectThrow(e)
+            context.exceptionCollector.throwLast()
         }
 
         _nodes.forEachWithIndexer { node ->
@@ -128,6 +129,12 @@ internal fun <C : PipelineContext, InitialIn, FinalOut> PipelineConfiguration<C,
     }
 })
 
+@TestOnly
+internal fun <C : PipelineContext, InitialIn, FinalOut> PipelineConfiguration<C, InitialIn, FinalOut>.replacePhase(
+    name: String,
+    doPhase: suspend C.(input: Any?) -> Any?
+): Boolean = replacePhase({ it.name == name }, name, doPhase)
+
 @TestOnly
 internal fun <C : PipelineContext, InitialIn, FinalOut> PipelineConfiguration<C, InitialIn, FinalOut>.insertPhase(
     before: (node: Node<C, *, *>) -> Boolean,

mirai-core/src/commonTest/kotlin/message/outgoing/AbstractOutgoingPipelineTest.kt

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2019-2021 Mamoe Technologies and contributors.
+ *
+ * 此源代码的使用受 GNU AFFERO GENERAL PUBLIC LICENSE version 3 许可证的约束, 可以在以下链接找到该许可证.
+ * Use of this source code is governed by the GNU AGPLv3 license that can be found through the following link.
+ *
+ * https://github.com/mamoe/mirai/blob/dev/LICENSE
+ */
+
+@file:JvmBlockingBridge
+
+package net.mamoe.mirai.internal.message.outgoing
+
+import net.mamoe.kjbb.JvmBlockingBridge
+import net.mamoe.mirai.internal.contact.AbstractContact
+import net.mamoe.mirai.internal.contact.GroupImpl
+import net.mamoe.mirai.internal.network.components.OutgoingMessagePipelineFactory
+import net.mamoe.mirai.internal.network.components.OutgoingMessagePipelineFactoryImpl
+import net.mamoe.mirai.internal.network.framework.AbstractMockNetworkHandlerTest
+import net.mamoe.mirai.internal.network.framework.replace
+import net.mamoe.mirai.internal.network.message.MessagePipelineConfiguration
+import net.mamoe.mirai.internal.network.message.MessagePipelineContext
+import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_FINAL_MESSAGE_CHAIN
+import net.mamoe.mirai.internal.network.pipeline.replacePhase
+import net.mamoe.mirai.internal.network.protocol.packet.OutgoingPacket
+import net.mamoe.mirai.internal.notice.processors.GroupExtensions
+import net.mamoe.mirai.message.MessageReceipt
+import net.mamoe.mirai.message.data.FileMessage
+import net.mamoe.mirai.message.data.MusicShare
+import net.mamoe.mirai.utils.cast
+
+internal abstract class AbstractOutgoingPipelineTest : AbstractMockNetworkHandlerTest(), GroupExtensions {
+    init {
+        components.replace(OutgoingMessagePipelineFactory) { origin ->
+            object : OutgoingMessagePipelineFactoryImpl() {
+                override fun createForGroup(group: GroupImpl): MessagePipelineConfiguration<GroupImpl> {
+                    return (origin?.createForGroup(group) ?: super.createForGroup(group)).apply {
+                        replacePhase("CreatePacketsFallback") {
+                            it ?: listOf<OutgoingPacket>().also {
+                                attributes[MessagePipelineContext.KEY_PACKET_TRACE] = "CreatePacketsFallback"
+                            }
+                        }
+                        replacePhase("CreatePacketsForMusicShare") {
+                            if (MusicShare !in attributes[KEY_FINAL_MESSAGE_CHAIN]) return@replacePhase it
+                            it ?: listOf<OutgoingPacket>().also {
+                                attributes[MessagePipelineContext.KEY_PACKET_TRACE] = "CreatePacketsForMusicShare"
+                            }
+                        }
+                        replacePhase("CreatePacketsForFileMessage") {
+                            if (FileMessage !in attributes[KEY_FINAL_MESSAGE_CHAIN]) return@replacePhase it
+                            it ?: listOf<OutgoingPacket>().also {
+                                attributes[MessagePipelineContext.KEY_PACKET_TRACE] = "CreatePacketsForFileMessage"
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    fun <T : AbstractContact> MessagePipelineConfiguration<T>.replaceSendPacketPhase(block: suspend MessagePipelineContext<T>.(packets: List<OutgoingPacket>) -> MessageReceipt<T>) {
+        replacePhase({ it.name == "SendPacketsAndCreateReceipt" }, "test monitor") { input ->
+            input.cast<List<OutgoingPacket>>()
+            block(input)
+        }
+    }
+}

mirai-core/src/commonTest/kotlin/message/outgoing/GroupOutgoingPipelineTest.kt

@@ -0,0 +1,119 @@
+/*
+ * Copyright 2019-2021 Mamoe Technologies and contributors.
+ *
+ * 此源代码的使用受 GNU AFFERO GENERAL PUBLIC LICENSE version 3 许可证的约束, 可以在以下链接找到该许可证.
+ * Use of this source code is governed by the GNU AGPLv3 license that can be found through the following link.
+ *
+ * https://github.com/mamoe/mirai/blob/dev/LICENSE
+ */
+
+package net.mamoe.mirai.internal.message.outgoing
+
+import net.mamoe.kjbb.JvmBlockingBridge
+import net.mamoe.mirai.contact.MemberPermission
+import net.mamoe.mirai.internal.contact.GroupImpl
+import net.mamoe.mirai.internal.message.OnlineMessageSourceToGroupImpl
+import net.mamoe.mirai.internal.message.createMessageReceipt
+import net.mamoe.mirai.internal.network.message.MessagePipelineContext
+import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_PACKET_TRACE
+import net.mamoe.mirai.internal.network.message.MessagePipelineContext.Companion.KEY_STATE_CONTROLLER
+import net.mamoe.mirai.internal.network.message.SendMessageState
+import net.mamoe.mirai.message.data.EmptyMessageChain
+import net.mamoe.mirai.message.data.FileMessage
+import net.mamoe.mirai.message.data.MusicKind
+import net.mamoe.mirai.message.data.MusicShare
+import org.junit.jupiter.api.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+
+@JvmBlockingBridge
+internal class GroupOutgoingPipelineTest : AbstractOutgoingPipelineTest() {
+
+    @Test
+    suspend fun `group send normal`() {
+        val group = bot.addGroup(1, 1).apply {
+            addMember(1, permission = MemberPermission.OWNER)
+        }
+        group.sendMessagePipeline.replaceSendPacketPhase {
+            assertEquals("CreatePacketsFallback", attributes[KEY_PACKET_TRACE])
+            createReceipt(group)
+        }
+        group.sendMessage("Test message")
+    }
+
+    @Test
+    suspend fun `group send states`() {
+        val group = bot.addGroup(1, 1).apply {
+            addMember(1, permission = MemberPermission.OWNER)
+        }
+        var called = 0
+        group.sendMessagePipeline.replaceSendPacketPhase {
+            assertEquals("CreatePacketsFallback", attributes[KEY_PACKET_TRACE])
+            when (called) {
+                0 -> {
+                    called++
+                    assertEquals(SendMessageState.ORIGIN, attributes[KEY_STATE_CONTROLLER].state)
+                }
+                1 -> {
+                    called++
+                    assertEquals(SendMessageState.LONG, attributes[KEY_STATE_CONTROLLER].state)
+                }
+                2 -> {
+                    called++
+                    assertEquals(SendMessageState.FRAGMENTED, attributes[KEY_STATE_CONTROLLER].state)
+                }
+            }
+            error("fake failure")
+        }
+        assertFailsWith<IllegalStateException> { group.sendMessage("a".repeat(10_000)) }.run {
+            assertEquals("fake failure", message)
+        }
+    }
+
+    @Test
+    suspend fun `group send file`() {
+        val group = bot.addGroup(1, 1).apply {
+            addMember(1, permission = MemberPermission.OWNER)
+        }
+        group.sendMessagePipeline.replaceSendPacketPhase {
+            assertEquals("CreatePacketsForFileMessage", attributes[KEY_PACKET_TRACE])
+            createReceipt(group)
+        }
+        group.sendMessage(FileMessage("id", 1, "name", 2))
+    }
+
+    @Test
+    suspend fun `group send music`() {
+        val group = bot.addGroup(1, 1).apply {
+            addMember(1, permission = MemberPermission.OWNER)
+        }
+        group.sendMessagePipeline.replaceSendPacketPhase {
+            assertEquals("CreatePacketsForMusicShare", attributes[KEY_PACKET_TRACE])
+            createReceipt(group)
+        }
+        group.sendMessage(
+            MusicShare(
+                kind = MusicKind.NeteaseCloudMusic,
+                title = "ファッション",
+                summary = "rinahamu/Yunomi",
+                brief = "",
+                jumpUrl = "http://music.163.com/song/1338728297/?userid=324076307",
+                pictureUrl = "http://p2.music.126.net/y19E5SadGUmSR8SZxkrNtw==/109951163785855539.jpg",
+                musicUrl = "http://music.163.com/song/media/outer/url?id=1338728297&userid=324076307"
+            )
+        )
+    }
+
+    private fun MessagePipelineContext<GroupImpl>.createReceipt(
+        group: GroupImpl
+    ) = OnlineMessageSourceToGroupImpl(
+        bot,
+        intArrayOf(1),
+        1,
+        EmptyMessageChain,
+        bot,
+        group,
+        intArrayOf(1)
+    ).createMessageReceipt(group, true)
+
+}

mirai-core/src/commonTest/kotlin/network/framework/AbstractMockNetworkHandlerTest.kt

@@ -14,6 +14,7 @@ package net.mamoe.mirai.internal.network.framework
 import net.mamoe.mirai.Bot
 import net.mamoe.mirai.internal.MockBot
 import net.mamoe.mirai.internal.QQAndroidBot
+import net.mamoe.mirai.internal.network.component.ComponentKey
 import net.mamoe.mirai.internal.network.component.ConcurrentComponentStorage
 import net.mamoe.mirai.internal.network.components.EventDispatcher
 import net.mamoe.mirai.internal.network.components.SsoProcessor
@@ -57,6 +58,9 @@ internal abstract class AbstractMockNetworkHandlerTest : AbstractNetworkHandlerT
     }
 
     fun NetworkHandler.assertState(state: NetworkHandler.State) {
-        assertEquals(state, state)
+        assertEquals(state, this.state)
     }
-}
+}
+
+internal fun <T : Any> ConcurrentComponentStorage.replace(key: ComponentKey<T>, block: (origin: T?) -> T) =
+    set(key, block(getOrNull(key)))