#269

Author: mgramin

src/main/kotlin/com/github/mgramin/sqlboot/model/resourcetype/impl/sql/SqlResourceType.kt

@@ -38,7 +38,6 @@ 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.scheduler.Schedulers
 
 /**
  * Created by MGramin on 12.07.2017.
@@ -63,15 +62,22 @@ class SqlResourceType(
 
     override fun read(uri: Uri): Sequence<DbResource> {
         val mergeSequential: Flux<Map<String, Any>> =
-                Flux.mergeSequential(
+                Flux.merge(
                         connections
-                                .map { connection -> createQuery(uri, connection).execute(hashMapOf("uri" to uri)) }
-                                .map { it.parallel().runOn(Schedulers.parallel()) }
-                                .map { connection -> connection.map { it } })
+                                .asSequence()
+                                .map { connection ->
+                                    return@map createQuery(uri, connection).execute(hashMapOf("uri" to uri))
+                                            .map<Map<String, Any>?> {
+                                                val mutableMap = it.toMutableMap()
+                                                mutableMap["database"] = connection.name()
+                                                mutableMap
+                                            }
+                                }
+                                .toList()
+
+                )
 
         return mergeSequential
-                .parallel()
-                .runOn(Schedulers.parallel())
                 .map { o ->
                     val path = o.entries
                             .filter { v -> v.key.startsWith("@") }
@@ -80,10 +86,11 @@ class SqlResourceType(
                     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()
+                }
+                .collectList()
+                .block()
+                .asSequence()
     }
 
     override fun metaData(): Map<String, String> {

src/main/kotlin/com/github/mgramin/sqlboot/sql/select/SelectQuery.kt

@@ -40,13 +40,6 @@ interface SelectQuery {
      */
     fun query(): String
 
-    /**
-     *  Select columns
-     *
-     * @return Map of column names and column comments
-     */
-    fun columns(): Map<String, String>
-//    fun columns(): List<Column>
 
     /**
      * Execute select query with parameters
@@ -56,6 +49,15 @@ interface SelectQuery {
     fun execute(variables: Map<String, Any>): Flux<Map<String, Any>>
 
 
+    /**
+     *  Select columns
+     *
+     * @return Map of column names and column comments
+     */
+    fun columns(): Map<String, String>
+//    fun columns(): List<Column>
+
+
     class Column(private val name: String, private val comment: String, private val properties: Map<String, String>) {
 
         fun name(): String {

src/main/kotlin/com/github/mgramin/sqlboot/sql/select/wrappers/JdbcSelectQuery.kt

@@ -31,6 +31,7 @@ import org.springframework.jdbc.core.JdbcTemplate
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.core.publisher.toFlux
+import reactor.core.scheduler.Schedulers
 import javax.sql.DataSource
 
 /**
@@ -63,6 +64,7 @@ class JdbcSelectQuery(
         logger.info(sqlText)
 
         return Mono.fromCallable {
+            logger.info(Thread.currentThread().toString())
             val rowSet = JdbcTemplate(dataSource).queryForRowSet(sqlText)
             return@fromCallable object : Iterator<Map<String, Any>> {
                 override fun hasNext(): Boolean {
@@ -78,11 +80,10 @@ class JdbcSelectQuery(
                 }
             }
         }
+                .publishOn(Schedulers.parallel())
                 .map { it.toFlux() }
                 .toFlux()
                 .flatMap { it }
-
-
     }
 
 }