Merge pull request #31 from UrbanCompass/skip-first

add skip first

Author: TaniaGoswami

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

@@ -14,4 +14,5 @@ interface IObservable<T> {
     suspend fun block(): BlockResult<T>
     fun throttle(delayMs: Long): Observable<T>
     fun debounce(delayMs: Long): Observable<T>
+    fun skip(first: UInt): Observable<T>
 }

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

@@ -7,6 +7,7 @@ import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.launch
 import java.util.concurrent.Semaphore
+import kotlin.math.max
 
 open class Observable<T> : IObservable<T> {
     private var isStopped = 0
@@ -145,4 +146,21 @@ open class Observable<T> : IObservable<T> {
         }, error = { observable.error(it) }, done = { observable.done() })
         return observable
     }
+
+    override fun skip(first: UInt): Observable<T> {
+        val observable = Observable<T>()
+        var count = first.toInt()
+        subscribe(next = {
+            if (count == 0) {
+                observable.next(it)
+            }
+            count = max(count - 1, 0)
+        }, error = {
+            observable.error(it)
+        }, done = {
+            observable.done()
+        })
+
+        return observable
+    }
 }

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

@@ -190,4 +190,21 @@ class ObservableTests {
         assertEquals(received.count(), 1)
         assertEquals(received.first(), "2")
     }
+
+    @Test
+    fun testSkipFirst() {
+        val observable = Observable<String>()
+        var received = mutableListOf<String>()
+
+        observable.skip(2.toUInt()).subscribe(next = {
+            received.add(it)
+
+            assertEquals(received.count(), 1)
+            assertEquals(received.first(), "3")
+        })
+
+        observable.next("1")
+        observable.next("2")
+        observable.next("3")
+    }
 }