#269

mgramin · mgramin · commit e2ebf25ae02f · 2019-02-26T17:29:24.000+03:00
diff --git a/src/main/kotlin/com/github/mgramin/sqlboot/model/resourcetype/impl/sql/SqlResourceType.kt b/src/main/kotlin/com/github/mgramin/sqlboot/model/resourcetype/impl/sql/SqlResourceType.kt
@@ -30,15 +30,14 @@ import com.github.mgramin.sqlboot.model.resource.impl.DbResourceImpl
 import com.github.mgramin.sqlboot.model.resourcetype.ResourceType
 import com.github.mgramin.sqlboot.model.uri.Uri
 import com.github.mgramin.sqlboot.model.uri.impl.DbUri
+import com.github.mgramin.sqlboot.sql.select.SelectQuery
 import com.github.mgramin.sqlboot.sql.select.impl.SimpleSelectQuery
 import com.github.mgramin.sqlboot.sql.select.wrappers.JdbcSelectQuery
 import com.github.mgramin.sqlboot.sql.select.wrappers.OrderedSelectQuery
 import com.github.mgramin.sqlboot.sql.select.wrappers.PaginatedSelectQuery
 import com.github.mgramin.sqlboot.template.generator.impl.GroovyTemplateGenerator
 import org.apache.commons.lang3.StringUtils.strip
 import reactor.core.publisher.Flux
-import reactor.core.publisher.Mono
-import reactor.core.publisher.toFlux
 import reactor.core.scheduler.Schedulers
 
 /**
@@ -63,35 +62,28 @@ class SqlResourceType(
     }
 
     override fun read(uri: Uri): Sequence<DbResource> {
-        val result: ArrayList<Mono<Sequence<DbResourceImpl>>> = arrayListOf()
-        for (connection in connections) {
+        val mergeSequential: Flux<Map<String, Any>> =
+                Flux.mergeSequential(
+                        connections
+                                .map { connection -> createQuery(uri, connection).execute(hashMapOf("uri" to uri)) }
+                                .map { it.parallel().runOn(Schedulers.parallel()) }
+                                .map { connection -> connection.map { it } })
 
-            val fromCallable = Mono.fromCallable {
-                val map: Sequence<DbResourceImpl> =
-                        createQuery(uri, connection)
-                                .execute(hashMapOf("uri" to uri))
-                                .map { o ->
-                                    val path = o.entries
-                                            .filter { v -> v.key.startsWith("@") }
-                                            .map { it.value.toString() }
-                                    val name = path[path.size - 1]
-                                    val headers = o.entries
-                                            .map { strip(it.key, "@") to it.value }
-                                            .toMap()
-                                    val newHeaders = headers.toMutableMap()
-                                    newHeaders["database"] = connection.name()
-                                    DbResourceImpl(name, this, DbUri(this.name(), path), newHeaders)
-                                }
-                return@fromCallable map
-            }.publishOn(Schedulers.parallel())
-
-            result.add(fromCallable)
-        }
-
-        val mergeSequential: Flux<Sequence<DbResourceImpl>> = Flux.mergeSequential(result)
-        val map = mergeSequential.map { it.toFlux() }.flatMap { it }.collectList().block()
-
-        return map.asSequence()
+        return mergeSequential
+                .parallel()
+                .runOn(Schedulers.parallel())
+                .map { o ->
+                    val path = o.entries
+                            .filter { v -> v.key.startsWith("@") }
+                            .map { it.value.toString() }
+                    val name = path[path.size - 1]
+                    val headers = o.entries
+                            .map { strip(it.key, "@") to it.value }
+                            .toMap()
+//                    val newHeaders = headers.toMutableMap()
+//                    newHeaders["database"] = connection.name()
+                    DbResourceImpl(name, this, DbUri(this.name(), path), headers) as DbResource
+                }.sequential().collectList().block().asSequence()
     }
 
     override fun metaData(): Map<String, String> {
@@ -101,7 +93,7 @@ class SqlResourceType(
         return newColumns
     }
 
-    private fun createQuery(uri: Uri, connection: DbConnection): JdbcSelectQuery {
+    private fun createQuery(uri: Uri, connection: DbConnection): SelectQuery {
         return JdbcSelectQuery(
                 PaginatedSelectQuery(
                         OrderedSelectQuery(
diff --git a/src/main/kotlin/com/github/mgramin/sqlboot/sql/select/SelectQuery.kt b/src/main/kotlin/com/github/mgramin/sqlboot/sql/select/SelectQuery.kt
@@ -24,6 +24,8 @@
 
 package com.github.mgramin.sqlboot.sql.select
 
+import reactor.core.publisher.Flux
+
 /**
  * Simple select SQL-query
  *
@@ -51,7 +53,7 @@ interface SelectQuery {
      *
      * @return query result
      */
