code: private void connectAndPrepareStatement(Subscriber<? super T> subscriber, State state)
              throws SQLException {
          log.debug("connectionProvider={}", query.context().connectionProvider());
          if (!subscriber.isUnsubscribed()) {
              log.debug("getting connection");
              state.con = query.context().connectionProvider().get();
              log.debug("preparing statement,sql={}", query.sql());
              state.ps = state.con.prepareStatement(query.sql(), ResultSet.TYPE_FORWARD_ONLY,
                      ResultSet.CONCUR_READ_ONLY);
              if (query.context().fetchSize() != null) {
                  state.ps.setFetchSize(query.context().fetchSize());
              }
              log.debug("setting parameters");
              Util.setParameters(state.ps, parameters, query.names());
          }
      }
docs: Obtains connection, creates prepared statement and assigns parameters to
  the prepared statement.
  
  @param subscriber
  @param state
  
  @throws SQLException