Merge remote-tracking branch 'origin/master'

CHANGELOG.md

@@ -2,6 +2,22 @@
 
 开发版本. 频繁更新, 不保证高稳定性
 
+## `0.20.0` 2020/2/23
+
+### mirai-core
+- 支持图片下载: `image.channel(): ByteReadChannel`, `image.url()`
+
+- 添加 `LockFreeLinkedList<E>.iterator`
+- 添加 `LockFreeLinkedList<E>.forEachNode`
+
+- 并行处理事件监听
+- 添加 `nextMessageContaining` 和相关可空版本
+
+- '撤回' 从 `Contact` 移动到 `Bot`
+- 删除 `MessageSource.sourceMessage`
+- 让 MessageSource 拥有唯一的 long 类型 id, 删除原 `uid` 和 `sequence` 结构.
+- 修复 `Message.eq` 歧义
+
 ## `0.19.1` 2020/2/21
 
 ### mirai-core

gradle.properties

@@ -1,8 +1,8 @@
 # style guide
 kotlin.code.style=official
 # config
-mirai_version=0.19.1
+mirai_version=0.20.0
-mirai_japt_version=1.0.1
+mirai_japt_version=1.1.0
 kotlin.incremental.multiplatform=true
 kotlin.parallel.tasks.in.project=true
 # kotlin

mirai-api-http/src/main/kotlin/net/mamoe/mirai/api/http/data/common/MessageDTO.kt

@@ -99,11 +99,9 @@ fun MessageChain.toDTOChain() = mutableListOf(this[MessageSource].toDTO()).apply
 fun MessageChainDTO.toMessageChain(contact: Contact) =
     MessageChain().apply { this@toMessageChain.forEach { add(it.toMessage(contact)) } }
 
-internal fun MessageSource.calMessageId() = (messageUid.toLong() shl 32) or (sequenceId.toLong() and 0xFFFFFFFF)
-
 @UseExperimental(ExperimentalUnsignedTypes::class)
 fun Message.toDTO() = when (this) {
-    is MessageSource -> MessageSourceDTO(calMessageId())
+    is MessageSource -> MessageSourceDTO(id)
     is At -> AtDTO(target, display)
     is AtAll -> AtAllDTO(0L)
     is Face -> FaceDTO(id)

mirai-api-http/src/main/kotlin/net/mamoe/mirai/api/http/queue/MessageQueue.kt

@@ -11,7 +11,6 @@ package net.mamoe.mirai.api.http.queue
 
 import net.mamoe.mirai.api.http.data.common.EventDTO
 import net.mamoe.mirai.api.http.data.common.IgnoreEventDTO
-import net.mamoe.mirai.api.http.data.common.calMessageId
 import net.mamoe.mirai.api.http.data.common.toDTO
 import net.mamoe.mirai.event.events.BotEvent
 import net.mamoe.mirai.message.GroupMessage
@@ -47,7 +46,7 @@ class MessageQueue : ConcurrentLinkedDeque<BotEvent>() {
     }
 
     private fun addQuoteCache(msg: GroupMessage) {
-        quoteCache[msg.message[MessageSource].calMessageId()] = msg
+        quoteCache[msg.message[MessageSource].id] = msg
         if (quoteCache.size > quoteCacheSize) {
             quoteCache.remove(quoteCache.firstKey())
         }

mirai-core-qqandroid/src/commonMain/kotlin/net/mamoe/mirai/qqandroid/ContactImpl.kt

@@ -22,7 +22,6 @@ import net.mamoe.mirai.qqandroid.network.highway.HighwayHelper
 import net.mamoe.mirai.qqandroid.network.highway.postImage
 import net.mamoe.mirai.qqandroid.network.protocol.data.jce.StTroopMemberInfo
 import net.mamoe.mirai.qqandroid.network.protocol.data.proto.Cmd0x352
-import net.mamoe.mirai.qqandroid.network.protocol.packet.chat.PbMessageSvc
 import net.mamoe.mirai.qqandroid.network.protocol.packet.chat.TroopManagement
 import net.mamoe.mirai.qqandroid.network.protocol.packet.chat.image.ImgStore
 import net.mamoe.mirai.qqandroid.network.protocol.packet.chat.image.LongConn
@@ -80,7 +79,7 @@ internal class QQImpl(
                 ) { source = it }.sendAndExpect<MessageSvc.PbSendMsg.Response>() is MessageSvc.PbSendMsg.Response.SUCCESS
             ) { "send message failed" }
         }
-        return MessageReceipt(message, source, this)
+        return MessageReceipt(source, this)
     }
 
     override suspend fun uploadImage(image: ExternalImage): Image = try {
@@ -505,25 +504,12 @@ internal class GroupImpl(
             }
         }
 
+    @MiraiExperimentalAPI
     override suspend fun quit(): Boolean {
         check(botPermission != MemberPermission.OWNER) { "An owner cannot quit from a owning group" }
         TODO("not implemented")
     }
 
