import io.vertx.core.Future;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.sql.ResultSet;
import io.vertx.rxjava.core.AbstractVerticle;
import io.vertx.rxjava.core.Vertx;
import io.vertx.rxjava.ext.jdbc.JDBCClient;
import rx.Single;

import java.util.List;

public class JDBC extends AbstractVerticle {

    private static final String query1 = "Select 1;";

    private static JDBCClient createJDBCClient() {
        return JDBCClient.createShared(Vertx.currentContext().owner(), new JsonObject()
                .put("url", "jdbc:mysql://127.0.0.1/mysql?autoReconnect=true&useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&failOverReadOnly=false&maxReconnects=10")
                .put("user", "root")
                .put("password", "root1234")
                .put("driver_class", "com.mysql.cj.jdbc.Driver")
                .put("max_pool_size", 5)
        );
    }

    @Override
    public void start(Future<Void> startFuture) throws Exception {
        JDBCClient jdbcClient = createJDBCClient();

        vertx.setPeriodic(1000, handler -> {
            Single<ResultSet> mainSingle = jdbcClient
                    .rxGetConnection()
                    .flatMap(sqlConnection ->
                        sqlConnection.rxQueryWithParams(query1, new JsonArray())
                                .doAfterTerminate(() -> {
                                    System.out.println("Closing the connection.");
                                    sqlConnection.close();
                                })
                    );

            Single<JsonArray> s1 = mainSingle
                    .flatMap(resultSet -> resultSet.getNumRows() < 2 ?
                            Single.error(new Exception("Got Error.")) :
                            Single.just(resultSet.getOutput()));

            Single<List<JsonObject>> s2 = mainSingle
                    .map(ResultSet::getRows);


            Single.zip(s1, s2, (rs1, rs2) -> new JsonObject().put("rs1", rs1).put("rs2", rs2))
                    .subscribe(System.out::println, Throwable::printStackTrace);
        });

        super.start(startFuture);
    }
}