-    fun execute(variables: Map<String, Any>): Sequence<Map<String, Any>>
+    fun execute(variables: Map<String, Any>): Flux<Map<String, Any>>
 
 
     class Column(private val name: String, private val comment: String, private val properties: Map<String, String>) {
diff --git a/src/main/kotlin/com/github/mgramin/sqlboot/sql/select/impl/FakeSelectQuery.kt b/src/main/kotlin/com/github/mgramin/sqlboot/sql/select/impl/FakeSelectQuery.kt
@@ -25,6 +25,7 @@
 package com.github.mgramin.sqlboot.sql.select.impl
 
 import com.github.mgramin.sqlboot.sql.select.SelectQuery
+import reactor.core.publisher.Flux
 
 class FakeSelectQuery : SelectQuery {
 
@@ -40,7 +41,7 @@ class FakeSelectQuery : SelectQuery {
         return mapOf("n" to "First name", "mail" to "Personal email")
     }
 
-    override fun execute(variables: Map<String, Any>): Sequence<Map<String, Any>> {
+    override fun execute(variables: Map<String, Any>): Flux<Map<String, Any>> {
         TODO("not implemented")
     }
 
diff --git a/src/main/kotlin/com/github/mgramin/sqlboot/sql/select/impl/SimpleSelectQuery.kt b/src/main/kotlin/com/github/mgramin/sqlboot/sql/select/impl/SimpleSelectQuery.kt
@@ -31,6 +31,7 @@ import com.github.mgramin.sqlboot.sql.select.impl.parser.SELECTParser
 import com.github.mgramin.sqlboot.template.generator.TemplateGenerator
 import org.antlr.v4.runtime.ANTLRInputStream
 import org.antlr.v4.runtime.CommonTokenStream
+import reactor.core.publisher.Flux
 
 class SimpleSelectQuery(private val templateGenerator: TemplateGenerator) : SelectQuery {
 
@@ -45,7 +46,7 @@ class SimpleSelectQuery(private val templateGenerator: TemplateGenerator) : Sele
                 .toMap()
     }
 
-    override fun execute(variables: Map<String, Any>): Sequence<Map<String, Any>> {
+    override fun execute(variables: Map<String, Any>): Flux<Map<String, Any>> {
         throw RuntimeException("Not allow here")
     }
 
diff --git a/src/main/kotlin/com/github/mgramin/sqlboot/sql/select/wrappers/JdbcSelectQuery.kt b/src/main/kotlin/com/github/mgramin/sqlboot/sql/select/wrappers/JdbcSelectQuery.kt
@@ -28,6 +28,9 @@ import com.github.mgramin.sqlboot.sql.select.SelectQuery
 import com.github.mgramin.sqlboot.template.generator.impl.GroovyTemplateGenerator
 import org.slf4j.LoggerFactory
 import org.springframework.jdbc.core.JdbcTemplate
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import reactor.core.publisher.toFlux
 import javax.sql.DataSource
 
 /**
@@ -39,7 +42,6 @@ import javax.sql.DataSource
  */
 class JdbcSelectQuery(
         private val origin: SelectQuery,
-        @Deprecated("")
         private val dataSource: DataSource,
         private val nullAlias: String
 ) : SelectQuery {
@@ -56,23 +58,31 @@ class JdbcSelectQuery(
         return origin.columns()
     }
 
-    override fun execute(variables: Map<String, Any>): Sequence<Map<String, Any>> {
+    override fun execute(variables: Map<String, Any>): Flux<Map<String, Any>> {
         val sqlText = GroovyTemplateGenerator(origin.query()).generate(variables)
         logger.info(sqlText)
-        val rowSet = JdbcTemplate(dataSource).queryForRowSet(sqlText)
-        return object : Iterator<Map<String, Any>> {
-            override fun hasNext(): Boolean {
-                return rowSet.next()
-            }
 
-            override fun next(): Map<String, Any> {
-                return rowSet
-                        .metaData
-                        .columnNames
-                        .map { it.toLowerCase() to (rowSet.getObject(it) ?: nullAlias) }
-                        .toMap()
+        return Mono.fromCallable {
+            val rowSet = JdbcTemplate(dataSource).queryForRowSet(sqlText)
+            return@fromCallable object : Iterator<Map<String, Any>> {
+                override fun hasNext(): Boolean {
+                    return rowSet.next()
+                }
+
+                override fun next(): Map<String, Any> {
+                    return rowSet
+                            .metaData
+                            .columnNames
+                            .map { it.toLowerCase() to (rowSet.getObject(it) ?: nullAlias) }
+                            .toMap()
+                }
             }
-        }.asSequence()
+        }
+                .map { it.toFlux() }
+                .toFlux()
+                .flatMap { it }
+
+
     }
 
 }
diff --git a/src/test/kotlin/com/github/mgramin/sqlboot/sql/select/wrappers/JdbcSelectQueryTest.kt b/src/test/kotlin/com/github/mgramin/sqlboot/sql/select/wrappers/JdbcSelectQueryTest.kt
@@ -47,7 +47,7 @@ class JdbcSelectQueryTest {
 
     @Test
     fun execute() {
-        val rows = JdbcSelectQuery(FakeSelectQuery(), this.dataSource!!).execute(hashMapOf()).toList()
+        val rows = JdbcSelectQuery(FakeSelectQuery(), this.dataSource!!).execute(hashMapOf()).collectList().block()
         assertEquals(arrayListOf(linkedMapOf("n" to "mkyong", "mail" to "mkyong@gmail.com"),
                 linkedMapOf("n" to "alex", "mail" to "alex@yahoo.com"),
                 linkedMapOf("n" to "joel", "mail" to "joel@gmail.com")),
diff --git a/src/test/kotlin/com/github/mgramin/sqlboot/sql/select/wrappers/OrderedSelectQueryTest.kt b/src/test/kotlin/com/github/mgramin/sqlboot/sql/select/wrappers/OrderedSelectQueryTest.kt
@@ -60,7 +60,7 @@ internal class OrderedSelectQueryTest {
                                 FakeSelectQuery(),
                                 mapOf("n" to "asc", "mail" to "desc")),
                         this.dataSource!!)
-        val execute: Sequence<Map<String, Any>> = selectQuery.execute(emptyMap())
+        val execute: Sequence<Map<String, Any>> = selectQuery.execute(emptyMap()).collectList().block().asSequence()
         println("result = ${execute.toList()}")
     }
 

Original file line number	Diff line number	Diff line change
`@@ -24,6 +24,8 @@`
`24`	`24`
`25`	`25`	`package com.github.mgramin.sqlboot.sql.select`
`26`	`26`
	`27`	`+import reactor.core.publisher.Flux`
	`28`	`+`
`27`	`29`	`/**`
`28`	`30`	`* Simple select SQL-query`
`29`	`31`	`*`
`@@ -51,7 +53,7 @@ interface SelectQuery {`
`51`	`53`	`*`
`52`	`54`	`* @return query result`
`53`	`55`	`*/`
`54`		`- fun execute(variables: Map<String, Any>): Sequence<Map<String, Any>>`
	`56`	`+ fun execute(variables: Map<String, Any>): Flux<Map<String, Any>>`
`55`	`57`
`56`	`58`
`57`	`59`	`class Column(private val name: String, private val comment: String, private val properties: Map<String, String>) {`
Original file line number	Diff line number	Diff line change
`@@ -25,6 +25,7 @@`
`25`	`25`	`package com.github.mgramin.sqlboot.sql.select.impl`
`26`	`26`
`27`	`27`	`import com.github.mgramin.sqlboot.sql.select.SelectQuery`
	`28`	`+import reactor.core.publisher.Flux`
`28`	`29`
`29`	`30`	`class FakeSelectQuery : SelectQuery {`
`30`	`31`
`@@ -40,7 +41,7 @@ class FakeSelectQuery : SelectQuery {`
`40`	`41`	`return mapOf("n" to "First name", "mail" to "Personal email")`
`41`	`42`	`}`
`42`	`43`
`43`		`- override fun execute(variables: Map<String, Any>): Sequence<Map<String, Any>> {`
	`44`	`+ override fun execute(variables: Map<String, Any>): Flux<Map<String, Any>> {`
`44`	`45`	`TODO("not implemented")`
`45`	`46`	`}`
`46`	`47`
Original file line number	Diff line number	Diff line change
`@@ -31,6 +31,7 @@ import com.github.mgramin.sqlboot.sql.select.impl.parser.SELECTParser`
`31`	`31`	`import com.github.mgramin.sqlboot.template.generator.TemplateGenerator`
`32`	`32`	`import org.antlr.v4.runtime.ANTLRInputStream`
`33`	`33`	`import org.antlr.v4.runtime.CommonTokenStream`
	`34`	`+import reactor.core.publisher.Flux`
`34`	`35`
`35`	`36`	`class SimpleSelectQuery(private val templateGenerator: TemplateGenerator) : SelectQuery {`
`36`	`37`
`@@ -45,7 +46,7 @@ class SimpleSelectQuery(private val templateGenerator: TemplateGenerator) : Sele`
`45`	`46`	`.toMap()`
`46`	`47`	`}`
`47`	`48`
`48`		`- override fun execute(variables: Map<String, Any>): Sequence<Map<String, Any>> {`
	`49`	`+ override fun execute(variables: Map<String, Any>): Flux<Map<String, Any>> {`
`49`	`50`	`throw RuntimeException("Not allow here")`
`50`	`51`	`}`
`51`	`52`
Original file line number	Diff line number	Diff line change
`@@ -60,7 +60,7 @@ internal class OrderedSelectQueryTest {`
`60`	`60`	`FakeSelectQuery(),`
`61`	`61`	`mapOf("n" to "asc", "mail" to "desc")),`
`62`	`62`	`this.dataSource!!)`
`63`		`- val execute: Sequence<Map<String, Any>> = selectQuery.execute(emptyMap())`
	`63`	`+ val execute: Sequence<Map<String, Any>> = selectQuery.execute(emptyMap()).collectList().block().asSequence()`
`64`	`64`	`println("result = ${execute.toList()}")`
`65`	`65`	`}`
`66`	`66`