package org.davidmoten.rx.jdbc;

import java.sql.Connection;
import java.sql.ResultSet;
import java.util.List;
import java.util.concurrent.Callable;

import io.reactivex.Emitter;
import io.reactivex.Flowable;
import io.reactivex.functions.BiConsumer;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

public class Select {

    public static <T> Flowable<T> create(Flowable<Connection> connections, List<Object> parameters, String sql,
            Function<? super ResultSet, T> mapper) {
        return connections //
                .firstOrError() //
                .toFlowable() //
                .flatMap(con -> create(con, sql, parameters, mapper));
    }

    private static <T> Flowable<? extends T> create(Connection con, String sql, List<Object> parameters,
            Function<? super ResultSet, T> mapper) {
        Callable<ResultSet> initialState = () -> Util //
                .setParameters(con.prepareStatement(sql), parameters) //
                .getResultSet();
        BiConsumer<ResultSet, Emitter<T>> generator = (rs, emitter) -> {
            if (rs.next()) {
                emitter.onNext(mapper.apply(rs));
            } else {
                emitter.onComplete();
            }
        };
        Consumer<ResultSet> disposeState = rs -> Util.closeAll(rs);
        return Flowable.generate(initialState, generator, disposeState);
    }

}

@@ -54,16 +54,51 @@
 ection;%0A
+import java.sql.PreparedStatement;%0A
 import j
@@ -116,16 +116,16 @@
 ultSet;%0A
-
 import j
@@ -458,16 +458,25 @@
 ctions, 
+Flowable%3C
 List%3CObj
@@ -471,32 +471,33 @@
 ble%3CList%3CObject%3E
+%3E
  parameters, Str
@@ -492,28 +492,16 @@
 ameters,
- String sql,
 %0A       
@@ -496,32 +496,44 @@
 ers,%0A           
+ String sql,
  Function%3C? supe
@@ -734,16 +734,592 @@
 %0A    %7D%0A%0A
+    private static %3CT%3E Flowable%3CT%3E create(Connection con, String sql, Flowable%3CList%3CObject%3E%3E parameterGroups,%0A            Function%3C? super ResultSet, T%3E mapper) %7B%0A        Callable%3CPreparedStatement%3E initialState = () -%3E con.prepareStatement(sql);%0A        Function%3CPreparedStatement, Flowable%3CT%3E%3E observableFactory = ps -%3E parameterGroups%0A                .flatMap(parameters -%3E create(con, ps, parameters, mapper), true, 1);%0A        Consumer%3CPreparedStatement%3E disposer = ps -%3E ps.close();%0A        return Flowable.using(initialState, observableFactory, disposer, true);%0A    %7D%0A%0A
     priv
@@ -1378,26 +1378,36 @@
 on con, 
-String sql
+PreparedStatement ps
 , List%3CO
@@ -1570,33 +1570,10 @@
 ers(
-con.prepareStatement(sql)
+ps
 , pa
@@ -1907,24 +1907,17 @@
  -%3E 
-Util
+rs
 .close
-All(rs
+(
 );%0A 

update select for stream of parameter lists