Add extension fuctions to concat Iterables (#142)

* Add extension fuctions to concat Iterable<Single>, Iterable<Maybe> and Iterable<Completable>

* Add extension fuctions to concat Iterable<Flowable>, update descriptions

* Rename concat* extension functions to concatAll

* Add Iterable<Observable>.concatAll(), rename tests

Author: bmaslakov

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

@@ -1,6 +1,7 @@
 package io.reactivex.rxkotlin
 
 import io.reactivex.Completable
+import io.reactivex.CompletableSource
 import io.reactivex.Flowable
 import io.reactivex.Observable
 import io.reactivex.functions.Action
@@ -24,3 +25,8 @@ fun  Observable<Completable>.mergeAllCompletables() = flatMapCompletable { it }
  * Merges the emissions of a Flowable<Completable>. Same as calling `flatMap { it }`.
  */
 fun  Flowable<Completable>.mergeAllCompletables() = flatMapCompletable { it }
+
+/**
+ * Concats an Iterable of completables into flowable. Same as calling `Completable.concat(this)`
+ */
+fun Iterable<CompletableSource>.concatAll() = Completable.concat(this)

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

@@ -3,6 +3,7 @@ package io.reactivex.rxkotlin
 import io.reactivex.Flowable
 import io.reactivex.functions.BiFunction
 import io.reactivex.functions.Function3
+import org.reactivestreams.Publisher
 
 
 fun BooleanArray.toFlowable(): Flowable<Boolean> = asIterable().toFlowable()
@@ -108,4 +109,9 @@ fun <A: Any, B: Any> Flowable<Pair<A, B>>.toMap() = toMap({it.first},{it.second}
 /**
  * Collects `Pair` emission into a multimap
  */
-fun <A: Any, B: Any> Flowable<Pair<A, B>>.toMultimap() = toMultimap({it.first},{it.second})
+fun <A: Any, B: Any> Flowable<Pair<A, B>>.toMultimap() = toMultimap({it.first},{it.second})
+
+/**
+ * Concats an Iterable of flowables into flowable. Same as calling `Flowable.concat(this)`
+ */
+fun <T : Any> Iterable<Publisher<T>>.concatAll() = Flowable.concat(this)

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

@@ -2,6 +2,7 @@ package io.reactivex.rxkotlin
 
 import io.reactivex.Flowable
 import io.reactivex.Maybe
+import io.reactivex.MaybeSource
 import io.reactivex.Observable
 import java.util.concurrent.Callable
 import java.util.concurrent.Future
@@ -27,3 +28,8 @@ 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 }
+
+/**
+ * Concats an Iterable of maybes into flowable. Same as calling `Maybe.concat(this)`
+ */
+fun <T : Any> Iterable<MaybeSource<T>>.concatAll() = Maybe.concat(this)

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

@@ -1,6 +1,7 @@
 package io.reactivex.rxkotlin
 
 import io.reactivex.Observable
+import io.reactivex.ObservableSource
 import io.reactivex.functions.BiFunction
 import io.reactivex.functions.Function3
 
@@ -101,4 +102,6 @@ fun <A: Any, B: Any> Observable<Pair<A,B>>.toMap() = toMap({it.first},{it.second
 /**
  * Collects `Pair` emission into a multimap
  */
-fun <A: Any, B: Any> Observable<Pair<A,B>>.toMultimap() = toMultimap({it.first},{it.second})
+fun <A: Any, B: Any> Observable<Pair<A,B>>.toMultimap() = toMultimap({it.first},{it.second})
+
+fun  <T : Any> Iterable<ObservableSource<T>>.concatAll() = Observable.concat(this)

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

@@ -3,6 +3,7 @@ package io.reactivex.rxkotlin
 import io.reactivex.Flowable
 import io.reactivex.Observable
 import io.reactivex.Single
+import io.reactivex.SingleSource
 import java.util.concurrent.Callable
 import java.util.concurrent.Future
 
@@ -25,3 +26,8 @@ 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 }
+
+/**
+ * Concats an Iterable of singles into flowable. Same as calling `Single.concat(this)`
+ */
+fun <T : Any> Iterable<SingleSource<T>>.concatAll() = Single.concat(this)

src/test/kotlin/io/reactivex/rxkotlin/CompletableTest.kt

@@ -1,7 +1,9 @@
 package io.reactivex.rxkotlin
 
+import io.reactivex.Completable
 import io.reactivex.Single
 import io.reactivex.functions.Action
+import org.junit.Assert
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertNotNull
 import org.junit.Test
@@ -40,4 +42,14 @@ class CompletableTest {
         c1.toObservable<String>().blockingFirst()
     }
 
+    @Test fun testConcatAll() {
+        var list = emptyList<Int>()
+        (0 until 10)
+                .map { v -> Completable.create { list += v } }
+                .concatAll()
+                .subscribe {
+                    Assert.assertEquals((0 until 10).toList(), list)
+                }
+    }
+
 }

src/test/kotlin/io/reactivex/rxkotlin/FlowableTest.kt

@@ -199,4 +199,14 @@ class FlowableTest {
 
         testSubscriber.assertValues(Triple("Alpha",1, 100), Triple("Beta",2, 200), Triple("Gamma",3, 300))
     }
+
+    @Test fun testConcatAll() {
+        (0 until 10)
+                .map { Flowable.just(it) }
+                .concatAll()
+                .toList()
+                .subscribe { result ->
+                    Assert.assertEquals((0 until 10).toList(), result)
+                }
+    }
 }

src/test/kotlin/io/reactivex/rxkotlin/MaybeTest.kt

@@ -16,4 +16,14 @@ class MaybeTest {
                 }
         Assert.assertTrue(first.get() == "Alpha")
     }
+
+    @Test fun testConcatAll() {
+        (0 until 10)
+                .map { Maybe.just(it) }
+                .concatAll()
+                .toList()
+                .subscribe { result ->
+                    Assert.assertEquals((0 until 10).toList(), result)
+                }
+    }
 }

src/test/kotlin/io/reactivex/rxkotlin/ObservableTest.kt

@@ -2,6 +2,7 @@ package io.reactivex.rxkotlin
 
 import io.reactivex.Observable
 import io.reactivex.observers.TestObserver
+import org.junit.Assert
 import org.junit.Assert.*
 import org.junit.Ignore
 import org.junit.Test
@@ -223,4 +224,14 @@ class ObservableTest {
         testObserver.assertValues(Triple("Alpha",1, 100), Triple("Beta",2, 200), Triple("Gamma",3, 300))
     }
 
+    @Test fun testConcatAll() {
+        var counter = 0
+        (0 until 10)
+                .map { Observable.just(counter++, counter++, counter++) }
+                .concatAll()
+                .toList()
+                .subscribe { result ->
+                    Assert.assertEquals((0 until 30).toList(), result)
+                }
+    }
 }

src/test/kotlin/io/reactivex/rxkotlin/SingleTest.kt

@@ -2,6 +2,7 @@ package io.reactivex.rxkotlin
 
 import io.reactivex.Observable
 import io.reactivex.Single
+import org.junit.Assert
 import org.junit.Test
 import org.mockito.Mockito
 import org.mockito.Mockito.mock
@@ -56,4 +57,14 @@ class SingleTest : KotlinTests() {
         verify(a, Mockito.times(1))
                 .received("Alpha")
     }
+
+    @Test fun testConcatAll() {
+        (0 until 10)
+                .map { Single.just(it) }
+                .concatAll()
+                .toList()
+                .subscribe { result ->
+                    Assert.assertEquals((0 until 10).toList(), result)
+                }
+    }
 }