Java 类org.jooq.Query 实例源码

@Override
public <P> void fetchOne(Query query, Function<JsonObject, P> mapper, Handler<AsyncResult<P>> resultHandler) {
    getConnection().setHandler(sqlConnectionResult->{
        if(sqlConnectionResult.succeeded()){
            log("Fetch one", ()-> query.getSQL(ParamType.INLINED));
            sqlConnectionResult.result().queryWithParams(
                    query.getSQL(),
                    getBindValues(query),
                    executeAndClose(rs -> {
                                if(rs.getRows().size() > 1){
                                    throw new TooManyRowsException(String.format("Got more than one row: %d",rs.getRows().size()));
                                }
                                Optional<P> optional = rs.getRows().stream().findFirst().map(mapper);
                                return (optional.orElseGet(() -> null));
                            },
                            sqlConnectionResult.result(),
                            resultHandler)
            );
        }else{
            resultHandler.handle(Future.failedFuture(sqlConnectionResult.cause()));
        }
    });
}

@Override
public void execute(Query query, Handler<AsyncResult<Integer>> resultHandler) {
    getConnection().setHandler(sqlConnectionResult->{
        if(sqlConnectionResult.succeeded()){
            log("Execute", ()-> query.getSQL(ParamType.INLINED));
            sqlConnectionResult.result().updateWithParams(
                    query.getSQL(),
                    getBindValues(query),
                    executeAndClose(UpdateResult::getUpdated,
                            sqlConnectionResult.result(),
                            resultHandler)
            );
        }else{
            resultHandler.handle(Future.failedFuture(sqlConnectionResult.cause()));
        }
    });
}

@Override
public void insertReturning(Query query, Handler<AsyncResult<Long>> resultHandler) {
    getConnection().setHandler(sqlConnectionResult->{
        if(sqlConnectionResult.succeeded()){
            log("Insert Returning", ()-> query.getSQL(ParamType.INLINED));
            sqlConnectionResult.result().update(
                    query.getSQL(ParamType.INLINED),
                    executeAndClose(res -> res.getKeys().getLong(0),
                            sqlConnectionResult.result(),
                            resultHandler)
            );
        }else{
            resultHandler.handle(Future.failedFuture(sqlConnectionResult.cause()));
        }
    });
}

private Execute(
  Connection connection,
  QueryBuilder<? extends Query> queryBuilder
) {
  super(
    subscriber -> {
      try (Query query = queryBuilder.build(connection)) {
        setupUnsubscription(subscriber, query);
        log.debug("Execute {}", query);
        int i = query.execute();
        if (!subscriber.isUnsubscribed()) {
          subscriber.onNext(i);
          subscriber.onCompleted();
        }
      } catch (Throwable t) {
        handleException(t, subscriber);
      }
    }
  );
}

public boolean updateRoles(String userName, List<Role> newRoles) {
    try {
        dsl.transaction(config -> {
            LOG.info("Removing existing roles for: " + userName);
            DSL.using(config)
                    .delete(USER_ROLE)
                    .where(USER_ROLE.USER_NAME.eq(userName))
                    .execute();

            LOG.info("Inserting roles for " + userName + " / " + newRoles) ;
            DSLContext batcher = DSL.using(config);
            List<Query> inserts = map(newRoles, r -> batcher
                    .insertInto(USER_ROLE, USER_ROLE.USER_NAME, USER_ROLE.ROLE)
                    .values(userName, r.name()));

            batcher.batch(inserts)
                    .execute();
        });
        return true;
    } catch (Exception e) {
        return false;
    }

}

public int[] write(List<ChangeLog> changeLogs) {
    checkNotNull(changeLogs, "changeLogs must not be null");

    Query[] queries = changeLogs
            .stream()
            .map(changeLog -> DSL.insertInto(CHANGE_LOG)
                    .set(CHANGE_LOG.MESSAGE, changeLog.message())
                    .set(CHANGE_LOG.PARENT_ID, changeLog.parentReference().id())
                    .set(CHANGE_LOG.PARENT_KIND, changeLog.parentReference().kind().name())
                    .set(CHANGE_LOG.USER_ID, changeLog.userId())
                    .set(CHANGE_LOG.SEVERITY, changeLog.severity().name())
                    .set(CHANGE_LOG.CHILD_KIND, changeLog.childKind().map(ck -> ck.name()).orElse(null))
                    .set(CHANGE_LOG.OPERATION, changeLog.operation().name())
                    .set(CHANGE_LOG.CREATED_AT, Timestamp.valueOf(changeLog.createdAt())))
            .toArray(Query[]::new);
    return dsl.batch(queries).execute();
}

