增加修改缓存事件个数的功能

app/src/main/java/io/github/guowenlong/app/GuideActivity.kt

@@ -27,6 +27,9 @@ class GuideActivity : AppCompatActivity() {
         setContentView(binding.root)
         binding.btnStart.setOnClickListener {
             startActivity(Intent(this, MainActivity::class.java))
+            binding.root.postDelayed({
+                CoroutineBus.unsubscribe(this,RandomNumEvent::class.java)
+            }, 5000)
         }
         CoroutineBus.subscribeByLifecycle(
             this,

app/src/main/java/io/github/guowenlong/app/LeftFragment.kt

@@ -11,6 +11,7 @@ import androidx.fragment.app.Fragment
 import io.github.guowenlong.app.databinding.FragmentLeftBinding
 import io.github.guowenlong.coroutinebus.CoroutineBus
 import kotlinx.coroutines.Dispatchers
+import kotlin.random.Random
 
 /**
  * Description: 左侧Fragment
@@ -37,25 +38,39 @@ class LeftFragment : Fragment() {
         super.onViewCreated(view, savedInstanceState)
 
         binding.btnSend.setOnClickListener {
-            CoroutineBus.post(RandomNumEvent(22))
+            CoroutineBus.post(RandomNumEvent(Random.nextInt(0, 100)))
         }
 
         binding.btnRegister.setOnClickListener {
-            CoroutineBus.subscribe(this, isSticky = false ,dispatcher = Dispatchers.Main) { event: RandomNumEvent ->
+            CoroutineBus.subscribe(
+                this,
+                isSticky = false,
+                replay = 100,
+                dispatcher = Dispatchers.Main
+            ) { event: RandomNumEvent ->
                 Log.e("Left", "LeftFragment")
-                binding.tvContent.text = "${binding.tvContent.text} \n ${event.number}"
+                binding.tvContent.text = "${binding.tvContent.text} \n 非粘性事件 收到 ${event.number}"
             }
         }
 
         binding.btnRegisterSticky.setOnClickListener {
             CoroutineBus.subscribe(
                 this,
                 isSticky = true,
+                replay = 100,
                 dispatcher = Dispatchers.Main
             ) { event: RandomNumEvent ->
-                Toast.makeText(requireContext(), "sticky: $event", Toast.LENGTH_SHORT).show()
-                binding.tvContent.text = "${binding.tvContent.text} \n ${event.number}"
+                binding.tvContent.text = "${binding.tvContent.text} \n 粘性事件 收到 : ${event.number}"
             }
         }
+        binding.btnUnregister.setOnClickListener {
+            CoroutineBus.unsubscribe(this, RandomNumEvent::class.java)
+        }
+
+
+    }
+
+    override fun onDestroy() {
+        super.onDestroy()
     }
 }

app/src/main/java/io/github/guowenlong/app/MainActivity.kt

@@ -1,7 +1,11 @@
 package io.github.guowenlong.app
 
 import android.os.Bundle
+import android.widget.Toast
 import androidx.appcompat.app.AppCompatActivity
+import io.github.guowenlong.coroutinebus.CoroutineBus
+import io.github.guowenlong.coroutinebus.subscribeByLifecycle
+import kotlinx.coroutines.Dispatchers
 
 /**
  * Description: 首页

app/src/main/java/io/github/guowenlong/app/RightFragment.kt

@@ -42,7 +42,7 @@ class RightFragment : Fragment() {
         //onStart的时候注册,onStop的时候取消注册
         //dispatcher = Dispatchers.Main 可以不设置 默认 dispatcher = Dispatchers.Default
         CoroutineBus.subscribeByLifecycle(
-            this, isSticky = true, lifecycleOwner = this
+            this, isSticky = false, lifecycleOwner = this
         ) { event: RandomNumEvent ->
             Toast.makeText(
                 requireContext(),

app/src/main/res/layout/fragment_left.xml

@@ -25,6 +25,13 @@
         android:text="register with sticky"
         app:layout_constraintTop_toBottomOf="@id/btn_register" />
 
+    <Button
+        android:id="@+id/btn_unregister"
+        android:layout_width="match_parent"
+        android:layout_height="wrap_content"
+        android:text="unregister"
+        app:layout_constraintTop_toBottomOf="@id/btn_register_sticky" />
+
     <TextView
         android:padding="12dp"
         android:layout_margin="12dp"
@@ -33,7 +40,7 @@
         app:layout_constraintBottom_toBottomOf="parent"
         android:textColor="@android:color/white"
         android:text="收到的内容"
-        app:layout_constraintTop_toBottomOf="@id/btn_register_sticky"
+        app:layout_constraintTop_toBottomOf="@id/btn_unregister"
         android:layout_width="match_parent"
         android:layout_height="0dp"/>
 </androidx.constraintlayout.widget.ConstraintLayout>

library/build/kotlin/compileDebugKotlin/cacheable/caches-jvm/jvm/kotlin/proto.tab.values.s

@@ -0,0 +1 @@
+��!

library/build/kotlin/compileDebugKotlin/cacheable/caches-jvm/lookups/counters.tab

@@ -1,2 +1,2 @@
-19
+25
 0

library/build/kotlin/compileDebugKotlin/cacheable/caches-jvm/lookups/lookups.tab.values.s

@@ -1 +1 @@
-�+�(�:�%�&�%�#
+�+�(�:�%�&�%�#�#�(�3�.

library/src/main/java/io/github/guowenlong/coroutinebus/CoroutineBus.kt

@@ -23,17 +23,19 @@ object CoroutineBus {
      * 订阅事件
      *
      * @param id 订阅者的唯一标识
-     * @param isSticky 是否接收粘性事件
+     * @param isSticky 是否接收粘性事件(注意:如果[replay]为0 那么粘性事件将无法使用)
+     * @param replay 缓存事件个数.默认缓存100个.如果为0,那么粘性事件将无法使用
      * @param dispatcher 指定协程的调度器
      * @param callback 回调函数
      */
     inline fun <reified T : Any> subscribe(
         id: Any,
         isSticky: Boolean = false,
+        replay: Int = 100,
         dispatcher: CoroutineDispatcher,
         noinline callback: suspend (event: T) -> Unit
     ) {
-        return subscribeTo(id, T::class.java, isSticky, dispatcher, callback)
+        return subscribeTo(id, T::class.java, isSticky, replay, dispatcher, callback)
     }
 
     /**
@@ -46,6 +48,30 @@ object CoroutineBus {
         subscribers[clazz]?.remove(id)?.cancel()
     }
 
+    /**
+     * 取消订阅该 [id] 的所有订阅
+     *
+     * @param id 订阅者的唯一标识
+     */
+    fun unsubscribe(id: Any) {
+        subscribers.forEach { (_, value) ->
+            value.remove(id)?.cancel()
+        }
+    }
+
+    /**
+     * 清理所有订阅者和该事件的数据源
+     *
+     * @param clazz 订阅的事件类型
+     */
+    fun <T : Any> cleanEvent(clazz: Class<T>) {
+        subscribers[clazz]?.forEach {
+            it.value.cancel()
+        }
+        subscribers[clazz]?.clear()
+        producers.remove(clazz)
+    }
+
     /**
      * 发送事件
      *
@@ -65,14 +91,16 @@ object CoroutineBus {
      *
      * @param id 订阅者的唯一标识
      * @param clazz 订阅的事件类型
-     * @param isSticky 是否接收粘性事件
+     * @param isSticky 是否接收粘性事件(注意:如果[replay]为0 那么粘性事件将无法使用)
+     * @param replay 缓存事件个数 默认缓存100个,如果为0 那么粘性事件将无法使用
      * @param dispatcher 指定协程的调度器
      * @param callback 回调函数
      */
     fun <T : Any> subscribeTo(
         id: Any,
         clazz: Class<T>,
         isSticky: Boolean,
+        replay: Int,
         dispatcher: CoroutineDispatcher,
         callback: suspend (event: T) -> Unit
     ) {
@@ -84,7 +112,7 @@ object CoroutineBus {
         }
 
         val job = CoroutineScope(Job() + Dispatchers.Default + exceptionHandler).launch {
-            getOrPutProducers(clazz, isSticky)
+            getOrPutProducers(clazz, replay)
                 .also {
                     if (it.replayCache.isEmpty()) {
                         it.filterNotNull()
@@ -94,10 +122,20 @@ object CoroutineBus {
                                 }
                             }
                     } else {
-                        it.drop(if (isSticky) 0 else 1).collect {
-                            withContext(dispatcher) {
-                                callback(it)
-                            }
+                        if (isSticky) {
+                            it.filterNotNull()
+                                .collect {
+                                    withContext(dispatcher) {
+                                        callback(it)
+                                    }
+                                }
+                        } else {
+                            it.drop(it.replayCache.size)
+                                .collect {
+                                    withContext(dispatcher) {
+                                        callback(it)
+                                    }
+                                }
                         }
                     }
                 }
@@ -110,16 +148,16 @@ object CoroutineBus {
      * 获取或创建Flow
      *
      * @param clazz 事件类型
-     * @param isSticky 是否接收粘性事件
+     * @param replay 缓存大小
      */
     @Suppress("UNCHECKED_CAST")
-    private fun <T : Any> getOrPutProducers(
+    fun <T : Any> getOrPutProducers(
         clazz: Class<T>,
-        isSticky: Boolean = true
+        replay: Int = 100
     ): MutableSharedFlow<T> =
-        producers.getOrPut(clazz) { MutableSharedFlow<T>(replay = if (isSticky) 1 else 0) } as MutableSharedFlow<T>
+        producers.getOrPut(clazz) { MutableSharedFlow<T>(replay) } as MutableSharedFlow<T>
 
-    private fun <T : Any> getOrPutSubscribers(
+    fun <T : Any> getOrPutSubscribers(
         id: Any? = null,
         clazz: Class<T>,
         job: Job

library/src/main/java/io/github/guowenlong/coroutinebus/KotlinExtensions.kt

@@ -3,8 +3,7 @@ package io.github.guowenlong.coroutinebus
 import androidx.lifecycle.Lifecycle
 import androidx.lifecycle.LifecycleEventObserver
 import androidx.lifecycle.LifecycleOwner
-import kotlinx.coroutines.CoroutineDispatcher
-import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.*
 
 /**
  * Description: CoroutineBus的扩展类
@@ -19,21 +18,24 @@ import kotlinx.coroutines.Dispatchers
  *
  * @param id 订阅者的唯一标识
  * @param lifecycleOwner LifecycleOwner
- * @param isSticky 是否接收粘性事件
+ * @param isSticky 是否接收粘性事件(注意:如果[replay]为0 那么粘性事件将无法使用)
+ * @param replay 缓存事件个数 默认缓存100个,如果为0 那么粘性事件将无法使用
  * @param dispatcher 指定协程的调度器
  * @param callback 回调函数
  */
 inline fun <reified T : Any> CoroutineBus.subscribeByLifecycle(
     id: Any,
     lifecycleOwner: LifecycleOwner,
     isSticky: Boolean = false,
+    replay: Int = 100,
     dispatcher: CoroutineDispatcher = Dispatchers.Main,
     noinline callback: suspend (event: T) -> Unit
 ) {
     return CoroutineBus.subscribeByLifecycleTo(
         id,
         T::class.java,
         isSticky,
+        replay,
         dispatcher,
         lifecycleOwner,
         callback
@@ -46,14 +48,16 @@ inline fun <reified T : Any> CoroutineBus.subscribeByLifecycle(
  * @param id 订阅者的唯一标识
  * @param clazz 订阅的事件类型
  * @param lifecycleOwner LifecycleOwner
- * @param isSticky 是否接收粘性事件
+ * @param isSticky 是否接收粘性事件(注意:如果[replay]为0 那么粘性事件将无法使用)
+ * @param replay 缓存事件个数 默认缓存100个,如果为0 那么粘性事件将无法使用
  * @param dispatcher 指定协程的调度器
  * @param callback 回调函数
  */
 fun <T : Any> CoroutineBus.subscribeByLifecycleTo(
     id: Any,
     clazz: Class<T>,
     isSticky: Boolean,
+    replay: Int,
     dispatcher: CoroutineDispatcher,
     lifecycleOwner: LifecycleOwner,
     callback: suspend (event: T) -> Unit
@@ -62,7 +66,7 @@ fun <T : Any> CoroutineBus.subscribeByLifecycleTo(
         override fun onStateChanged(source: LifecycleOwner, event: Lifecycle.Event) {
             when (event) {
                 Lifecycle.Event.ON_CREATE -> {
-                    subscribeTo(id, clazz, isSticky, dispatcher, callback)
+                    subscribeTo(id, clazz, isSticky, replay, dispatcher, callback)
                 }
                 Lifecycle.Event.ON_DESTROY -> {
                     unsubscribe(id, clazz)
@@ -72,4 +76,20 @@ fun <T : Any> CoroutineBus.subscribeByLifecycleTo(
             }
         }
     })
+}
+
+/**
+ * 发送事件
+ * 注意:如果在订阅前就已经先发送事件 那么创建[event]的数据源将缓存[replay]个
+ *
+ * @param event 事件
+ * @param replay 缓存事件的数量
+ * @param dispatcher 指定协程的调度器 默认在主线程 接受者接收的默认线程也是主线程
+ */
+fun <T : Any> post(event: T, replay: Int, dispatcher: CoroutineDispatcher = Dispatchers.Main) {
+    CoroutineScope(Job() + dispatcher).launch {
+        CoroutineBus.getOrPutProducers(clazz = event.javaClass, replay = replay).also {
+            it.emit(event)
+        }
+    }
 }