-    override suspend fun recall(source: MessageSource) {
-        if (source.senderId != bot.uin) {
-            checkBotPermissionOperator()
-        }
-
-        source.ensureSequenceIdAvailable()
-
-        bot.network.run {
-            val response = PbMessageSvc.PbMsgWithDraw.Group(bot.client, this@GroupImpl.id, source.sequenceId, source.messageUid.toInt())
-                .sendAndExpect<PbMessageSvc.PbMsgWithDraw.Response>()
-            check(response is PbMessageSvc.PbMsgWithDraw.Response.Success) { "Failed to recall message #${source.sequenceId}: $response" }
-        }
-    }
-
     @UseExperimental(MiraiExperimentalAPI::class)
     override fun Member(memberInfo: MemberInfo): Member {
         return MemberImpl(
@@ -567,7 +553,7 @@ internal class GroupImpl(
 
         source.startWaitingSequenceId(this)
 
-        return MessageReceipt(message, source, this)
+        return MessageReceipt(source, this)
     }
 
     override suspend fun uploadImage(image: ExternalImage): Image = try {

mirai-core-qqandroid/src/commonMain/kotlin/net/mamoe/mirai/qqandroid/QQAndroidBot.kt

@@ -9,21 +9,23 @@
 
 package net.mamoe.mirai.qqandroid
 
-import kotlinx.io.core.ByteReadPacket
+import io.ktor.client.request.get
+import io.ktor.client.statement.HttpResponse
+import io.ktor.utils.io.ByteReadChannel
 import net.mamoe.mirai.BotAccount
 import net.mamoe.mirai.BotImpl
-import net.mamoe.mirai.contact.ContactList
+import net.mamoe.mirai.contact.*
-import net.mamoe.mirai.contact.Group
-import net.mamoe.mirai.contact.QQ
-import net.mamoe.mirai.contact.filteringGetOrNull
 import net.mamoe.mirai.data.AddFriendResult
 import net.mamoe.mirai.data.FriendInfo
 import net.mamoe.mirai.data.GroupInfo
 import net.mamoe.mirai.data.MemberInfo
-import net.mamoe.mirai.message.data.Image
+import net.mamoe.mirai.message.data.*
+import net.mamoe.mirai.qqandroid.message.CustomFaceFromServer
+import net.mamoe.mirai.qqandroid.message.NotOnlineImageFromServer
 import net.mamoe.mirai.qqandroid.network.QQAndroidBotNetworkHandler
 import net.mamoe.mirai.qqandroid.network.QQAndroidClient
 import net.mamoe.mirai.qqandroid.network.protocol.packet.chat.GroupInfoImpl
+import net.mamoe.mirai.qqandroid.network.protocol.packet.chat.PbMessageSvc
 import net.mamoe.mirai.qqandroid.network.protocol.packet.chat.TroopManagement
 import net.mamoe.mirai.qqandroid.network.protocol.packet.list.FriendList
 import net.mamoe.mirai.utils.*
@@ -119,13 +121,50 @@ internal abstract class QQAndroidBotBase constructor(
         TODO("not implemented")
     }
 
-    override suspend fun Image.download(): ByteReadPacket {
+    override suspend fun recall(source: MessageSource) {
-        TODO("not implemented")
+        if (source.senderId != uin) {
+            getGroup(source.groupId).checkBotPermissionOperator()
+        }
+
+        source.ensureSequenceIdAvailable()
+
+        network.run {
+            val response: PbMessageSvc.PbMsgWithDraw.Response =
+                PbMessageSvc.PbMsgWithDraw.Group(bot.client, source.groupId, source.sequenceId, source.messageRandom)
+                    .sendAndExpect()
+            check(response is PbMessageSvc.PbMsgWithDraw.Response.Success) { "Failed to recall message #${source.sequenceId}: $response" }
+        }
     }
 
-    @Suppress("OverridingDeprecatedMember")
+    override suspend fun recall(groupId: Long, senderId: Long, messageId: Long) {
-    override suspend fun Image.downloadAsByteArray(): ByteArray {
+        if (senderId != uin) {
-        TODO("not implemented")
+            getGroup(groupId).checkBotPermissionOperator()
+        }
+
+        val sequenceId = (messageId shr 32).toInt()
+
+        network.run {
+            val response: PbMessageSvc.PbMsgWithDraw.Response =
+                PbMessageSvc.PbMsgWithDraw.Group(bot.client, groupId, sequenceId, messageId.toInt())
+                    .sendAndExpect()
+            check(response is PbMessageSvc.PbMsgWithDraw.Response.Success) { "Failed to recall message #$sequenceId: $response" }
+        }
+    }
+
+    override suspend fun queryImageUrl(image: Image): String = "http://gchat.qpic.cn" + when (image) {
+        is NotOnlineImageFromServer -> image.delegate.origUrl
+        is CustomFaceFromServer -> image.delegate.origUrl
+        is CustomFaceFromFile -> {
+            TODO()
+        }
+        is NotOnlineImageFromFile -> {
+            TODO()
+        }
+        else -> error("unsupported image class: ${image::class.simpleName}")
+    }
+
+    override suspend fun openChannel(image: Image): ByteReadChannel {
+        return Http.get<HttpResponse>(queryImageUrl(image)).content
     }
 
     override suspend fun approveFriendAddRequest(id: Long, remark: String?) {

mirai-core-qqandroid/src/commonMain/kotlin/net/mamoe/mirai/qqandroid/message/MessageSourceFromMsg.kt

@@ -11,6 +11,7 @@ package net.mamoe.mirai.qqandroid.message
 
 import net.mamoe.mirai.contact.Group
 import net.mamoe.mirai.message.data.MessageSource
+import net.mamoe.mirai.message.data.messageRandom
 import net.mamoe.mirai.qqandroid.io.serialization.loadAs
 import net.mamoe.mirai.qqandroid.io.serialization.toByteArray
 import net.mamoe.mirai.qqandroid.network.protocol.data.proto.ImMsgBody
@@ -21,13 +22,16 @@ internal inline class MessageSourceFromServer(
     val delegate: ImMsgBody.SourceMsg
 ) : MessageSource {
     override val time: Long get() = delegate.time.toLong() and 0xFFFFFFFF
-    override val sequenceId: Int get() = delegate.origSeqs?.firstOrNull() ?: error("cannot find sequenceId from ImMsgBody.SourceMsg")
+
+    override val id: Long
+        get() = (delegate.origSeqs?.firstOrNull() ?: error("cannot find sequenceId from ImMsgBody.SourceMsg")).toLong().shl(32) or
+                (delegate.pbReserve.loadAs(SourceMsg.ResvAttr.serializer()).origUids!!.toInt()).toLong().and(0xFFFFFFFF)
+
 
     override suspend fun ensureSequenceIdAvailable() {
         // nothing to do
     }
 
-    override val messageUid: Int get() = delegate.pbReserve.loadAs(SourceMsg.ResvAttr.serializer()).origUids!!.toInt()
     // override val sourceMessage: MessageChain get() = delegate.toMessageChain()
     override val senderId: Long get() = delegate.senderUin
     override val groupId: Long get() = Group.calculateGroupCodeByGroupUin(delegate.toUin)
@@ -39,12 +43,14 @@ internal inline class MessageSourceFromMsg(
     val delegate: MsgComm.Msg
 ) : MessageSource {
     override val time: Long get() = delegate.msgHead.msgTime.toLong() and 0xFFFFFFFF
-    override val sequenceId: Int get() = delegate.msgHead.msgSeq
+    override val id: Long
+        get() = delegate.msgHead.msgSeq.toLong().shl(32) or
+                delegate.msgBody.richText.attr!!.random.toLong().and(0xFFFFFFFF)
+
     override suspend fun ensureSequenceIdAvailable() {
         // nothing to do
     }
 
-    override val messageUid: Int get() = delegate.msgBody.richText.attr!!.random
     // override val sourceMessage: MessageChain get() = delegate.toMessageChain()
     override val senderId: Long get() = delegate.msgHead.fromUin
     override val groupId: Long get() = delegate.msgHead.groupInfo!!.groupCode
@@ -62,7 +68,7 @@ internal inline class MessageSourceFromMsg(
             type = 0,
             time = delegate.msgHead.msgTime,
             pbReserve = SourceMsg.ResvAttr(
-                origUids = messageUid.toLong() and 0xffFFffFF
+                origUids = messageRandom.toLong() and 0xffFFffFF
             ).toByteArray(SourceMsg.ResvAttr.serializer()),
             srcMsg = MsgComm.Msg(
                 msgHead = MsgComm.MsgHead(
@@ -72,7 +78,7 @@ internal inline class MessageSourceFromMsg(
                     c2cCmd = delegate.msgHead.c2cCmd,
                     msgSeq = delegate.msgHead.msgSeq,
                     msgTime = delegate.msgHead.msgTime,
-                    msgUid = messageUid.toLong() and 0xffFFffFF
+                    msgUid = messageRandom.toLong() and 0xffFFffFF
                     , // ok
                     groupInfo = MsgComm.GroupInfo(groupCode = delegate.msgHead.groupInfo.groupCode),
                     isSrcMsg = true

mirai-core-qqandroid/src/commonMain/kotlin/net/mamoe/mirai/qqandroid/network/protocol/packet/chat/receive/MessageSvc.kt

@@ -272,7 +272,7 @@ internal class MessageSvc {
         }
 
         internal class MessageSourceFromSend(
-            override val messageUid: Int,
+            val messageRandom: Int,
             override val time: Long,
             override val senderId: Long,
             override val groupId: Long// ,
@@ -280,19 +280,20 @@ internal class MessageSvc {
         ) : MessageSource {
             private lateinit var sequenceIdDeferred: Deferred<Int>
 
+            @UseExperimental(ExperimentalCoroutinesApi::class)
+            override val id: Long
+                get() = sequenceIdDeferred.getCompleted().toLong().shl(32) or
+                        messageRandom.toLong().and(0xFFFFFFFF)
+
             @UseExperimental(MiraiExperimentalAPI::class)
             fun startWaitingSequenceId(contact: Contact) {
                 sequenceIdDeferred = contact.subscribingGetAsync<OnlinePush.PbPushGroupMsg.SendGroupMessageReceipt, Int> {
-                    if (it.messageRandom == messageUid) {
+                    if (it.messageRandom == this@MessageSourceFromSend.messageRandom) {
                         it.sequenceId
                     } else null
                 }
             }
 
-            @UseExperimental(ExperimentalCoroutinesApi::class)
-            override val sequenceId: Int
-                get() = sequenceIdDeferred.getCompleted()
-
             override suspend fun ensureSequenceIdAvailable() {
                 sequenceIdDeferred.join()
             }
@@ -309,7 +310,7 @@ internal class MessageSvc {
             crossinline sourceCallback: (MessageSource) -> Unit
         ): OutgoingPacket {
             val source = MessageSourceFromSend(
-                messageUid = Random.nextInt().absoluteValue,
+                messageRandom = Random.nextInt().absoluteValue,
                 senderId = client.uin,
                 time = currentTimeSeconds + client.timeDifference,
                 groupId = 0//
@@ -327,7 +328,7 @@ internal class MessageSvc {
             client: QQAndroidClient,
             toUin: Long,
             message: MessageChain,
-            source: MessageSource
+            source: MessageSourceFromSend
         ): OutgoingPacket = buildOutgoingUniPacket(client) {
             ///writeFully("0A 08 0A 06 08 89 FC A6 8C 0B 12 06 08 01 10 00 18 00 1A 1F 0A 1D 12 08 0A 06 0A 04 F0 9F 92 A9 12 11 AA 02 0E 88 01 00 9A 01 08 78 00 F8 01 00 C8 02 00 20 9B 7A 28 F4 CA 9B B8 03 32 34 08 92 C2 C4 F1 05 10 92 C2 C4 F1 05 18 E6 ED B9 C3 02 20 89 FE BE A4 06 28 89 84 F9 A2 06 48 DE 8C EA E5 0E 58 D9 BD BB A0 09 60 1D 68 92 C2 C4 F1 05 70 00 40 01".hexToBytes())
 
@@ -342,7 +343,7 @@ internal class MessageSvc {
                         )
                     ),
                     msgSeq = client.atomicNextMessageSequenceId(),
-                    msgRand = source.messageUid,
+                    msgRand = source.messageRandom,
                     syncCookie = SyncCookie(time = source.time).toByteArray(SyncCookie.serializer())
                     // msgVia = 1
                 )
@@ -358,7 +359,7 @@ internal class MessageSvc {
         ): OutgoingPacket {
 
             val source = MessageSourceFromSend(
-                messageUid = Random.nextInt().absoluteValue,
+                messageRandom = Random.nextInt().absoluteValue,
                 senderId = client.uin,
                 time = currentTimeSeconds + client.timeDifference,
                 groupId = groupCode//,
@@ -372,11 +373,11 @@ internal class MessageSvc {
          * 发送群消息
          */
         @Suppress("FunctionName")
-        fun ToGroup(
+        private fun ToGroup(
             client: QQAndroidClient,
             groupCode: Long,
             message: MessageChain,
-            source: MessageSource
+            source: MessageSourceFromSend
         ): OutgoingPacket = buildOutgoingUniPacket(client) {
             ///writeFully("0A 08 0A 06 08 89 FC A6 8C 0B 12 06 08 01 10 00 18 00 1A 1F 0A 1D 12 08 0A 06 0A 04 F0 9F 92 A9 12 11 AA 02 0E 88 01 00 9A 01 08 78 00 F8 01 00 C8 02 00 20 9B 7A 28 F4 CA 9B B8 03 32 34 08 92 C2 C4 F1 05 10 92 C2 C4 F1 05 18 E6 ED B9 C3 02 20 89 FE BE A4 06 28 89 84 F9 A2 06 48 DE 8C EA E5 0E 58 D9 BD BB A0 09 60 1D 68 92 C2 C4 F1 05 70 00 40 01".hexToBytes())
 
@@ -393,7 +394,7 @@ internal class MessageSvc {
                         )
                     ),
                     msgSeq = client.atomicNextMessageSequenceId(),
-                    msgRand = source.messageUid,
+                    msgRand = source.messageRandom,
                     syncCookie = EMPTY_BYTE_ARRAY,
                     msgVia = 1
                 )

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/Bot.kt

@@ -11,22 +11,24 @@
 
 package net.mamoe.mirai
 
+import io.ktor.utils.io.ByteReadChannel
+import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
-import kotlinx.io.OutputStream
+import kotlinx.coroutines.launch
-import kotlinx.io.core.ByteReadPacket
-import kotlinx.io.core.IoBuffer
-import kotlinx.io.core.use
 import net.mamoe.mirai.contact.*
 import net.mamoe.mirai.data.AddFriendResult
 import net.mamoe.mirai.data.FriendInfo
 import net.mamoe.mirai.data.GroupInfo
 import net.mamoe.mirai.data.MemberInfo
 import net.mamoe.mirai.message.data.Image
+import net.mamoe.mirai.message.data.MessageChain
+import net.mamoe.mirai.message.data.MessageSource
 import net.mamoe.mirai.network.BotNetworkHandler
 import net.mamoe.mirai.network.LoginFailedException
 import net.mamoe.mirai.utils.*
-import net.mamoe.mirai.utils.io.transferTo
+import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.EmptyCoroutineContext
 import kotlin.jvm.JvmStatic
 
 /**
@@ -208,13 +210,44 @@ abstract class Bot : CoroutineScope {
 
     // region actions
 
-    @Deprecated("内存使用效率十分低下", ReplaceWith("this.download()"), DeprecationLevel.WARNING)
+    /**
-    abstract suspend fun Image.downloadAsByteArray(): ByteArray
+     * 撤回这条消息.
+     *
+     * [Bot] 撤回自己的消息不需要权限.
+     * [Bot] 撤回群员的消息需要管理员权限.
+     *
+     * @throws PermissionDeniedException 当 [Bot] 无权限操作时
+     * @see Bot.recall (扩展函数) 接受参数 [MessageChain]
+     */// source.groupId, source.sequenceId, source.messageUid
+    abstract suspend fun recall(source: MessageSource)
 
     /**
-     * 将图片下载到内存中 (使用 [IoBuffer.Pool])
+     * 撤回一条消息. 可撤回自己 2 分钟内发出的消息, 和任意时间的群成员的消息.
+     *
+     * [Bot] 撤回自己的消息不需要权限.
+     * [Bot] 撤回群员的消息需要管理员权限.
+     *
+     * @param senderId 这条消息的发送人. 可以为 [Bot.uin] 或 [Member.id]
+     * @param messageId 即 [MessageSource.id]
+     *
+     * @throws PermissionDeniedException 当 [Bot] 无权限操作时
+     * @see Bot.recall (扩展函数) 接受参数 [MessageChain]
+     * @see recall 请优先使用这个函数
      */
-    abstract suspend fun Image.download(): ByteReadPacket
+    abstract suspend fun recall(groupId: Long, senderId: Long, messageId: Long)
+
+    /**
+     * 获取图片下载链接
+     */
+    abstract suspend fun queryImageUrl(image: Image): String
+
+    /**
+     * 获取图片下载链接并开始下载.
+     *
+     * @see ByteReadChannel.copyAndClose
+     * @see ByteReadChannel.copyTo
+     */
+    abstract suspend fun openChannel(image: Image): ByteReadChannel
 
     /**
      * 添加一个好友
@@ -222,11 +255,13 @@ abstract class Bot : CoroutineScope {
      * @param message 若需要验证请求时的验证消息.
      * @param remark 好友备注
      */
+    @MiraiExperimentalAPI("未支持")
     abstract suspend fun addFriend(id: Long, message: String? = null, remark: String? = null): AddFriendResult
 
     /**
      * 同意来自陌生人的加好友请求
      */
+    @MiraiExperimentalAPI("未支持")
     abstract suspend fun approveFriendAddRequest(id: Long, remark: String?)
 
     // endregion
@@ -243,19 +278,54 @@ abstract class Bot : CoroutineScope {
      */
     abstract fun close(cause: Throwable? = null)
 
-    // region extensions
+    final override fun toString(): String = "Bot(${uin})"
+}
 
-    final override fun toString(): String {
+/**
-        return "Bot(${uin})"
+ * 撤回这条消息.
-    }
+ * 根据 [message] 内的 [MessageSource] 进行相关判断.
+ *
+ * [Bot] 撤回自己的消息不需要权限.
+ * [Bot] 撤回群员的消息需要管理员权限.
+ *
+ * @throws PermissionDeniedException 当 [Bot] 无权限操作时
+ * @see Bot.recall
+ */
+@MiraiExperimentalAPI
+suspend inline fun Bot.recall(message: MessageChain) = this.recall(message[MessageSource])
 
-    /**
+/**
-     * 需要调用者自行 close [output]
+ * 在一段时间后撤回这条消息.
-     */
+ * 将根据 [MessageSource.groupId] 判断消息是群消息还是好友消息.
-    suspend inline fun Image.downloadTo(output: OutputStream) =
+ *
-        download().use { input -> input.transferTo(output) }
+ * @param millis 延迟的时间, 单位为毫秒
+ * @param coroutineContext 额外的 [CoroutineContext]
+ * @see recall
+ */
+fun Bot.recallIn(
+    source: MessageSource,
+    millis: Long,
+    coroutineContext: CoroutineContext = EmptyCoroutineContext
+): Job = this.launch(coroutineContext + CoroutineName("MessageRecall")) {
+    kotlinx.coroutines.delay(millis)
+    recall(source)
+}
 
-    // endregion
+/**
+ * 在一段时间后撤回这条消息.
+ *
+ * @param millis 延迟的时间, 单位为毫秒
+ * @param coroutineContext 额外的 [CoroutineContext]
+ * @see recall
+ */
+@MiraiExperimentalAPI
+fun Bot.recallIn(
+    message: MessageChain,
+    millis: Long,
+    coroutineContext: CoroutineContext = EmptyCoroutineContext
+): Job = this.launch(coroutineContext + CoroutineName("MessageRecall")) {
+    kotlinx.coroutines.delay(millis)
+    recall(message)
 }
 
 /**

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/contact/Contact.kt

@@ -12,6 +12,7 @@
 package net.mamoe.mirai.contact
 
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
 import net.mamoe.mirai.Bot
 import net.mamoe.mirai.event.events.BeforeImageUploadEvent
 import net.mamoe.mirai.event.events.EventCancelledException
@@ -20,8 +21,13 @@ import net.mamoe.mirai.event.events.MessageSendEvent.FriendMessageSendEvent
 import net.mamoe.mirai.event.events.MessageSendEvent.GroupMessageSendEvent
 import net.mamoe.mirai.message.MessageReceipt
 import net.mamoe.mirai.message.data.*
+import net.mamoe.mirai.recall
+import net.mamoe.mirai.recallIn
 import net.mamoe.mirai.utils.ExternalImage
+import net.mamoe.mirai.utils.MiraiExperimentalAPI
 import net.mamoe.mirai.utils.WeakRefProperty
+import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.EmptyCoroutineContext
 
 
 /**
@@ -92,6 +98,44 @@ interface Contact : CoroutineScope {
     override fun toString(): String
 }
 
-suspend inline fun Contact.sendMessage(message: Message) = sendMessage(message.toChain())
+/**
+ * @see Bot.recall
+ */
+@MiraiExperimentalAPI
+suspend inline fun Contact.recall(source: MessageChain) = this.bot.recall(source)
 
-suspend inline fun Contact.sendMessage(plain: String) = sendMessage(plain.toMessage())
+/**
+ * @see Bot.recall
+ */
+suspend inline fun Contact.recall(source: MessageSource) = this.bot.recall(source)
+
+/**
+ * @see Bot.recallIn
+ */
+@MiraiExperimentalAPI
+fun Contact.recallIn(
+    message: MessageChain,
+    millis: Long,
+    coroutineContext: CoroutineContext = EmptyCoroutineContext
+): Job = this.bot.recallIn(message, millis, coroutineContext)
+
+/**
+ * @see Bot.recallIn
+ */
+fun Contact.recallIn(
+    source: MessageSource,
+    millis: Long,
+    coroutineContext: CoroutineContext = EmptyCoroutineContext
+): Job = this.bot.recallIn(source, millis, coroutineContext)
+
+/**
+ * @see Contact.sendMessage
+ */
+@Suppress("UNCHECKED_CAST")
+suspend inline fun <C : Contact> C.sendMessage(message: Message): MessageReceipt<C> =
+    sendMessage(message.toChain()) as? MessageReceipt<C> ?: error("Internal class cast mistake")
+
+/**
+ * @see Contact.sendMessage
+ */
+suspend inline fun <C : Contact> C.sendMessage(plain: String): MessageReceipt<C> = sendMessage(plain.toMessage())

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/contact/Group.kt

@@ -11,10 +11,7 @@
 
 package net.mamoe.mirai.contact
 
-import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Job
-import kotlinx.coroutines.launch
 import net.mamoe.mirai.Bot
 import net.mamoe.mirai.data.MemberInfo
 import net.mamoe.mirai.event.events.*
@@ -22,10 +19,7 @@ import net.mamoe.mirai.event.events.MessageSendEvent.FriendMessageSendEvent
 import net.mamoe.mirai.event.events.MessageSendEvent.GroupMessageSendEvent
 import net.mamoe.mirai.message.MessageReceipt
 import net.mamoe.mirai.message.data.MessageChain
-import net.mamoe.mirai.message.data.MessageSource
 import net.mamoe.mirai.utils.MiraiExperimentalAPI
-import kotlin.coroutines.CoroutineContext
-import kotlin.coroutines.EmptyCoroutineContext
 import kotlin.jvm.JvmName
 
 /**
@@ -151,17 +145,6 @@ interface Group : Contact, CoroutineScope {
     @MiraiExperimentalAPI("还未支持")
     suspend fun quit(): Boolean
 
-    /**
-     * 撤回这条消息.
-     *
-     * [Bot] 撤回自己的消息不需要权限.
-     * [Bot] 撤回群员的消息需要管理员权限.
-     *
-     * @throws PermissionDeniedException 当 [Bot] 无权限操作时
-     * @see Group.recall (扩展函数) 接受参数 [MessageChain]
-     */
-    suspend fun recall(source: MessageSource)
-
     /**
      * 构造一个 [Member].
      * 非特殊情况请不要使用这个函数. 优先使用 [get].
@@ -226,49 +209,6 @@ interface Group : Contact, CoroutineScope {
     fun toFullString(): String = "Group(id=${this.id}, name=$name, owner=${owner.id}, members=${members.idContentString})"
 }
 
-/**
- * 撤回这条消息.
- *
- * [Bot] 撤回自己的消息不需要权限.
- * [Bot] 撤回群员的消息需要管理员权限.
- *
- * @throws PermissionDeniedException 当 [Bot] 无权限操作时
- * @see Group.recall
- */
-suspend inline fun Group.recall(message: MessageChain) = this.recall(message[MessageSource])
-
-/**
- * 在一段时间后撤回这条消息.
- *
- * @param millis 延迟的时间, 单位为毫秒
- * @param coroutineContext 额外的 [CoroutineContext]
- * @see recall
- */
-fun Group.recallIn(
-    message: MessageSource,
-    millis: Long,
-    coroutineContext: CoroutineContext = EmptyCoroutineContext
-): Job = this.launch(coroutineContext + CoroutineName("MessageRecall")) {
-    kotlinx.coroutines.delay(millis)
-    recall(message)
-}
-
-/**
- * 在一段时间后撤回这条消息.
- *
- * @param millis 延迟的时间, 单位为毫秒
- * @param coroutineContext 额外的 [CoroutineContext]
- * @see recall
- */
-fun Group.recallIn(
-    message: MessageChain,
-    millis: Long,
-    coroutineContext: CoroutineContext = EmptyCoroutineContext
-): Job = this.launch(coroutineContext + CoroutineName("MessageRecall")) {
-    kotlinx.coroutines.delay(millis)
-    recall(message)
-}
-
 /**
  * 返回机器人是否正在被禁言
  *

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/event/internal/InternalEventListeners.kt

@@ -10,6 +10,8 @@
 package net.mamoe.mirai.event.internal
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import net.mamoe.mirai.event.Event
 import net.mamoe.mirai.event.EventDisabled
 import net.mamoe.mirai.event.Listener
@@ -73,6 +75,9 @@ internal class Handler<in E : Event>
             ListeningStatus.LISTENING
         }
     }
+
+    @MiraiInternalAPI
+    override val lock: Mutex = Mutex()
 }
 
 /**
@@ -138,23 +143,29 @@ internal object EventListenerManager {
 
 // inline: NO extra Continuation
 @Suppress("UNCHECKED_CAST")
-internal suspend inline fun Event.broadcastInternal() {
+internal suspend inline fun Event.broadcastInternal() = coroutineScope {
-    if (EventDisabled) return
+    if (EventDisabled) return@coroutineScope
 
     EventLogger.info { "Event broadcast: $this" }
 
-    val listeners = this::class.listeners()
+    val listeners = this@broadcastInternal::class.listeners()
-    callAndRemoveIfRequired(listeners)
+    callAndRemoveIfRequired(this@broadcastInternal, listeners)
     listeners.supertypes.forEach {
-        callAndRemoveIfRequired(it.listeners())
+        callAndRemoveIfRequired(this@broadcastInternal, it.listeners())
     }
 }
 
-private suspend inline fun <E : Event> E.callAndRemoveIfRequired(listeners: EventListeners<E>) {
+@UseExperimental(MiraiInternalAPI::class)
+private fun <E : Event> CoroutineScope.callAndRemoveIfRequired(event: E, listeners: EventListeners<E>) {
     // atomic foreach
-    listeners.forEach {
+    listeners.forEachNode { node ->
-        if (it.onEvent(this) == ListeningStatus.STOPPED) {
+        launch {
-            listeners.remove(it) // atomic remove
+            val listener = node.nodeValue
+            listener.lock.withLock {
+                if (!node.isRemoved() && listener.onEvent(event) == ListeningStatus.STOPPED) {
+                    listeners.remove(listener) // atomic remove
+                }
+            }
         }
     }
 }

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/event/subscriber.kt

@@ -13,6 +13,7 @@ import kotlinx.coroutines.CompletableJob
 import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.sync.Mutex
 import net.mamoe.mirai.Bot
 import net.mamoe.mirai.event.events.BotEvent
 import net.mamoe.mirai.event.internal.Handler
@@ -51,6 +52,12 @@ enum class ListeningStatus {
  * 取消监听: [complete]
  */
 interface Listener<in E : Event> : CompletableJob {
+    /**
+     * [onEvent] 的锁
+     */
+    @MiraiInternalAPI
+    val lock: Mutex
+
     suspend fun onEvent(event: E): ListeningStatus
 }
 

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/message/GroupMessage.kt

@@ -11,11 +11,16 @@ package net.mamoe.mirai.message
 
 import kotlinx.coroutines.Job
 import net.mamoe.mirai.Bot
-import net.mamoe.mirai.contact.*
+import net.mamoe.mirai.contact.Group
+import net.mamoe.mirai.contact.Member
+import net.mamoe.mirai.contact.MemberPermission
 import net.mamoe.mirai.event.Event
 import net.mamoe.mirai.message.data.Message
 import net.mamoe.mirai.message.data.MessageChain
 import net.mamoe.mirai.message.data.MessageSource
+import net.mamoe.mirai.recall
+import net.mamoe.mirai.recallIn
+import net.mamoe.mirai.utils.MiraiExperimentalAPI
 import net.mamoe.mirai.utils.getValue
 import net.mamoe.mirai.utils.unsafeWeakRef
 import kotlin.jvm.JvmName
@@ -59,10 +64,13 @@ class GroupMessage(
     @JvmName("reply2")
     suspend inline fun MessageChain.quoteReply(): MessageReceipt<Group> = quoteReply(this)
 
-    suspend inline fun MessageChain.recall() = group.recall(this)
+    @MiraiExperimentalAPI
-    suspend inline fun MessageSource.recall() = group.recall(this)
+    suspend inline fun MessageChain.recall() = bot.recall(this)
-    inline fun MessageSource.recallIn(delay: Long): Job = group.recallIn(this, delay)
+
-    inline fun MessageChain.recallIn(delay: Long): Job = group.recallIn(this, delay)
+    suspend inline fun MessageSource.recall() = bot.recall(this)
+    inline fun MessageSource.recallIn(delay: Long): Job = bot.recallIn(this, delay)
+    @MiraiExperimentalAPI
+    inline fun MessageChain.recallIn(delay: Long): Job = bot.recallIn(this, delay)
 
     override fun toString(): String =
         "GroupMessage(group=${group.id}, senderName=$senderName, sender=${sender.id}, permission=${permission.name}, message=$message)"

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/message/MessagePacket.kt

@@ -11,9 +11,7 @@
 
 package net.mamoe.mirai.message
 
-import kotlinx.io.core.ByteReadPacket
+import io.ktor.utils.io.ByteReadChannel
-import kotlinx.io.core.IoBuffer
-import kotlinx.io.core.readBytes
 import net.mamoe.mirai.Bot
 import net.mamoe.mirai.contact.Contact
 import net.mamoe.mirai.contact.Group
@@ -22,7 +20,7 @@ import net.mamoe.mirai.contact.QQ
 import net.mamoe.mirai.data.Packet
 import net.mamoe.mirai.event.events.BotEvent
 import net.mamoe.mirai.event.subscribingGet
-import net.mamoe.mirai.event.subscribingGetAsync
+import net.mamoe.mirai.event.subscribingGetOrNull
 import net.mamoe.mirai.message.data.*
 import net.mamoe.mirai.utils.*
 import kotlin.jvm.JvmName
@@ -128,20 +126,22 @@ abstract class MessagePacketBase<TSender : QQ, TSubject : Contact> : Packet, Bot
     // endregion
 
     // region 下载图片
+
+
     /**
-     * 将图片下载到内存.
+     * 获取图片下载链接
      *
-     * 非常不推荐这样做.
+     * @return "http://gchat.qpic.cn/gchatpic_new/..."
      */
-    @Deprecated("内存使用效率十分低下", ReplaceWith("this.download()"), DeprecationLevel.WARNING)
+    suspend inline fun Image.url(): String = bot.queryImageUrl(this@url)
-    suspend inline fun Image.downloadAsByteArray(): ByteArray = bot.run { download().readBytes() }
-
-    // TODO: 2020/2/5 为下载图片添加文件系统的存储方式
 
     /**
-     * 将图片下载到内存缓存中 (使用 [IoBuffer.Pool])
+     * 获取图片下载链接并开始下载.
+     *
+     * @see ByteReadChannel.copyAndClose
+     * @see ByteReadChannel.copyTo
      */
-    suspend inline fun Image.download(): ByteReadPacket = bot.run { download() }
+    suspend inline fun Image.channel(): ByteReadChannel = bot.openChannel(this)
     // endregion
 }
 
@@ -153,14 +153,14 @@ fun MessagePacket<*, *>.isContextIdenticalWith(another: MessagePacket<*, *>): Bo
 }
 
 /**
- * 挂起当前协程, 等待下一条 [MessagePacket.sender] 和 [MessagePacket.subject] 与 [P] 相同且通过 [筛选][filter] 的 [MessagePacket]
+ * 挂起当前协程, 等待下一条 [MessagePacket.sender] 和 [MessagePacket.subject] 与 [this] 相同且通过 [筛选][filter] 的 [MessagePacket]
  *
  * 若 [filter] 抛出了一个异常, 本函数会立即抛出这个异常.
  *
  * @param timeoutMillis 超时. 单位为毫秒. `-1` 为不限制
  * @param filter 过滤器. 返回非 null 则代表得到了需要的值. [subscribingGet] 会返回这个值
  *
- * @see subscribingGetAsync 本函数的异步版本
+ * @see subscribingGet
  */
 suspend inline fun <reified P : MessagePacket<*, *>> P.nextMessage(
     timeoutMillis: Long = -1,
@@ -172,13 +172,33 @@ suspend inline fun <reified P : MessagePacket<*, *>> P.nextMessage(
 }
 
 /**
- * 挂起当前协程, 等待下一条 [MessagePacket.sender] 和 [MessagePacket.subject] 与 [P] 相同的 [MessagePacket]
+ * 挂起当前协程, 等待下一条 [MessagePacket.sender] 和 [MessagePacket.subject] 与 [this] 相同且通过 [筛选][filter] 的 [MessagePacket]
+ *
+ * 若 [filter] 抛出了一个异常, 本函数会立即抛出这个异常.
+ *
+ * @param timeoutMillis 超时. 单位为毫秒. `-1` 为不限制
+ * @param filter 过滤器. 返回非 null 则代表得到了需要的值. [subscribingGet] 会返回这个值
+ * @return 消息链. 超时时返回 `null`
+ *
+ * @see subscribingGetOrNull
+ */
+suspend inline fun <reified P : MessagePacket<*, *>> P.nextMessageOrNull(
+    timeoutMillis: Long = -1,
+    crossinline filter: P.(P) -> Boolean
+): MessageChain? {
+    return subscribingGetOrNull<P, P>(timeoutMillis) {
+        takeIf { this.isContextIdenticalWith(this@nextMessageOrNull) }?.takeIf { filter(it, it) }
+    }?.message
+}
+
+/**
+ * 挂起当前协程, 等待下一条 [MessagePacket.sender] 和 [MessagePacket.subject] 与 [this] 相同的 [MessagePacket]
  *
  * 若 [filter] 抛出了一个异常, 本函数会立即抛出这个异常.
  *
  * @param timeoutMillis 超时. 单位为毫秒. `-1` 为不限制
  *
- * @see subscribingGetAsync 本函数的异步版本
+ * @see subscribingGet
  */
 suspend inline fun <reified P : MessagePacket<*, *>> P.nextMessage(
     timeoutMillis: Long = -1
@@ -187,3 +207,56 @@ suspend inline fun <reified P : MessagePacket<*, *>> P.nextMessage(
         takeIf { this.isContextIdenticalWith(this@nextMessage) }
     }.message
 }
+
+/**
+ * 挂起当前协程, 等待下一条 [MessagePacket.sender] 和 [MessagePacket.subject] 与 [this] 相同的 [MessagePacket]
+ *
+ * 若 [filter] 抛出了一个异常, 本函数会立即抛出这个异常.
+ *
+ * @param timeoutMillis 超时. 单位为毫秒. `-1` 为不限制
+ * @return 消息链. 超时时返回 `null`
+ *
+ * @see subscribingGetOrNull
+ */
+suspend inline fun <reified P : MessagePacket<*, *>> P.nextMessageOrNull(
+    timeoutMillis: Long = -1
+): MessageChain? {
+    return subscribingGetOrNull<P, P>(timeoutMillis) {
+        takeIf { this.isContextIdenticalWith(this@nextMessageOrNull) }
+    }?.message
+}
+
+/**
+ * 挂起当前协程, 等待下一条 [MessagePacket.sender] 和 [MessagePacket.subject] 与 [this] 相同的 [MessagePacket]
+ *
+ * 若 [filter] 抛出了一个异常, 本函数会立即抛出这个异常.
+ *
+ * @param timeoutMillis 超时. 单位为毫秒. `-1` 为不限制
+ *
+ * @see subscribingGet
+ */
+suspend inline fun <reified M : Message> MessagePacket<*, *>.nextMessageContaining(
+    timeoutMillis: Long = -1
+): M {
+    return subscribingGet<MessagePacket<*, *>, MessagePacket<*, *>>(timeoutMillis) {
+        takeIf { this.isContextIdenticalWith(this@nextMessageContaining) }
+    }.message.first()
+}
+
+/**
+ * 挂起当前协程, 等待下一条 [MessagePacket.sender] 和 [MessagePacket.subject] 与 [this] 相同并含有 [M] 类型的消息的 [MessagePacket]
+ *
+ * 若 [filter] 抛出了一个异常, 本函数会立即抛出这个异常.
+ *
+ * @param timeoutMillis 超时. 单位为毫秒. `-1` 为不限制
+ * @return 指定类型的消息. 超时时返回 `null`
+ *
+ * @see subscribingGetOrNull
+ */
+suspend inline fun <reified M : Message> MessagePacket<*, *>.nextMessageContainingOrNull(
+    timeoutMillis: Long = -1
+): M? {
+    return subscribingGetOrNull<MessagePacket<*, *>, MessagePacket<*, *>>(timeoutMillis) {
+        takeIf { this.isContextIdenticalWith(this@nextMessageContainingOrNull) }
+    }?.message?.first()
+}

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/message/MessageReceipt.kt

@@ -12,8 +12,11 @@ package net.mamoe.mirai.message
 import kotlinx.atomicfu.atomic
 import kotlinx.coroutines.Job
 import net.mamoe.mirai.Bot
-import net.mamoe.mirai.contact.*
+import net.mamoe.mirai.contact.Contact
+import net.mamoe.mirai.contact.Group
+import net.mamoe.mirai.contact.QQ
 import net.mamoe.mirai.message.data.*
+import net.mamoe.mirai.recallIn
 import net.mamoe.mirai.utils.MiraiExperimentalAPI
 import net.mamoe.mirai.utils.getValue
 import net.mamoe.mirai.utils.unsafeWeakRef
@@ -27,7 +30,6 @@ import net.mamoe.mirai.utils.unsafeWeakRef
  * @see QQ.sendMessage 发送群消息, 返回回执（此对象）
  */
 open class MessageReceipt<C : Contact>(
-    val originalMessage: MessageChain,
     private val source: MessageSource,
     target: C
 ) {
@@ -54,7 +56,7 @@ open class MessageReceipt<C : Contact>(
         if (_isRecalled.compareAndSet(false, true)) {
             when (val contact = target) {
                 is Group -> {
-                    contact.recall(source)
+                    contact.bot.recall(source)
                 }
                 is QQ -> {
                     TODO()
@@ -77,7 +79,7 @@ open class MessageReceipt<C : Contact>(
         if (_isRecalled.compareAndSet(false, true)) {
             when (val contact = target) {
                 is Group -> {
-                    return contact.recallIn(source, millis)
+                    return contact.bot.recallIn(source, millis)
                 }
                 is QQ -> {
                     TODO()

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/message/data/MessageSource.kt

@@ -27,9 +27,11 @@ interface MessageSource : Message {
     companion object Key : Message.Key<MessageSource>
 
     /**
-     * 序列号. 若是机器人发出去的消息, 请先 [确保 sequenceId 可用][ensureSequenceIdAvailable]
+     * 在 Mirai 中使用的 id.
+     * 高 32 位为 [sequenceId],
+     * 低 32 位为 [messageRandom]
      */
-    val sequenceId: Int
+    val id: Long
 
     /**
      * 等待 [sequenceId] 获取, 确保其可用.
@@ -38,11 +40,6 @@ interface MessageSource : Message {
      */
     suspend fun ensureSequenceIdAvailable()
 
-    /**
-     * 实际上是个随机数, 但服务器确实是用它当做 uid
-     */
-    val messageUid: Int
-
     /**
      * 发送时间, 单位为秒
      */
@@ -65,11 +62,33 @@ interface MessageSource : Message {
 }
 
 /**
- * 消息唯一标识符. 实际上是个随机数, 但服务器确实是用它当做 uid
+ * 序列号. 若是机器人发出去的消息, 请先 [确保 sequenceId 可用][MessageSource.ensureSequenceIdAvailable]
+ * @see MessageSource.id
  */
-val MessageChain.messageUid get() = this[MessageSource].messageUid
+val MessageSource.sequenceId: Int get() = (this.id shr 32).toInt()
+
+/**
+ * 消息随机数. 由服务器或客户端指定后不能更改. 它是消息 id 的一部分.
+ * @see MessageSource.id
+ */
+val MessageSource.messageRandom: Int get() = this.id.toInt()
+
+// For MessageChain
+
+/**
+ * 消息 id.
+ * @see MessageSource.id
+ */
+val MessageChain.id: Long get() = this[MessageSource].id
 
 /**
  * 消息序列号, 可能来自服务器也可以发送时赋值, 不唯一.
+ * @see MessageSource.id
  */
-val MessageChain.sequenceId get() = this[MessageSource].sequenceId
+val MessageChain.sequenceId: Int get() = this[MessageSource].sequenceId
+
+/**
+ * 消息随机数. 由服务器或客户端指定后不能更改. 它是消息 id 的一部分.
+ * @see MessageSource.id
+ */
+val MessageChain.messageRandom: Int get() = this[MessageSource].messageRandom

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/utils/LockFreeLinkedList.kt

@@ -132,7 +132,7 @@ open class LockFreeLinkedList<E> {
         addLastNode(element.asNode(tail))
     }
 
-    private fun addLastNode(node: Node<E>) {
+    private fun addLastNode(node: LockFreeLinkedListNode<E>) {
         while (true) {
             val tail = head.iterateBeforeFirst { it === tail } // find the last node.
             if (tail.nextNodeRef.compareAndSet(this.tail, node)) { // ensure the last node is the last node
@@ -146,9 +146,9 @@ open class LockFreeLinkedList<E> {
      */
     @Suppress("DuplicatedCode")
     open fun addAll(iterable: Iterable<E>) {
-        var firstNode: Node<E>? = null
+        var firstNode: LockFreeLinkedListNode<E>? = null
 
-        var currentNode: Node<E>? = null
+        var currentNode: LockFreeLinkedListNode<E>? = null
         iterable.forEach {
             val nextNode = it.asNode(tail)
             if (firstNode == null) {
@@ -166,9 +166,9 @@ open class LockFreeLinkedList<E> {
      */
     @Suppress("DuplicatedCode")
     open fun addAll(iterable: Sequence<E>) {
-        var firstNode: Node<E>? = null
+        var firstNode: LockFreeLinkedListNode<E>? = null
 
-        var currentNode: Node<E>? = null
+        var currentNode: LockFreeLinkedListNode<E>? = null
         iterable.forEach {
             val nextNode = it.asNode(tail)
             if (firstNode == null) {
@@ -190,7 +190,7 @@ open class LockFreeLinkedList<E> {
         val node = LazyNode(tail, supplier)
 
         while (true) {
-            var current: Node<E> = head
+            var current: LockFreeLinkedListNode<E> = head
 
             findLastNode@ while (true) {
                 if (current.isValidElementNode() && filter(current.nodeValue))
@@ -208,13 +208,14 @@ open class LockFreeLinkedList<E> {
     }
 
     @PublishedApi // limitation by atomicfu
-    internal fun <E> Node<E>.compareAndSetNextNodeRef(expect: Node<E>, update: Node<E>) = this.nextNodeRef.compareAndSet(expect, update)
+    internal fun <E> LockFreeLinkedListNode<E>.compareAndSetNextNodeRef(expect: LockFreeLinkedListNode<E>, update: LockFreeLinkedListNode<E>) =
+        this.nextNodeRef.compareAndSet(expect, update)
 
     override fun toString(): String = joinToString()
 
     @Suppress("unused")
     internal fun getLinkStructure(): String = buildString {
-        head.childIterateReturnsLastSatisfying<Node<*>>({
+        head.childIterateReturnsLastSatisfying<LockFreeLinkedListNode<*>>({
             append(it.toString())
             append(" <- ")
             it.nextNode
@@ -240,7 +241,7 @@ open class LockFreeLinkedList<E> {
 
 
             // physically remove: try to fix the link
-            var next: Node<E> = toRemove.nextNode
+            var next: LockFreeLinkedListNode<E> = toRemove.nextNode
             while (next !== tail && next.isRemoved()) {
                 next = next.nextNode
             }
@@ -269,7 +270,7 @@ open class LockFreeLinkedList<E> {
 
 
             // physically remove: try to fix the link
-            var next: Node<E> = toRemove.nextNode
+            var next: LockFreeLinkedListNode<E> = toRemove.nextNode
             while (next !== tail && next.isRemoved()) {
                 next = next.nextNode
             }
@@ -282,7 +283,7 @@ open class LockFreeLinkedList<E> {
     /**
      * 动态计算的大小
      */
-    val size: Int get() = head.countChildIterate<Node<E>>({ it.nextNode }, { it !is Tail }) - 1 // empty head is always included
+    val size: Int get() = head.countChildIterate<LockFreeLinkedListNode<E>>({ it.nextNode }, { it !is Tail }) - 1 // empty head is always included
 
     open operator fun contains(element: E): Boolean {
         forEach { if (it == element) return true }
@@ -295,7 +296,7 @@ open class LockFreeLinkedList<E> {
     open fun isEmpty(): Boolean = head.allMatching { it.isValidElementNode().not() }
 
     inline fun forEach(block: (E) -> Unit) {
-        var node: Node<E> = head
+        var node: LockFreeLinkedListNode<E> = head
         while (true) {
             if (node === tail) return
             node.letValueIfValid(block)
@@ -303,6 +304,15 @@ open class LockFreeLinkedList<E> {
         }
     }
 
+    inline fun forEachNode(block: (LockFreeLinkedListNode<E>) -> Unit) {
+        var node: LockFreeLinkedListNode<E> = head
+        while (true) {
+            if (node === tail) return
+            node.letValueIfValid { block(node) }
+            node = node.nextNode
+        }
+    }
+
     @Suppress("unused")
     open fun clear() {
         val first = head.nextNode
@@ -638,14 +648,14 @@ open class LockFreeLinkedList<E> {
 // region internal
 
 @Suppress("NOTHING_TO_INLINE")
-private inline fun <E> E.asNode(nextNode: Node<E>): Node<E> = Node(nextNode, this)
+private inline fun <E> E.asNode(nextNode: LockFreeLinkedListNode<E>): LockFreeLinkedListNode<E> = LockFreeLinkedListNode(nextNode, this)
 
 /**
  * Self-iterate using the [iterator], until [mustBeTrue] returns `false`.
  * Returns the element at the last time when the [mustBeTrue] returns `true`
  */
 @PublishedApi
-internal inline fun <N : Node<*>> N.childIterateReturnsLastSatisfying(iterator: (N) -> N, mustBeTrue: (N) -> Boolean): N {
+internal inline fun <N : LockFreeLinkedListNode<*>> N.childIterateReturnsLastSatisfying(iterator: (N) -> N, mustBeTrue: (N) -> Boolean): N {
     if (!mustBeTrue(this)) return this
     var value: N = this
 
@@ -703,9 +713,9 @@ private inline fun <E> E.countChildIterate(iterator: (E) -> E, mustBeTrue: (E) -
 
 @PublishedApi
 internal class LazyNode<E> @PublishedApi internal constructor(
-    nextNode: Node<E>,
+    nextNode: LockFreeLinkedListNode<E>,
     private val valueComputer: () -> E
-) : Node<E>(nextNode, null) {
+) : LockFreeLinkedListNode<E>(nextNode, null) {
     private val initialized = atomic(false)
 
     private val value: AtomicRef<E?> = atomic(null)
@@ -727,20 +737,19 @@ internal class LazyNode<E> @PublishedApi internal constructor(
 }
 
 @PublishedApi
-internal class Head<E>(nextNode: Node<E>) : Node<E>(nextNode, null) {
+internal class Head<E>(nextNode: LockFreeLinkedListNode<E>) : LockFreeLinkedListNode<E>(nextNode, null) {
     override fun toString(): String = "Head"
     override val nodeValue: Nothing get() = error("Internal error: trying to get the value of a Head")
 }
 
 @PublishedApi
-internal open class Tail<E> : Node<E>(null, null) {
+internal open class Tail<E> : LockFreeLinkedListNode<E>(null, null) {
     override fun toString(): String = "Tail"
     override val nodeValue: Nothing get() = error("Internal error: trying to get the value of a Tail")
 }
 
-@PublishedApi
+open class LockFreeLinkedListNode<E>(
-internal open class Node<E>(
+    nextNode: LockFreeLinkedListNode<E>?,
-    nextNode: Node<E>?,
     private var initialNodeValue: E?
 ) {
     /*
@@ -754,10 +763,11 @@ internal open class Node<E>(
 
     open val nodeValue: E get() = initialNodeValue ?: error("Internal error: nodeValue is not initialized")
 
-    val removed = atomic(false)
+    @PublishedApi
+    internal val removed = atomic(false)
 
     @Suppress("LeakingThis")
-    val nextNodeRef: AtomicRef<Node<E>> = atomic(nextNode ?: this)
+    internal val nextNodeRef: AtomicRef<LockFreeLinkedListNode<E>> = atomic(nextNode ?: this)
 
     inline fun <R> letValueIfValid(block: (E) -> R): R? {
         if (!this.isValidElementNode()) {
@@ -770,7 +780,8 @@ internal open class Node<E>(
     /**
      * Short cut for accessing [nextNodeRef]
      */
-    var nextNode: Node<E>
+    @PublishedApi
+    internal var nextNode: LockFreeLinkedListNode<E>
         get() = nextNodeRef.value
         set(value) {
             nextNodeRef.value = value
@@ -779,7 +790,7 @@ internal open class Node<E>(
     /**
      * Returns the former node of the last node whence [filter] returns true
      */
-    inline fun iterateBeforeFirst(filter: (Node<E>) -> Boolean): Node<E> =
+    inline fun iterateBeforeFirst(filter: (LockFreeLinkedListNode<E>) -> Boolean): LockFreeLinkedListNode<E> =
         this.childIterateReturnsLastSatisfying({ it.nextNode }, { !filter(it) })
 
     /**
@@ -788,7 +799,8 @@ internal open class Node<E>(
      * Head, which is this, is also being tested.
      * [Tail], is not being tested.
      */
-    inline fun allMatching(condition: (Node<E>) -> Boolean): Boolean = this.childIterateReturnsLastSatisfying({ it.nextNode }, condition) !is Tail
+    inline fun allMatching(condition: (LockFreeLinkedListNode<E>) -> Boolean): Boolean =
+        this.childIterateReturnsLastSatisfying({ it.nextNode }, condition) !is Tail
 
     /**
      * Stop on and returns the former element of the element that is [equals] to the [element]
@@ -796,23 +808,23 @@ internal open class Node<E>(
      * E.g.: for `head <- 1 <- 2 <- 3 <- tail`, `iterateStopOnNodeValue(2)` returns the node whose value is 1
      */
     @Suppress("NOTHING_TO_INLINE")
-    internal inline fun iterateBeforeNodeValue(element: E): Node<E> = this.iterateBeforeFirst { it.isValidElementNode() && it.nodeValue == element }
+    internal inline fun iterateBeforeNodeValue(element: E): LockFreeLinkedListNode<E> =
+        this.iterateBeforeFirst { it.isValidElementNode() && it.nodeValue == element }
 
 }
 
-@PublishedApi // DO NOT INLINE: ATOMIC OPERATION
+fun <E> LockFreeLinkedListNode<E>.isRemoved() = this.removed.value
-internal fun <E> Node<E>.isRemoved() = this.removed.value
 
 @PublishedApi
 @Suppress("NOTHING_TO_INLINE")
-internal inline fun Node<*>.isValidElementNode(): Boolean = !isHead() && !isTail() && !isRemoved()
+internal inline fun LockFreeLinkedListNode<*>.isValidElementNode(): Boolean = !isHead() && !isTail() && !isRemoved()
 
 @PublishedApi
 @Suppress("NOTHING_TO_INLINE")
-internal inline fun Node<*>.isHead(): Boolean = this is Head
+internal inline fun LockFreeLinkedListNode<*>.isHead(): Boolean = this is Head
 
 @PublishedApi
 @Suppress("NOTHING_TO_INLINE")
-internal inline fun Node<*>.isTail(): Boolean = this is Tail
+internal inline fun LockFreeLinkedListNode<*>.isTail(): Boolean = this is Tail
 
 // end region

mirai-core/src/commonMain/kotlin/net.mamoe.mirai/utils/channels.kt

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2020 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/master/LICENSE
+ */
+
+
+@file:JvmName("Utils")
+@file:JvmMultifileClass
+
+package net.mamoe.mirai.utils
+
+
+import io.ktor.utils.io.ByteReadChannel
+import io.ktor.utils.io.readAvailable
+import kotlinx.io.OutputStream
+import kotlinx.io.core.Output
+import kotlinx.io.pool.useInstance
+import net.mamoe.mirai.utils.io.ByteArrayPool
+import kotlin.jvm.JvmMultifileClass
+import kotlin.jvm.JvmName
+
+// copyTo
+
+/**
+ * 从接收者管道读取所有数据并写入 [dst]. 不会关闭 [dst]
+ */
+suspend fun ByteReadChannel.copyTo(dst: OutputStream) {
+    ByteArrayPool.useInstance {
+        do {
+            val size = this.readAvailable(it)
+            dst.write(it, 0, size)
+        } while (size != 0)
+    }
+}
+
+/**
+ * 从接收者管道读取所有数据并写入 [dst]. 不会关闭 [dst]
+ */
+suspend fun ByteReadChannel.copyTo(dst: Output) {
+    ByteArrayPool.useInstance {
+        do {
+            val size = this.readAvailable(it)
+            dst.writeFully(it, 0, size)
+        } while (size != 0)
+    }
+}
+
+
+/* // 垃圾 kotlin, Unresolved reference: ByteWriteChannel
+/**
+ * 从接收者管道读取所有数据并写入 [dst]. 不会关闭 [dst]
+ */
+suspend fun ByteReadChannel.copyTo(dst: kotlinx.coroutines.io.ByteWriteChannel) {
+    ByteArrayPool.useInstance {
+        do {
+            val size = this.readAvailable(it)
+            dst.writeFully(it, 0, size)
+        } while (size != 0)
+    }
+}
+*/
+
+// copyAndClose
+
+
+/**
+ * 从接收者管道读取所有数据并写入 [dst], 最终关闭 [dst]
+ */
+suspend fun ByteReadChannel.copyAndClose(dst: OutputStream) {
+    try {
+        ByteArrayPool.useInstance {
+            do {
+                val size = this.readAvailable(it)
+                dst.write(it, 0, size)
+            } while (size != 0)
+        }
+    } finally {
+        dst.close()
+    }
+}
+
+/**
+ * 从接收者管道读取所有数据并写入 [dst], 最终关闭 [dst]
+ */
+suspend fun ByteReadChannel.copyAndClose(dst: Output) {
+    try {
+        ByteArrayPool.useInstance {
+            do {
+                val size = this.readAvailable(it)
+                dst.writeFully(it, 0, size)
+            } while (size != 0)
+        }
+    } finally {
+        dst.close()
+    }
+}
+
+/*// 垃圾 kotlin, Unresolved reference: ByteWriteChannel
+/**
+ * 从接收者管道读取所有数据并写入 [dst], 最终关闭 [dst]
+ */
+suspend fun ByteReadChannel.copyAndClose(dst: kotlinx.coroutines.io.ByteWriteChannel) {
+    dst.close(kotlin.runCatching {
+        ByteArrayPool.useInstance {
+            do {
+                val size = this.readAvailable(it)
+                dst.writeFully(it, 0, size)
+            } while (size != 0)
+        }
+    }.exceptionOrNull())
+}
+
+ */

mirai-core/src/jvmMain/kotlin/net/mamoe/mirai/message/MessagePacket.kt

@@ -11,23 +11,19 @@
 
 package net.mamoe.mirai.message
 
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.withContext
 import kotlinx.io.core.Input
 import kotlinx.io.core.use
-import kotlinx.io.streams.inputStream
 import net.mamoe.mirai.contact.Contact
 import net.mamoe.mirai.contact.QQ
 import net.mamoe.mirai.message.data.Image
-import net.mamoe.mirai.utils.ExternalImage
 import net.mamoe.mirai.utils.MiraiInternalAPI
-import net.mamoe.mirai.utils.toExternalImage
+import net.mamoe.mirai.utils.copyAndClose
+import net.mamoe.mirai.utils.copyTo
 import java.awt.image.BufferedImage
 import java.io.File
 import java.io.InputStream
 import java.io.OutputStream
 import java.net.URL
-import javax.imageio.ImageIO
 
 /**
  * 一条从服务器接收到的消息事件.
@@ -72,16 +68,22 @@ actual abstract class MessagePacket<TSender : QQ, TSubject : Contact> actual con
     // endregion 发送图片 (扩展)
 
     // region 下载图片 (扩展)
-    suspend inline fun Image.downloadTo(file: File): Long = file.outputStream().use { downloadTo(it) }
+    suspend inline fun Image.downloadTo(file: File) = file.outputStream().use { downloadTo(it) }
 
     /**
-     * 这个函数结束后不会关闭 [output]. 请务必解决好 [OutputStream.close]
+     * 下载图片到 [output] 但不关闭这个 [output]
      */
-    suspend inline fun Image.downloadTo(output: OutputStream): Long =
+    suspend inline fun Image.downloadTo(output: OutputStream) = channel().copyTo(output)
-        download().inputStream().use { input -> withContext(Dispatchers.IO) { input.copyTo(output) } }
 
-    suspend inline fun Image.downloadAsStream(): InputStream = download().inputStream()
+    /**
-    suspend inline fun Image.downloadAsExternalImage(): ExternalImage = withContext(Dispatchers.IO) { download().toExternalImage() }
+     * 下载图片到 [output] 并关闭这个 [output]
+     */
+    suspend inline fun Image.downloadAndClose(output: OutputStream) = channel().copyAndClose(output)
+
+    /*
+    suspend inline fun Image.downloadAsStream(): InputStream = channel().asInputStream()
+    suspend inline fun Image.downloadAsExternalImage(): ExternalImage = withContext(Dispatchers.IO) { downloadAsStream().toExternalImage() }
     suspend inline fun Image.downloadAsBufferedImage(): BufferedImage = withContext(Dispatchers.IO) { ImageIO.read(downloadAsStream()) }
+     */
     // endregion
 }

mirai-japt/src/main/java/net/mamoe/mirai/japt/BlockingBot.java

@@ -1,6 +1,5 @@
 package net.mamoe.mirai.japt;
 
-import kotlinx.io.core.ByteReadPacket;
 import net.mamoe.mirai.Bot;
 import net.mamoe.mirai.BotAccount;
 import net.mamoe.mirai.BotFactoryJvmKt;
@@ -153,18 +152,16 @@ public interface BlockingBot {
 
     // region actions
 
-    @NotNull
-    byte[] downloadAsByteArray(@NotNull Image image);
-
-    @NotNull
-    ByteReadPacket download(@NotNull Image image);
-
     /**
      * 下载图片到 {@code outputStream}.
-     * <p>
      * 不会自动关闭 {@code outputStream}
      */
-    void download(@NotNull Image image, @NotNull OutputStream outputStream);
+    void downloadTo(@NotNull Image image, @NotNull OutputStream outputStream);
+
+    /**
+     * 下载图片到 {@code outputStream} 并关闭 stream
+     */
+    void downloadAndClose(@NotNull Image image, @NotNull OutputStream outputStream);
 
     /**
      * 添加一个好友

mirai-japt/src/main/kotlin/net/mamoe/mirai/japt/internal/BlockingBotImpl.kt

@@ -10,8 +10,6 @@
 package net.mamoe.mirai.japt.internal
 
 import kotlinx.coroutines.runBlocking
-import kotlinx.io.core.ByteReadPacket
-import kotlinx.io.core.readBytes
 import net.mamoe.mirai.Bot
 import net.mamoe.mirai.BotAccount
 import net.mamoe.mirai.contact.QQ
@@ -23,14 +21,12 @@ import net.mamoe.mirai.japt.BlockingGroup
 import net.mamoe.mirai.japt.BlockingQQ
 import net.mamoe.mirai.message.data.Image
 import net.mamoe.mirai.network.BotNetworkHandler
-import net.mamoe.mirai.utils.MiraiExperimentalAPI
+import net.mamoe.mirai.utils.*
-import net.mamoe.mirai.utils.MiraiInternalAPI
-import net.mamoe.mirai.utils.MiraiLogger
-import net.mamoe.mirai.utils.toList
 import java.io.OutputStream
 import java.util.stream.Stream
 import kotlin.streams.asStream
 
+@UseExperimental(MiraiInternalAPI::class, MiraiExperimentalAPI::class)
 internal class BlockingBotImpl(private val bot: Bot) : BlockingBot {
     @MiraiInternalAPI
     override fun getAccount(): BotAccount = bot.account
@@ -51,7 +47,6 @@ internal class BlockingBotImpl(private val bot: Bot) : BlockingBot {
     override fun getFriend(id: Long): BlockingQQ = bot.getFriend(id).blocking()
     override fun queryGroupList(): Stream<Long> = runBlocking { bot.queryGroupList() }.asStream()
 
-    @UseExperimental(MiraiInternalAPI::class)
     override fun getGroupList(): List<BlockingGroup> = bot.groups.delegate.toList().map { it.blocking() }
 
     override fun queryGroupInfo(code: Long): GroupInfo = runBlocking { bot.queryGroupInfo(code) }
@@ -59,10 +54,8 @@ internal class BlockingBotImpl(private val bot: Bot) : BlockingBot {
     override fun getGroup(id: Long): BlockingGroup = runBlocking { bot.getGroup(id).blocking() }
     override fun getNetwork(): BotNetworkHandler = bot.network
     override fun login() = runBlocking { bot.login() }
-    override fun downloadAsByteArray(image: Image): ByteArray = bot.run { runBlocking { image.download().readBytes() } }
+    override fun downloadTo(image: Image, outputStream: OutputStream) = bot.run { runBlocking { openChannel(image).copyTo(outputStream) } }
-    override fun download(image: Image): ByteReadPacket = bot.run { runBlocking { image.download() } }
+    override fun downloadAndClose(image: Image, outputStream: OutputStream) = bot.run { runBlocking { openChannel(image).copyAndClose(outputStream) } }
-    override fun download(image: Image, outputStream: OutputStream) = bot.run { runBlocking { image.downloadTo(outputStream) } }
-
     override fun addFriend(id: Long, message: String?, remark: String?): AddFriendResult = runBlocking { bot.addFriend(id, message, remark) }
     override fun approveFriendAddRequest(id: Long, remark: String?) = runBlocking { bot.approveFriendAddRequest(id, remark) }
     override fun close(throwable: Throwable?) = bot.close(throwable)