Skip to content

Commit adb4c8c

Browse files
mp911demp911de
committed
Introduce customization hook for PreparedStatementHandler.
CassandraTemplate and its asynchronous and reactive variants now define a createPreparedStatementHandler() template method to customize CQL preparation and binding. Closes #1237
1 parent f386f82 commit adb4c8c

File tree

3 files changed

+115
-38
lines changed

3 files changed

+115
-38
lines changed

‎spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java

+46-7
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
import java.util.Collections;
1919
import java.util.List;
20+
import java.util.concurrent.CompletionStage;
2021
import java.util.function.Consumer;
2122
import java.util.function.Function;
2223
import java.util.stream.Collectors;
@@ -51,6 +52,7 @@
5152
import org.springframework.dao.DataAccessException;
5253
import org.springframework.dao.OptimisticLockingFailureException;
5354
import org.springframework.dao.support.DataAccessUtils;
55+
import org.springframework.dao.support.PersistenceExceptionTranslator;
5456
import org.springframework.data.cassandra.SessionFactory;
5557
import org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity;
5658
import org.springframework.data.cassandra.core.convert.CassandraConverter;
@@ -824,6 +826,19 @@ public ListenableFuture<Void> truncate(Class<?> entityClass) {
824826
// Implementation hooks and utility methods
825827
// -------------------------------------------------------------------------
826828

829+
/**
830+
* Create a new statement-based {@link AsyncPreparedStatementHandler} using the statement passed in.
831+
* <p>
832+
* This method allows for the creation to be overridden by subclasses.
833+
*
834+
* @param statement the statement to be prepared.
835+
* @return the new {@link PreparedStatementHandler} to use.
836+
* @since 3.3.3
837+
*/
838+
protected AsyncPreparedStatementHandler createPreparedStatementHandler(Statement<?> statement) {
839+
return new PreparedStatementHandler(statement, exceptionTranslator);
840+
}
841+
827842
private <T> ListenableFuture<EntityWriteResult<T>> executeSave(T entity, CqlIdentifier tableName,
828843
SimpleStatement statement) {
829844

@@ -876,7 +891,7 @@ private <T> ListenableFuture<List<T>> doQuery(Statement<?> statement, RowMapper<
876891

877892
if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
878893

879-
PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
894+
AsyncPreparedStatementHandler statementHandler = createPreparedStatementHandler(statement);
880895
return getAsyncCqlOperations().query(statementHandler, statementHandler, rowMapper);
881896
}
882897

@@ -887,7 +902,7 @@ private ListenableFuture<Void> doQuery(Statement<?> statement, RowCallbackHandle
887902

888903
if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
889904

890-
PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
905+
AsyncPreparedStatementHandler statementHandler = createPreparedStatementHandler(statement);
891906
return getAsyncCqlOperations().query(statementHandler, statementHandler, callbackHandler);
892907
}
893908

@@ -902,7 +917,7 @@ private <T> ListenableFuture<T> doExecute(Statement<?> statement, Function<Async
902917

903918
if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
904919

905-
PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
920+
AsyncPreparedStatementHandler statementHandler = createPreparedStatementHandler(statement);
906921
return getAsyncCqlOperations().query(statementHandler, statementHandler,
907922
(AsyncResultSetExtractor<T>) resultSet -> new AsyncResult<>(mappingFunction.apply(resultSet)));
908923
}
@@ -1029,19 +1044,33 @@ protected T adapt(@Nullable S adapteeResult) {
10291044
}
10301045
}
10311046

1047+
/**
1048+
* General callback interface used to create and bind prepared CQL statements.
1049+
* <p>
1050+
* This interface prepares the CQL statement and sets values on a {@link PreparedStatement} as union-type comprised
1051+
* from {@link AsyncPreparedStatementCreator}, {@link PreparedStatementBinder}, and {@link CqlProvider}.
1052+
*
1053+
* @since 3.3.3
1054+
*/
1055+
public interface AsyncPreparedStatementHandler
1056+
extends AsyncPreparedStatementCreator, PreparedStatementBinder, CqlProvider {
1057+
1058+
}
1059+
10321060
/**
10331061
* Utility class to prepare a {@link SimpleStatement} and bind values associated with the statement to a
10341062
* {@link BoundStatement}.
10351063
*
10361064
* @since 3.2
10371065
*/
1038-
private class PreparedStatementHandler
1039-
implements AsyncPreparedStatementCreator, PreparedStatementBinder, CqlProvider {
1066+
public static class PreparedStatementHandler implements AsyncPreparedStatementHandler {
10401067

10411068
private final SimpleStatement statement;
1069+
private final PersistenceExceptionTranslator exceptionTranslator;
10421070

1043-
public PreparedStatementHandler(Statement<?> statement) {
1071+
public PreparedStatementHandler(Statement<?> statement, PersistenceExceptionTranslator exceptionTranslator) {
10441072
this.statement = PreparedStatementDelegate.getStatementForPrepare(statement);
1073+
this.exceptionTranslator = exceptionTranslator;
10451074
}
10461075

10471076
/*
@@ -1050,7 +1079,17 @@ public PreparedStatementHandler(Statement<?> statement) {
10501079
*/
10511080
@Override
10521081
public ListenableFuture<PreparedStatement> createPreparedStatement(CqlSession session) throws DriverException {
1053-
return new CassandraFutureAdapter<>(session.prepareAsync(statement), exceptionTranslator);
1082+
return new CassandraFutureAdapter<>(doPrepare(session), exceptionTranslator);
1083+
}
1084+
1085+
/**
1086+
* Invokes the statement preparation.
1087+
*
1088+
* @param session
1089+
* @return
1090+
*/
1091+
protected CompletionStage<PreparedStatement> doPrepare(CqlSession session) {
1092+
return session.prepareAsync(statement);
10541093
}
10551094

10561095
/*

‎spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java

+16-3
Original file line numberDiff line numberDiff line change
@@ -885,6 +885,19 @@ public ExecutableDelete delete(Class<?> domainType) {
885885
// Implementation hooks and utility methods
886886
// -------------------------------------------------------------------------
887887

888+
/**
889+
* Create a new statement-based {@link PreparedStatementHandler} using the statement passed in.
890+
* <p>
891+
* This method allows for the creation to be overridden by subclasses.
892+
*
893+
* @param statement the statement to be prepared.
894+
* @return the new {@link PreparedStatementHandler} to use.
895+
* @since 3.3.3
896+
*/
897+
protected PreparedStatementHandler createPreparedStatementHandler(Statement<?> statement) {
898+
return new PreparedStatementHandler(statement);
899+
}
900+
888901
private <T> EntityWriteResult<T> executeSave(T entity, CqlIdentifier tableName, SimpleStatement statement) {
889902
return executeSave(entity, tableName, statement, ignore -> {});
890903
}
@@ -921,7 +934,7 @@ private <T> List<T> doQuery(Statement<?> statement, RowMapper<T> rowMapper) {
921934

922935
if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
923936

924-
PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
937+
PreparedStatementHandler statementHandler = createPreparedStatementHandler(statement);
925938
return getCqlOperations().query(statementHandler, statementHandler, rowMapper);
926939
}
927940

@@ -936,7 +949,7 @@ private <T> Stream<T> doQueryForStream(Statement<?> statement, RowMapper<T> rowM
936949

937950
if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
938951

939-
PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
952+
PreparedStatementHandler statementHandler = createPreparedStatementHandler(statement);
940953
return getCqlOperations().queryForStream(statementHandler, statementHandler, rowMapper);
941954
}
942955

@@ -955,7 +968,7 @@ private <T> T doExecute(Statement<?> statement, Function<ResultSet, T> mappingFu
955968

956969
if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
957970

958-
PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
971+
PreparedStatementHandler statementHandler = createPreparedStatementHandler(statement);
959972
return getCqlOperations().query(statementHandler, statementHandler, mappingFunction::apply);
960973
}
961974

‎spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java

+53-28
Original file line numberDiff line numberDiff line change
@@ -15,33 +15,17 @@
1515
*/
1616
package org.springframework.data.cassandra.core;
1717

18+
import reactor.core.publisher.Flux;
19+
import reactor.core.publisher.Mono;
20+
import reactor.core.publisher.SynchronousSink;
21+
1822
import java.util.Collections;
1923
import java.util.function.BiConsumer;
2024
import java.util.function.Function;
2125

22-
import com.datastax.oss.driver.api.core.CqlIdentifier;
23-
import com.datastax.oss.driver.api.core.DriverException;
24-
import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
25-
import com.datastax.oss.driver.api.core.context.DriverContext;
26-
import com.datastax.oss.driver.api.core.cql.BatchType;
27-
import com.datastax.oss.driver.api.core.cql.BoundStatement;
28-
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
29-
import com.datastax.oss.driver.api.core.cql.Row;
30-
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
31-
import com.datastax.oss.driver.api.core.cql.Statement;
32-
import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
33-
import com.datastax.oss.driver.api.querybuilder.delete.Delete;
34-
import com.datastax.oss.driver.api.querybuilder.insert.Insert;
35-
import com.datastax.oss.driver.api.querybuilder.insert.RegularInsert;
36-
import com.datastax.oss.driver.api.querybuilder.select.Select;
37-
import com.datastax.oss.driver.api.querybuilder.truncate.Truncate;
38-
import com.datastax.oss.driver.api.querybuilder.update.Update;
3926
import org.reactivestreams.Publisher;
4027
import org.slf4j.Logger;
4128
import org.slf4j.LoggerFactory;
42-
import reactor.core.publisher.Flux;
43-
import reactor.core.publisher.Mono;
44-
import reactor.core.publisher.SynchronousSink;
4529

4630
import org.springframework.beans.BeansException;
4731
import org.springframework.context.ApplicationContext;
@@ -82,6 +66,24 @@
8266
import org.springframework.lang.Nullable;
8367
import org.springframework.util.Assert;
8468

69+
import com.datastax.oss.driver.api.core.CqlIdentifier;
70+
import com.datastax.oss.driver.api.core.DriverException;
71+
import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
72+
import com.datastax.oss.driver.api.core.context.DriverContext;
73+
import com.datastax.oss.driver.api.core.cql.BatchType;
74+
import com.datastax.oss.driver.api.core.cql.BoundStatement;
75+
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
76+
import com.datastax.oss.driver.api.core.cql.Row;
77+
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
78+
import com.datastax.oss.driver.api.core.cql.Statement;
79+
import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
80+
import com.datastax.oss.driver.api.querybuilder.delete.Delete;
81+
import com.datastax.oss.driver.api.querybuilder.insert.Insert;
82+
import com.datastax.oss.driver.api.querybuilder.insert.RegularInsert;
83+
import com.datastax.oss.driver.api.querybuilder.select.Select;
84+
import com.datastax.oss.driver.api.querybuilder.truncate.Truncate;
85+
import com.datastax.oss.driver.api.querybuilder.update.Update;
86+
8587
/**
8688
* Primary implementation of {@link ReactiveCassandraOperations}. It simplifies the use of Reactive Cassandra usage and
8789
* helps to avoid common errors. It executes core Cassandra workflow. This class executes CQL queries or updates,
@@ -852,6 +854,19 @@ public ReactiveUpdate update(Class<?> domainType) {
852854
// Implementation hooks and utility methods
853855
// -------------------------------------------------------------------------
854856

857+
/**
858+
* Create a new statement-based {@link ReactivePreparedStatementHandler} using the statement passed in.
859+
* <p>
860+
* This method allows for the creation to be overridden by subclasses.
861+
*
862+
* @param statement the statement to be prepared.
863+
* @return the new {@link PreparedStatementHandler} to use.
864+
* @since 3.3.3
865+
*/
866+
protected ReactivePreparedStatementHandler createPreparedStatementHandler(Statement<?> statement) {
867+
return new PreparedStatementHandler(statement);
868+
}
869+
855870
private <T> Mono<EntityWriteResult<T>> executeSave(T entity, CqlIdentifier tableName, SimpleStatement statement) {
856871
return executeSave(entity, tableName, statement, (writeResult, sink) -> sink.next(writeResult));
857872
}
@@ -888,7 +903,7 @@ private <T> Flux<T> doQuery(Statement<?> statement, RowMapper<T> rowMapper) {
888903

889904
if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
890905

891-
PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
906+
ReactivePreparedStatementHandler statementHandler = createPreparedStatementHandler(statement);
892907
return getReactiveCqlOperations().query(statementHandler, statementHandler, rowMapper);
893908
}
894909

@@ -899,7 +914,7 @@ private <T> Mono<T> doExecute(Statement<?> statement, Function<ReactiveResultSet
899914

900915
if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
901916

902-
PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
917+
ReactivePreparedStatementHandler statementHandler = createPreparedStatementHandler(statement);
903918
return getReactiveCqlOperations()
904919
.query(statementHandler, statementHandler, rs -> Mono.just(mappingFunction.apply(rs))).next();
905920
}
@@ -912,7 +927,7 @@ private <T> Mono<T> doExecuteAndFlatMap(Statement<?> statement,
912927

913928
if (PreparedStatementDelegate.canPrepare(isUsePreparedStatements(), statement, logger)) {
914929

915-
PreparedStatementHandler statementHandler = new PreparedStatementHandler(statement);
930+
ReactivePreparedStatementHandler statementHandler = createPreparedStatementHandler(statement);
916931
return getReactiveCqlOperations().query(statementHandler, statementHandler, mappingFunction::apply).next();
917932
}
918933

@@ -948,9 +963,7 @@ public String getCql() {
948963
}
949964
}
950965

951-
return getReactiveCqlOperations()
952-
.execute(new GetConfiguredPageSize())
953-
.single();
966+
return getReactiveCqlOperations().execute(new GetConfiguredPageSize()).single();
954967
}
955968

956969
@SuppressWarnings("unchecked")
@@ -1020,14 +1033,26 @@ protected <T> Mono<T> maybeCallBeforeSave(T object, CqlIdentifier tableName, Sta
10201033
return Mono.just(object);
10211034
}
10221035

1036+
/**
1037+
* General callback interface used to create and bind prepared CQL statements.
1038+
* <p>
1039+
* This interface prepares the CQL statement and sets values on a {@link PreparedStatement} as union-type comprised
1040+
* from {@link ReactivePreparedStatementCreator}, {@link PreparedStatementBinder}, and {@link CqlProvider}.
1041+
*
1042+
* @since 3.3.3
1043+
*/
1044+
public interface ReactivePreparedStatementHandler
1045+
extends ReactivePreparedStatementCreator, PreparedStatementBinder, CqlProvider {
1046+
1047+
}
1048+
10231049
/**
10241050
* Utility class to prepare a {@link SimpleStatement} and bind values associated with the statement to a
10251051
* {@link BoundStatement}.
10261052
*
10271053
* @since 3.2
10281054
*/
1029-
private static class PreparedStatementHandler
1030-
implements ReactivePreparedStatementCreator, PreparedStatementBinder, CqlProvider {
1055+
public static class PreparedStatementHandler implements ReactivePreparedStatementHandler {
10311056

10321057
private final SimpleStatement statement;
10331058

0 commit comments

Comments
 (0)