Add map and merge operators (WIP)

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

@@ -17,4 +17,6 @@ interface IObservable<T> {
     fun throttle(delayMs: Long): Observable<T>
     fun debounce(delayMs: Long): Observable<T>
     fun skip(first: Int): Observable<T>
+    fun <U> map(mappingFunction: (T) -> U): Observable<U>
+    fun merge(observables: List<Observable<T>>): Observable<T>
 }

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

@@ -170,4 +170,43 @@ open class Observable<T> : IObservable<T> {
 
         return observable
     }
+
+    override fun <U> map(mappingFunction: (T) -> U): Observable<U> {
+        val observable = Observable<U>()
+
+        this.subscribe(next = { unmapped ->
+            try {
+                val mapped: U = mappingFunction(unmapped)
+                observable.next(mapped)
+            } catch (e: Throwable) {
+                observable.error(e)
+            }
+        }, error = {
+            observable.error(it)
+        }, done = {
+            observable.done()
+        })
+
+        return observable
+    }
+
+    override fun merge(observables: List<Observable<T>>): Observable<T> {
+        val latest = Observable<T>()
+
+        observables.forEach {
+            it.forward(latest)
+        }
+
+        return latest
+    }
+
+    private fun forward(observable: Observable<T>) {
+        subscribe(next = {
+            observable.on(Event(next = Next(it)))
+        }, error = {
+            observable.on(Event(error = it))
+        }, done = {
+            observable.on(Event(done = true))
+        })
+    }
 }