add remove a single subscriber

snail-kotlin/src/main/java/com/compass/snail/Fail.kt

@@ -5,7 +5,9 @@ package com.compass.snail
 import kotlinx.coroutines.CoroutineDispatcher
 
 open class Fail<T>(private val _error: Throwable) : Observable<T>() {
-    override fun subscribe(dispatcher: CoroutineDispatcher?, next: ((T) -> Unit)?, error: ((Throwable) -> Unit)?, done: (() -> Unit)?) {
-        notify(Subscriber(dispatcher, createHandler(next, error, done)), Event(error = _error))
+    override fun subscribe(dispatcher: CoroutineDispatcher?, next: ((T) -> Unit)?, error: ((Throwable) -> Unit)?, done: (() -> Unit)?): Subscriber<T> {
+        val subscriber = Subscriber(dispatcher, createHandler(next, error, done))
+        notify(subscriber, Event(error = _error))
+        return subscriber
     }
 }

snail-kotlin/src/main/java/com/compass/snail/IObservable.kt

@@ -5,12 +5,13 @@ package com.compass.snail
 import kotlinx.coroutines.CoroutineDispatcher
 
 interface IObservable<T> {
-    fun subscribe(dispatcher: CoroutineDispatcher? = null, next: ((T) -> Unit)? = null, error: ((Throwable) -> Unit)? = null, done: (() -> Unit)? = null)
+    fun subscribe(dispatcher: CoroutineDispatcher? = null, next: ((T) -> Unit)? = null, error: ((Throwable) -> Unit)? = null, done: (() -> Unit)? = null): Subscriber<T>
     fun on(dispatcher: CoroutineDispatcher): Observable<T>
     fun next(value: T)
     fun error(error: Throwable)
     fun done()
     fun removeSubscribers()
+    fun removeSubscriber(subscriber: Subscriber<T>)
     suspend fun block(): BlockResult<T>
     fun throttle(delayMs: Long): Observable<T>
     fun debounce(delayMs: Long): Observable<T>

snail-kotlin/src/main/java/com/compass/snail/Just.kt

@@ -5,8 +5,10 @@ package com.compass.snail
 import kotlinx.coroutines.CoroutineDispatcher
 
 open class Just<T>(private val value: T) : Observable<T>() {
-    override fun subscribe(dispatcher: CoroutineDispatcher?, next: ((T) -> Unit)?, error: ((Throwable) -> Unit)?, done: (() -> Unit)?) {
-        notify(Subscriber(dispatcher, createHandler(next, error, done)), Event(next = Next(value)))
-        notify(Subscriber(dispatcher, createHandler(next, error, done)), Event(done = true))
+    override fun subscribe(dispatcher: CoroutineDispatcher?, next: ((T) -> Unit)?, error: ((Throwable) -> Unit)?, done: (() -> Unit)?): Subscriber<T> {
+        val subscriber = Subscriber(dispatcher, createHandler(next, error, done))
+        notify(subscriber, Event(next = Next(value)))
+        notify(subscriber, Event(done = true))
+        return subscriber
     }
 }

snail-kotlin/src/main/java/com/compass/snail/Observable.kt

@@ -14,12 +14,14 @@ open class Observable<T> : IObservable<T> {
     private var stoppedEvent: Event<T>? = null
     private var subscribers: MutableList<Subscriber<T>> = mutableListOf()
 
-    override fun subscribe(dispatcher: CoroutineDispatcher?, next: ((T) -> Unit)?, error: ((Throwable) -> Unit)?, done: (() -> Unit)?) {
+    override fun subscribe(dispatcher: CoroutineDispatcher?, next: ((T) -> Unit)?, error: ((Throwable) -> Unit)?, done: (() -> Unit)?) : Subscriber<T> {
+        val subscriber = Subscriber(dispatcher, createHandler(next, error, done))
         stoppedEvent?.let {
-            notify(Subscriber(dispatcher, createHandler(next, error, done)), it)
-            return
+            notify(subscriber, it)
+            return subscriber
         }
-        subscribers.add(Subscriber(dispatcher, createHandler(next, error, done)))
+        subscribers.add(subscriber)
+        return subscriber
     }
 
     override fun on(dispatcher: CoroutineDispatcher): Observable<T> {
@@ -46,6 +48,10 @@ open class Observable<T> : IObservable<T> {
         subscribers.clear()
     }
 
+    override fun removeSubscriber(subscriber: Subscriber<T>) {
+        subscribers.remove(subscriber)
+    }
+
     private fun on(event: Event<T>) {
         if (stoppedEvent != null) return
 

snail-kotlin/src/main/java/com/compass/snail/Replay.kt

@@ -7,9 +7,9 @@ import kotlinx.coroutines.CoroutineDispatcher
 open class Replay<T>(private val threshold: Int) : Observable<T>() {
     private var values: MutableList<T> = mutableListOf()
 
-    override fun subscribe(dispatcher: CoroutineDispatcher?, next: ((T) -> Unit)?, error: ((Throwable) -> Unit)?, done: (() -> Unit)?) {
-        super.subscribe(dispatcher, next, error, done)
+    override fun subscribe(dispatcher: CoroutineDispatcher?, next: ((T) -> Unit)?, error: ((Throwable) -> Unit)?, done: (() -> Unit)?): Subscriber<T> {
         replay(dispatcher, createHandler(next, error, done))
+        return super.subscribe(dispatcher, next, error, done)
     }
 
     override fun next(value: T) {

snail-kotlin/src/test/java/com/compass/snail/ObservableTests.kt

@@ -85,6 +85,36 @@ class ObservableTests {
         assertEquals("2", strings?.get(1))
     }
 
+    @Test
+    fun testRemoveSubscriber() {
+        val listOfString = mutableListOf<String>()
+        val listOfInt =  mutableListOf<Int>()
+        val stringSubscriber = subject?.subscribe(next = { listOfString.add(it) })
+        val intSubscriber = subject?.subscribe(next = { listOfInt.add(it.toInt()) })
+        subject?.next("1")
+
+        assertEquals(1, listOfString.size)
+        assertEquals(1, listOfInt.size)
+        assertEquals("1", listOfString[0])
+        assertEquals(1, listOfInt[0])
+
+        intSubscriber?.let { subject?.removeSubscriber(it) }
+        subject?.next("2")
+        assertEquals(2, listOfString.size)
+        assertEquals(1, listOfInt.size)
+        assertEquals("1", listOfString[0])
+        assertEquals("2", listOfString[1])
+        assertEquals(1, listOfInt[0])
+
+        stringSubscriber?.let { subject?.removeSubscriber(it) }
+        subject?.next("3")
+        assertEquals(2, listOfString.size)
+        assertEquals(1, listOfInt.size)
+        assertEquals("1", listOfString[0])
+        assertEquals("2", listOfString[1])
+        assertEquals(1, listOfInt[0])
+    }
+
     @Test
     fun testMultipleSubscribers() {
         val more = mutableListOf<String>()