@Override
public void addPermission(final String username, final PermissionView permissionView)
        throws FormException {
    final Query query =
            jooq.insertInto(USERS_PERMISSIONS,
                            USERS_PERMISSIONS.USER_ID,
                            USERS_PERMISSIONS.PERMISSION_ID,
                            USERS_PERMISSIONS.GROUP_SCOPE)
                .values(getUserId(username),
                        JooqDaoUtils.getPermissionId(permissionView.getName()),
                        permissionView.isGroupScoped()
                                ? JooqDaoUtils.getContestId(permissionView.getScope())
                                : null);
    try {
        query.execute();
    } catch (DataIntegrityViolationException ex) {
        throw new FormException().addError("permission", "unique");
    }
}

@Override
public void addPermissionGroup(final String username,
                               final PermissionGroupView permissionGroupView)
        throws FormException {
    final Query query =
            jooq.insertInto(USERS_PERMISSION_GROUPS,
                            USERS_PERMISSION_GROUPS.USER_ID,
                            USERS_PERMISSION_GROUPS.PERMISSION_GROUP_ID,
                            USERS_PERMISSION_GROUPS.GROUP_SCOPE)
                .values(getUserId(username),
                        JooqDaoUtils.getPermissionGroupId(permissionGroupView.getCode()),
                        permissionGroupView.isGroupScoped()
                                ? JooqDaoUtils.getContestId(permissionGroupView.getScope())
                                : null);
    try {
        query.execute();
    } catch (DataIntegrityViolationException ex) {
        throw new FormException().addError("permission", "unique");
    }
}

@Override
public PermissionGroupView createPermissionGroup(final PermissionGroupView permissionGroupView)
        throws FormException {
    final Query query =
            jooq.insertInto(PERMISSION_GROUPS,
                            PERMISSION_GROUPS.CODE,
                            PERMISSION_GROUPS.NAME,
                            PERMISSION_GROUPS.GROUP_SCOPED)
                .values(permissionGroupView.getCode(),
                        permissionGroupView.getName(),
                        permissionGroupView.isGroupScoped());
    try {
        query.execute();
    } catch (DataIntegrityViolationException ex) {
        throw new FormException().addError("code", "unique");
    }
    return permissionGroupView;
}

@Override
public <P> Single<List<P>> fetch(Query query, java.util.function.Function<JsonObject, P> mapper){
    return getConnection().flatMap(executeAndClose(sqlConnection ->
            sqlConnection.rxQueryWithParams(query.getSQL(), getBindValues(query)).map(rs ->
                            rs.getRows().stream().map(mapper).collect(Collectors.toList())
            )));
}

@Override
public <P> Single<P> fetchOne(Query query, Function<JsonObject, P> mapper){
    return getConnection().flatMap(executeAndClose(sqlConnection ->
        sqlConnection.rxQueryWithParams(query.getSQL(), getBindValues(query)).map(rs -> {
            Optional<P> optional = rs.getRows().stream().findFirst().map(mapper);
            return optional.orElseGet(() -> null);
        })));
}

@Override
public Single<Integer> execute(Query query){
    return getConnection()
            .flatMap(executeAndClose(sqlConnection ->
                    sqlConnection
                            .rxUpdateWithParams(query.getSQL(), getBindValues(query))
                            .map(UpdateResult::getUpdated))
            );
}

@Override
public Single<Long> insertReturning(Query query) {
    return getConnection()
            .flatMap(executeAndClose(sqlConnection ->
                            sqlConnection
                                    .rxUpdateWithParams(query.getSQL(), getBindValues(query))
                                    .map(updateResult -> updateResult.getKeys().getLong(0)))
            );
}

private JsonArray getBindValues(Query query) {
    JsonArray bindValues = new JsonArray();
    for (Param<?> param : query.getParams().values()) {
        Object value = convertToDatabaseType(param);
        if(value==null){
            bindValues.addNull();
        }else{
            bindValues.add(value);
        }
    }
    return bindValues;
}

@Override
public <P> void fetch(Query query, Function<JsonObject, P> mapper, Handler<AsyncResult<List<P>>> resultHandler) {
    getConnection().setHandler(sqlConnectionResult->{
        if(sqlConnectionResult.succeeded()){
            log("Fetch", ()-> query.getSQL(ParamType.INLINED));
            sqlConnectionResult.result().queryWithParams(
                    query.getSQL(),
                    getBindValues(query),
                    executeAndClose(rs -> rs.getRows().stream().map(mapper).collect(Collectors.toList()), sqlConnectionResult.result(), resultHandler)
            );
        }else{
            resultHandler.handle(Future.failedFuture(sqlConnectionResult.cause()));
        }
    });
}

private JsonArray getBindValues(Query query) {
    JsonArray bindValues = new JsonArray();
    for (Param<?> param : query.getParams().values()) {
        Object value = convertToDatabaseType(param);
        if(value==null){
            bindValues.addNull();
        }else{
            bindValues.add(value);
        }
    }
    return bindValues;
}

