implement toPipeline()

Author: thomasnield

README.md

@@ -135,3 +135,38 @@ val id = conn.select("SELECT * FROM USER WHERE ID = 1").blockingFirst { it.getIn
 
 println(id)
 ```
+
+## Allowing Choice of Sequence, Observable, or Flowable
+
+This library supports emitting `T` items built off a `ResultSet` in the form of:
+
+* `Sequence<T>`
+* `Observable<T>`
+* `Flowable<T>`
+* `Single<T>`
+* `Maybe<T>`
+
+If you are building an API, it may be handy to allow the user of the API to choose the means in which to receive the results.
+
+The `toPipeline()` function will allow mapping the `ResultSet` to `T` items, but defer to the API user how to receive the results.
+
+
+```kotlin
+
+data class User(val userName: String, val password: String)
+
+fun getUsers() = conn.select("SELECT * FROM USER")
+                .toPipeline {
+                    User(it.getString("USERNAME"), it.getString("PASSWORD"))
+                }
+
+fun main(args: Array<String>) {
+
+    getUsers().asFlowable().subscribe { println("Receiving $it via Flowable") }
+
+    getUsers().asObservable().subscribe { println("Receiving $it via Observable") }
+
+    getUsers().asSequence().forEach { println("Receiving $it via Sequence") }
+}
+
+```

build.gradle

@@ -46,7 +46,7 @@ signing {
 
 group = "org.nield"
 archivesBaseName = "rxkotlin-jdbc"
-version = "0.2.5"
+version = "0.2.6"
 
 uploadArchives {
     repositories {

src/main/kotlin/org/nield/rxkotlinjdbc/JdbcExtensions.kt

@@ -171,6 +171,8 @@ class SelectOperation(
 
     fun toCompletable() = toFlowable { Unit }.ignoreElements()
 
+    fun <T: Any> toPipeline(mapper: (ResultSet) -> T) = Pipeline(this, mapper = mapper)
+
     fun <T : Any> toSequence(mapper: (ResultSet) -> T): ResultSetSequence<T> {
         val cps = builder.toPreparedStatement()
         return ResultSetState({ cps.ps.executeQuery() }, cps.ps, cps.conn, autoClose).toSequence(mapper)
@@ -241,6 +243,8 @@ class InsertOperation(
 
     fun toCompletable() = toFlowable { Unit }.ignoreElements()
 
+    fun <T: Any> toPipeline(mapper: (ResultSet) -> T) = Pipeline(insertOperation = this, mapper = mapper)
+
     fun <T : Any> toSequence(mapper: (ResultSet) -> T): ResultSetSequence<T> {
         val cps = builder.toPreparedStatement()
         return ResultSetState({
@@ -390,6 +394,20 @@ class QueryIterator<out T>(val qs: ResultSetState,
     }
 }
 
+
+class Pipeline<T: Any>(val selectOperation: SelectOperation? = null,
+                       val insertOperation: InsertOperation? = null,
+                       val mapper: (ResultSet) -> T
+) {
+    fun toObservable(): Observable<T> = selectOperation?.toObservable(mapper) ?: insertOperation?.toObservable(mapper) ?: throw Exception("Operation must be provided")
+    fun toFlowable(): Flowable<T> = selectOperation?.toFlowable(mapper) ?: insertOperation?.toFlowable(mapper) ?: throw Exception("Operation must be provided")
+    fun toSingle(): Single<T> = selectOperation?.toSingle(mapper) ?: insertOperation?.toSingle(mapper) ?: throw Exception("Operation must be provided")
+    fun toMaybe(): Maybe<T> = selectOperation?.toMaybe(mapper) ?: insertOperation?.toMaybe(mapper) ?: throw Exception("Operation must be provided")
+    fun toSequence(): ResultSetSequence<T> = selectOperation?.toSequence(mapper) ?: insertOperation?.toSequence(mapper) ?: throw Exception("Operation must be provided")
+    fun blockingFirst() = selectOperation?.blockingFirst(mapper) ?: insertOperation?.blockingFirst(mapper) ?: throw Exception("Operation must be provided")
+    fun blockingFirstOrNull() = selectOperation?.blockingFirstOrNull(mapper) ?: insertOperation?.blockingFirstOrNull(mapper) ?: throw Exception("Operation must be provided")
+}
+
 fun PreparedStatement.processParameters(v: Array<out Any?>) = v.forEachIndexed { i,v2 -> processParameter(i,v2)}
 
 fun PreparedStatement.processParameter(pos: Int, argVal: Any?) {

src/test/kotlin/JdbcExtensionsTest.kt

@@ -4,6 +4,7 @@ import io.reactivex.observers.TestObserver
 import io.reactivex.subscribers.TestSubscriber
 import junit.framework.Assert.assertFalse
 import junit.framework.Assert.assertTrue
+import org.junit.Assert
 import org.junit.Test
 import org.nield.rxkotlinjdbc.*
 import java.sql.DriverManager
@@ -365,4 +366,42 @@ class DatabaseTest {
 
         testObserver.assertValues(mapOf("ID" to 1, "USERNAME" to "thomasnield", "PASSWORD" to "password123"))
     }
+
+    @Test
+    fun pipelineTest() {
+        val conn = connectionFactory()
+
+        data class User(val userName: String, val password: String)
+
+        val testObserver = TestObserver<User>()
+        val testSubscriber = TestSubscriber<User>()
+
+        val pipeline = conn.select("SELECT * FROM USER")
+                .toPipeline {
+                    User(it.getString("USERNAME"), it.getString("PASSWORD"))
+                }
+
+
+        pipeline.toObservable().subscribe(testObserver)
+
+        testObserver.assertValues(
+                User("thomasnield","password123"),
+                User("bobmarshal","batman43")
+        )
+
+
+        pipeline.toFlowable().subscribe(testSubscriber)
+
+        testSubscriber.assertValues(
+                User("thomasnield","password123"),
+                User("bobmarshal","batman43")
+        )
+
+        Assert.assertTrue(
+            pipeline.toSequence().toSet() == setOf(
+                    User("thomasnield","password123"),
+                    User("bobmarshal","batman43")
+            )
+        )
+    }
 }