More operators for Observable<Single>, Flowable<Single>, etc.

gradle.properties

@@ -1,2 +1,2 @@
-version=2.0.0-RC2
+version=2.0.0
 org.gradle.jvmargs=-Xms256m -Xmx1024m -XX:MaxPermSize=256m

src/main/kotlin/io/reactivex/rxkotlin/completable.kt

@@ -1,6 +1,8 @@
 package io.reactivex.rxkotlin
 
 import io.reactivex.Completable
+import io.reactivex.Flowable
+import io.reactivex.Observable
 import io.reactivex.functions.Action
 import java.util.concurrent.Callable
 import java.util.concurrent.Future
@@ -9,3 +11,16 @@ fun Action.toCompletable(): Completable = Completable.fromAction(this)
 fun Callable<out Any>.toCompletable(): Completable = Completable.fromCallable(this)
 fun Future<out Any>.toCompletable(): Completable = Completable.fromFuture(this)
 fun (() -> Any).toCompletable(): Completable = Completable.fromCallable(this)
+
+
+// EXTENSION FUNCTION OPERATORS
+
+/**
+ * Merges the emissions of a Observable<Completable>. Same as calling `flatMapSingle { it }`.
+ */
+fun  Observable<Completable>.mergeAllCompletables() = flatMapCompletable { it }
+
+/**
+ * Merges the emissions of a Flowable<Completable>. Same as calling `flatMap { it }`.
+ */
+fun  Flowable<Completable>.mergeAllCompletables() = flatMapCompletable { it }

src/main/kotlin/io/reactivex/rxkotlin/flowable.kt

@@ -61,4 +61,26 @@ inline fun <reified R : Any> Flowable<*>.ofType(): Flowable<R> = ofType(R::class
 
 private fun <T : Any> Iterator<T>.toIterable() = object : Iterable<T> {
     override fun iterator(): Iterator<T> = this@toIterable
-}
+}
+
+//EXTENSION FUNCTION OPERATORS
+
+/**
+ * Merges the emissions of a Flowable<Flowable<T>>. Same as calling `flatMap { it }`.
+ */
+fun <T : Any> Flowable<Flowable<T>>.mergeAll() = flatMap { it }
+
+
+/**
+ * Concatenates the emissions of an Flowable<Flowable<T>>. Same as calling `concatMap { it }`.
+ */
+fun <T : Any> Flowable<Flowable<T>>.concatAll() = concatMap { it }
+
+
+/**
+ * Emits the latest `Flowable<T>` emitted through an `Flowable<Flowable<T>>`. Same as calling `switchMap { it }`.
+ */
+fun <T : Any> Flowable<Flowable<T>>.switchLatest() = switchMap { it }
+
+
+fun <T : Any> Flowable<Flowable<T>>.switchOnNext(): Flowable<T> = Flowable.switchOnNext(this)

src/main/kotlin/io/reactivex/rxkotlin/maybe.kt

@@ -1,6 +1,8 @@
 package io.reactivex.rxkotlin
 
+import io.reactivex.Flowable
 import io.reactivex.Maybe
+import io.reactivex.Observable
 import java.util.concurrent.Callable
 import java.util.concurrent.Future
 
@@ -10,4 +12,18 @@ fun <T : Any> Callable<T>.toMaybe(): Maybe<T> = Maybe.fromCallable(this)
 fun <T : Any> (() -> T).toMaybe(): Maybe<T> = Maybe.fromCallable(this)
 
 inline fun <reified R : Any> Maybe<Any>.cast(): Maybe<R> = cast(R::class.java)
-inline fun <reified R : Any> Maybe<Any>.ofType(): Maybe<R> = ofType(R::class.java)
+inline fun <reified R : Any> Maybe<Any>.ofType(): Maybe<R> = ofType(R::class.java)
+
+
+
+// EXTENSION FUNCTION OPERATORS
+
+/**
+ * Merges the emissions of a Observable<Maybe<T>>. Same as calling `flatMapMaybe { it }`.
+ */
+fun <T : Any> Observable<Maybe<T>>.mergeAllMaybes() = flatMapMaybe { it }
+
+/**
+ * Merges the emissions of a Flowable<Maybe<T>>. Same as calling `flatMap { it }`.
+ */
+fun <T : Any> Flowable<Maybe<T>>.mergeAllMaybes() = flatMapMaybe { it }

src/main/kotlin/io/reactivex/rxkotlin/observable.kt

@@ -61,4 +61,25 @@ inline fun <reified R : Any> Observable<*>.ofType(): Observable<R> = ofType(R::c
 
 private fun <T : Any> Iterator<T>.toIterable() = object : Iterable<T> {
     override fun iterator(): Iterator<T> = this@toIterable
-}
+}
+
+
+// EXTENSION FUNCTION OPERATORS
+
+/**
+ * Merges the emissions of an Observable<Observable<T>>. Same as calling `flatMap { it }`.
+ */
+fun <T : Any> Observable<Observable<T>>.mergeAll() = flatMap { it }
+
+/**
+ * Concatenates the emissions of an Observable<Observable<T>>. Same as calling `concatMap { it }`.
+ */
+fun <T : Any> Observable<Observable<T>>.concatAll() = concatMap { it }
+
+/**
+ * Emits the latest `Observable<T>` emitted through an `Observable<Observable<T>>`. Same as calling `switchMap { it }`.
+ */
+fun <T : Any> Observable<Observable<T>>.switchLatest() = switchMap { it }
+
+fun <T : Any> Observable<Observable<T>>.switchOnNext(): Observable<T> = Observable.switchOnNext(this)
+

src/main/kotlin/io/reactivex/rxkotlin/single.kt

@@ -1,5 +1,7 @@
 package io.reactivex.rxkotlin
 
+import io.reactivex.Flowable
+import io.reactivex.Observable
 import io.reactivex.Single
 import java.util.concurrent.Callable
 import java.util.concurrent.Future
@@ -10,3 +12,16 @@ fun <T : Any> Callable<T>.toSingle(): Single<T> = Single.fromCallable(this)
 fun <T : Any> (() -> T).toSingle(): Single<T> = Single.fromCallable(this)
 
 inline fun <reified R : Any> Single<Any>.cast(): Single<R> = cast(R::class.java)
+
+
+// EXTENSION FUNCTION OPERATORS
+
+/**
+ * Merges the emissions of a Observable<Single<T>>. Same as calling `flatMapSingle { it }`.
+ */
+fun <T : Any> Observable<Single<T>>.mergeAllSingles() = flatMapSingle { it }
+
+/**
+ * Merges the emissions of a Flowable<Single<T>>. Same as calling `flatMap { it }`.
+ */
+fun <T : Any> Flowable<Single<T>>.mergeAllSingles() = flatMapSingle { it }