@Override
public <P> CompletableFuture<List<P>> fetch(Query query, Function<JsonObject, P> mapper){
    return getConnection().thenCompose(sqlConnection -> {
        CompletableFuture<List<P>> cf = new VertxCompletableFuture<>(vertx);
        sqlConnection.queryWithParams(
                query.getSQL(),
                getBindValues(query),
                executeAndClose(rs ->
                        rs.getRows().stream().map(mapper).collect(Collectors.toList()),
                        sqlConnection,
                        cf)
        );
        return cf;
    });
}

@Override
public <P> CompletableFuture<P> fetchOne(Query query, Function<JsonObject, P> mapper){
    return getConnection().thenCompose(sqlConnection -> {
        CompletableFuture<P> cf = new VertxCompletableFuture<P>(vertx);
        sqlConnection.queryWithParams(query.getSQL(), getBindValues(query), executeAndClose(rs -> {
            Optional<P> optional = rs.getRows().stream().findFirst().map(mapper);
            return optional.orElseGet(() -> null);
        }, sqlConnection, cf));
        return cf;
    });
}

@Override
public CompletableFuture<Integer> execute(Query query){
    return getConnection().thenCompose(sqlConnection -> {
        CompletableFuture<Integer> cf = new VertxCompletableFuture<>(vertx);
        JsonArray bindValues = getBindValues(query);
        sqlConnection.updateWithParams(query.getSQL(), bindValues, executeAndClose(UpdateResult::getUpdated,sqlConnection,cf));
        return cf;
    });
}

@Override
public CompletableFuture<Long> insertReturning(Query query) {
    return getConnection().thenCompose(sqlConnection -> {
        CompletableFuture<Long> cf = new VertxCompletableFuture<>(vertx);
        sqlConnection.update(query.getSQL(ParamType.INLINED), executeAndClose(updateResult->updateResult.getKeys().getLong(0), sqlConnection, cf));
        return cf;
    });
}

private JsonArray getBindValues(Query query) {
    JsonArray bindValues = new JsonArray();
    for (Param<?> param : query.getParams().values()) {
        Object value = convertToDatabaseType(param);
        if(value==null){
            bindValues.addNull();
        }else{
            bindValues.add(value);
        }
    }
    return bindValues;
}

private void init() {

        this.context.deleteFrom(CUSTOMER);

        Stream.of("a", "b", "c", "d", "e")
                .forEach(name -> this.context
                .insertInto(CUSTOMER)
                .columns(CUSTOMER.EMAIL)
                .values(name + "@email.com")
                .execute());

        this.context
                .select()
                .from(Customer.CUSTOMER)
                .stream()
                .forEach(cr -> {
                    Long customerId = cr.into(Customer.CUSTOMER.ID).value1();
                    Collection<Query> products =
                            IntStream
                                    .range(1, new Random().nextInt(10))
                                    .mapToObj(i -> this.context
                                            .insertInto(PRODUCT)
                                            .columns(PRODUCT.CUSTOMER_ID, PRODUCT.SKU)
                                            .values(customerId, "sku" + customerId))
                                    .collect(Collectors.toList());
                    this.context.batch(products).execute();
                });
    }

/**
 * Cancel and close the query quietly.
 */
public static void closeQuietly(Query query) {
  try {
    if (query != null) {
      query.cancel();
      log.debug("cancelled {}", query);
      query.close();
      log.debug("closed {}", query);
    }
  } catch (Exception e) {
    log.debug(e.getMessage(), e);
  }
}

private static <T> void setupUnsubscription(
  Subscriber<? super T> subscriber,
  Query query
) {
  subscriber.add(
    Subscriptions.create(
      () -> {
        log.debug("cancelling {}", query);
        Util.closeQuietly(query);
      }
    )
  );
}

private static <T> void setupUnsubscription(Subscriber<? super T> subscriber, Query query) {
  subscriber.add(
    Subscriptions.create(
      () -> closeQuietly(query)
    )
  );
}

public InsertReturningProducer(
  Subscriber<? super T> subscriber,
  Query query,
  Iterable<? extends R> result,
  RecordMapper<? super R, ? extends T> recordMapper
) {
  this.subscriber = subscriber;
  this.query = query;
  this.result = result.iterator();
  this.recordMapper = recordMapper;
}

private void jooqSql() {
    Query query = this.dsl.select(BOOK.TITLE, AUTHOR.FIRST_NAME, AUTHOR.LAST_NAME)
            .from(BOOK).join(AUTHOR).on(BOOK.AUTHOR_ID.equal(AUTHOR.ID))
            .where(BOOK.PUBLISHED_IN.equal(2015));
    Object[] bind = query.getBindValues().toArray(new Object[] {});
    List<String> list = this.jdbc.query(query.getSQL(), bind,
            new RowMapper<String>() {
                @Override
                public String mapRow(ResultSet rs, int rowNum) throws SQLException {
                    return rs.getString(1) + " : " + rs.getString(2) + " "
                            + rs.getString(3);
                }
            });
    System.out.println("jOOQ SQL " + list);
}

