<commit_before>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);
    }
<commit_msg>update select for stream of parameter lists<commit_after>package org.davidmoten.rx.jdbc;

import java.sql.Connection;
import java.sql.PreparedStatement;
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, Flowable<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<T> create(Connection con, String sql, Flowable<List<Object>> parameterGroups,
            Function<? super ResultSet, T> mapper) {
        Callable<PreparedStatement> initialState = () -> con.prepareStatement(sql);
        Function<PreparedStatement, Flowable<T>> observableFactory = ps -> parameterGroups
                .flatMap(parameters -> create(con, ps, parameters, mapper), true, 1);
        Consumer<PreparedStatement> disposer = ps -> ps.close();
        return Flowable.using(initialState, observableFactory, disposer, true);
    }

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

{'fraction_non_alphanumeric': 0.11736011477761836, 'fraction_numerical': 0.00028694404591104734, 'mean_word_length': 2.6927966101694913, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 51, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1684609', 'n_tokens_mistral': 981, 'n_tokens_neox': 932, 'n_words': 223}