private void jooqSql() {
    Query query = this.dsl.select(BOOK.TITLE, AUTHOR.FIRST_NAME, AUTHOR.LAST_NAME)
            .from(BOOK).join(AUTHOR).on(BOOK.AUTHOR_ID.equal(AUTHOR.ID))
            .where(BOOK.PUBLISHED_IN.equal(2015));
    Object[] bind = query.getBindValues().toArray(new Object[] {});
    List<String> list = this.jdbc.query(query.getSQL(), bind,
            new RowMapper<String>() {
                @Override
                public String mapRow(ResultSet rs, int rowNum) throws SQLException {
                    return rs.getString(1) + " : " + rs.getString(2) + " "
                            + rs.getString(3);
                }
            });
    System.out.println("jOOQ SQL " + list);
}

public int[] addApplications(long groupId, List<Long> applicationIds) {
    Query[] queries = applicationIds
            .stream()
            .map(id -> DSL.insertInto(APPLICATION_GROUP_ENTRY)
                    .set(APPLICATION_GROUP_ENTRY.GROUP_ID, groupId)
                    .set(APPLICATION_GROUP_ENTRY.APPLICATION_ID, id)
                    .onDuplicateKeyIgnore())
            .toArray(Query[]::new);
    return dsl.batch(queries).execute();
}

private void jooqSql() {
    Query query = this.dsl.select(BOOK.TITLE, AUTHOR.FIRST_NAME, AUTHOR.LAST_NAME)
            .from(BOOK).join(AUTHOR).on(BOOK.AUTHOR_ID.equal(AUTHOR.ID))
            .where(BOOK.PUBLISHED_IN.equal(2015));
    Object[] bind = query.getBindValues().toArray(new Object[] {});
    List<String> list = this.jdbc.query(query.getSQL(), bind,
            new RowMapper<String>() {
                @Override
                public String mapRow(ResultSet rs, int rowNum) throws SQLException {
                    return rs.getString(1) + " : " + rs.getString(2) + " "
                            + rs.getString(3);
                }
            });
    System.out.println("jOOQ SQL " + list);
}

@Override
public void addPermissionToGroup(final String permissionGroupCode,
                                 final PermissionView permissionView) throws FormException {
    final Query query =
            jooq.insertInto(PERMISSION_GROUPS_PERMISSIONS,
                            PERMISSION_GROUPS_PERMISSIONS.PERMISSION_GROUP_ID,
                            PERMISSION_GROUPS_PERMISSIONS.PERMISSION_ID)
                .values(getPermissionGroupId(permissionGroupCode),
                        JooqDaoUtils.getPermissionId(permissionView.getName()));
    try {
        query.execute();
    } catch (DataIntegrityViolationException ex) {
        throw new FormException().addError("permission", "unique");
    }
}

public static Execute using(Connection connection, QueryBuilder<? extends Query> queryBuilder) {
  return new Execute(connection, queryBuilder);
}

@Override
public Query getInsertStatement(Vertex element) {
    return null;
}

@Override
public CompletableFuture<Integer> execute(final Query query)
{
    return CompletableFuture.supplyAsync(query::execute);
}

@Override
public CompletableFuture<Integer> update(final Query query)
{
    return execute(query);
}

/**
 * @param query a jOOQ-query
 * @param mapper a function to map the result into another object.
 * @param <P> the type to fetch
 * @return A Single returning a List of P.
 */
<P> Single<List<P>> fetch(Query query, Function<JsonObject, P> mapper);

/**
 * @param query a jOOQ-query
 * @param mapper a function to map the result into another object.
 * @param <P> the type to fetch
 * @return A Single returning an object of P or <code>null</code>.
 */
<P> Single<P> fetchOne(Query query, Function<JsonObject, P> mapper);

/**
 * @param query a jOOQ-query
 * @return A Single returning the number of affected rows by this query.
 */
Single<Integer> execute(Query query);

/**
 * @param query a jOOQ-query to run the insert statement
 * @return A Single returning the lastId returned by mysql.
 */
Single<Long> insertReturning(Query query);

/**
 *
 * @param query a jOOQ-query
 * @param mapper a function to map the result into another object.
 * @param resultHandler A Handler containing the fetched results, each converted by the mapper.
 * @param <P> the type to fetch
 */
<P> void fetch(Query query, Function<JsonObject, P> mapper, Handler<AsyncResult<List<P>>> resultHandler);