Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polishing.

Browse Source 
Consistently use toCql(…) and String.format for CQL retrieval.

See #1186
This commit is contained in:
Mark Paluch
2021-10-25 09:10:02 +02:00
parent 99ad37985a
commit d9ddf0de16
5 changed files with 80 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/AsyncCqlTemplate.java
View File

@@ -22,6 +22,14 @@ import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.DriverException;
import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
import com.datastax.oss.driver.api.core.cql.ResultSet;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.cql.Statement;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.dao.support.PersistenceExceptionTranslator;
@@ -32,14 +40,6 @@ import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SettableListenableFuture;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.DriverException;
import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
import com.datastax.oss.driver.api.core.cql.ResultSet;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.cql.Statement;
/**
* <b>This is the central class in the CQL core package for asynchronous Cassandra data access.</b> It simplifies the
* use of CQL and helps to avoid common errors. It executes core CQL workflow, leaving application code to provide CQL
@@ -163,7 +163,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
try {
if (logger.isDebugEnabled()) {
logger.debug("Executing CQL statement [{}]", cql);
logger.debug(String.format("Executing CQL statement [%s]", cql));
}
CompletionStage<T> results = getCurrentSession().executeAsync(applyStatementSettings(newStatement(cql)))
@@ -285,7 +285,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
try {
if (logger.isDebugEnabled()) {
logger.debug("Executing statement [{}]", QueryExtractorDelegate.getCql(statement));
logger.debug(String.format("Executing statement [%s]", toCql(statement)));
}
CompletionStage<T> results = getCurrentSession() //
@@ -294,9 +294,9 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
.thenCompose(ListenableFuture::completable);
return new CassandraFutureAdapter<>(results,
ex -> translateExceptionIfPossible("Query", statement.toString(), ex));
ex -> translateExceptionIfPossible("Query", toCql(statement), ex));
} catch (DriverException e) {
throw translateException("Query", statement.toString(), e);
throw translateException("Query", toCql(statement), e);
}
}
@@ -446,7 +446,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
toCql(preparedStatementCreator), ex);
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
logger.debug(String.format("Preparing statement [%s] using %s", toCql(preparedStatementCreator), preparedStatementCreator));
}
CqlSession currentSession = getCurrentSession();
@@ -517,7 +517,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
logger.debug(String.format("Preparing statement [%s] using %s", toCql(preparedStatementCreator), preparedStatementCreator));
}
CqlSession session = getCurrentSession();
@@ -525,7 +525,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
ListenableFuture<Statement<?>> statementFuture = new MappingListenableFutureAdapter<>(
preparedStatementCreator.createPreparedStatement(session), preparedStatement -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing prepared statement [{}]", QueryExtractorDelegate.getCql(preparedStatement));
logger.debug(String.format("Executing prepared statement [%s]", toCql(preparedStatement)));
}
return applyStatementSettings(psb != null ? psb.bindValues(preparedStatement) : preparedStatement.bind());

24
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/CqlTemplate.java
View File

@@ -25,12 +25,6 @@ import java.util.function.Function;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.data.cassandra.SessionFactory;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.DriverException;
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
@@ -40,6 +34,12 @@ import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.cql.Statement;
import com.datastax.oss.driver.api.core.metadata.Node;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.data.cassandra.SessionFactory;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* <b>This is the central class in the CQL core package.</b> It simplifies the use of CQL and helps to avoid common
* errors. It executes core CQL workflow, leaving application code to provide CQL and extract results. This class
@@ -166,7 +166,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
try {
if (logger.isDebugEnabled()) {
logger.debug("Executing CQL statement [{}]", cql);
logger.debug(String.format("Executing CQL statement [%s]", cql));
}
Statement<?> statement = applyStatementSettings(newStatement(cql));
@@ -292,12 +292,12 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
try {
if (logger.isDebugEnabled()) {
logger.debug("Executing statement [{}]", QueryExtractorDelegate.getCql(statement));
logger.debug(String.format("Executing statement [%s]", toCql(statement)));
}
return resultSetExtractor.extractData(getCurrentSession().execute(applyStatementSettings(statement)));
} catch (DriverException e) {
throw translateException("Query", statement.toString(), e);
throw translateException("Query", toCql(statement), e);
}
}
@@ -453,7 +453,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
logger.debug(String.format("Preparing statement [%s] using %s", toCql(preparedStatementCreator), preparedStatementCreator));
}
CqlSession session = getCurrentSession();
@@ -523,7 +523,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
logger.debug(String.format("Preparing statement [%s] using %s", toCql(preparedStatementCreator), preparedStatementCreator));
}
CqlSession session = getCurrentSession();
@@ -531,7 +531,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
PreparedStatement preparedStatement = preparedStatementCreator.createPreparedStatement(session);
if (logger.isDebugEnabled()) {
logger.debug("Executing prepared statement [{}]", QueryExtractorDelegate.getCql(preparedStatement));
logger.debug(String.format("Executing prepared statement [%s]", QueryExtractorDelegate.getCql(preparedStatement)));
}
Statement<?> boundStatement = applyStatementSettings(

16
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/ReactiveCassandraAccessor.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.data.cassandra.core.cql;
import com.datastax.oss.driver.api.core.DriverException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -25,8 +26,6 @@ import org.springframework.data.cassandra.ReactiveSessionFactory;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import com.datastax.oss.driver.api.core.DriverException;
/**
* Base class for {@link ReactiveCqlTemplate} and other CQL-accessing DAO helpers, defining common properties such as
* {@link ReactiveSessionFactory} and exception translator.
@@ -149,4 +148,17 @@ public abstract class ReactiveCassandraAccessor implements InitializingBean {
return getExceptionTranslator().translate(task, cql, ex);
}
/**
* Determine CQL from potential provider object.
*
* @param cqlProvider object that's potentially a {@link CqlProvider}
* @return the CQL string, or {@literal null}
* @see CqlProvider
* @since 3.2.7
*/
@Nullable
protected static String toCql(@Nullable Object cqlProvider) {
return QueryExtractorDelegate.getCql(cqlProvider);
}
}

66
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/ReactiveCqlTemplate.java
View File

@@ -15,23 +15,9 @@
*/
package org.springframework.data.cassandra.core.cql;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.Map;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.data.cassandra.ReactiveResultSet;
import org.springframework.data.cassandra.ReactiveSession;
import org.springframework.data.cassandra.ReactiveSessionFactory;
import org.springframework.data.cassandra.core.cql.session.DefaultReactiveSessionFactory;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import com.datastax.oss.driver.api.core.ConsistencyLevel;
import com.datastax.oss.driver.api.core.CqlIdentifier;
import com.datastax.oss.driver.api.core.CqlSession;
@@ -43,6 +29,18 @@ import com.datastax.oss.driver.api.core.cql.Row;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.cql.Statement;
import com.datastax.oss.driver.api.core.retry.RetryPolicy;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.data.cassandra.ReactiveResultSet;
import org.springframework.data.cassandra.ReactiveSession;
import org.springframework.data.cassandra.ReactiveSessionFactory;
import org.springframework.data.cassandra.core.cql.session.DefaultReactiveSessionFactory;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* <b>This is the central class in the CQL core package for reactive Cassandra data access.</b> It simplifies the use of
@@ -302,7 +300,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
Assert.notNull(action, "Callback object must not be null");
return createFlux(action).onErrorMap(translateException("ReactiveSessionCallback", getCql(action)));
return createFlux(action).onErrorMap(translateException("ReactiveSessionCallback", toCql(action)));
}
// -------------------------------------------------------------------------
@@ -439,11 +437,12 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
return createFlux(statement, (session, stmt) -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing statement [{}]", QueryExtractorDelegate.getCql(statement));
logger.debug(String.format("Executing statement [%s]", toCql(statement)));
}
return session.execute(applyStatementSettings(statement)).flatMapMany(rse::extractData);
}).onErrorMap(translateException("Query", statement.toString()));
return session.execute(applyStatementSettings(statement))
.flatMapMany(rse::extractData);
}).onErrorMap(translateException("Query", toCql(statement)));
}
/* (non-Javadoc)
@@ -506,17 +505,17 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
return createMono(statement, (session, executedStatement) -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing statement [{}]", QueryExtractorDelegate.getCql(statement));
logger.debug(String.format("Executing statement [%s]", toCql(statement)));
}
return session.execute(applyStatementSettings(executedStatement));
}).onErrorMap(translateException("QueryForResultSet", statement.toString()));
}).onErrorMap(translateException("QueryForResultSet", toCql(statement)));
}
@Override
public Flux<Row> queryForRows(Statement<?> statement) throws DataAccessException {
return queryForResultSet(statement).flatMapMany(ReactiveResultSet::rows)
.onErrorMap(translateException("QueryForRows", statement.toString()));
.onErrorMap(translateException("QueryForRows", toCql(statement)));
}
// -------------------------------------------------------------------------
@@ -535,10 +534,11 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
return createFlux(session -> {
logger.debug("Preparing statement [{}] using {}", getCql(psc), psc);
logger.debug(String.format("Preparing statement [%s] using %s", toCql(psc), psc));
return psc.createPreparedStatement(session).flatMapMany(ps -> action.doInPreparedStatement(session, ps));
}).onErrorMap(translateException("ReactivePreparedStatementCallback", getCql(psc)));
return psc.createPreparedStatement(session)
.flatMapMany(ps -> action.doInPreparedStatement(session, ps));
}).onErrorMap(translateException("ReactivePreparedStatementCallback", toCql(psc)));
}
/* (non-Javadoc)
@@ -569,7 +569,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
return execute(psc, (session, preparedStatement) -> Mono.just(preparedStatement).flatMapMany(pps -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing prepared statement [{}]", QueryExtractorDelegate.getCql(preparedStatement));
logger.debug(String.format("Executing prepared statement [%s]", toCql(preparedStatement)));
}
BoundStatement boundStatement = (preparedStatementBinder != null
@@ -577,7 +577,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
: preparedStatement.bind());
return session.execute(applyStatementSettings(boundStatement));
}).flatMap(rse::extractData)).onErrorMap(translateException("Query", getCql(psc)));
}).flatMap(rse::extractData)).onErrorMap(translateException("Query", toCql(psc)));
}
/* (non-Javadoc)
@@ -737,7 +737,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
return execute(newReactivePreparedStatementCreator(cql), (session, ps) -> Flux.from(args).flatMap(objects -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing prepared CQL statement [{}]", cql);
logger.debug(String.format("Executing prepared CQL statement [%s]", cql));
}
BoundStatement boundStatement = newArgPreparedStatementBinder(objects).bindValues(ps);
@@ -908,18 +908,6 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
return sessionFactory.getSession();
}
/**
* Determine CQL from potential provider object.
*
* @param cqlProvider object that's potentially a {@link CqlProvider}
* @return the CQL string, or {@literal null}
* @see CqlProvider
*/
@Nullable
private static String getCql(@Nullable Object cqlProvider) {
return QueryExtractorDelegate.getCql(cqlProvider);
}
static class SimpleReactivePreparedStatementCreator implements ReactivePreparedStatementCreator, CqlProvider {
private final SimpleStatement statement;

26
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java
View File

@@ -15,24 +15,12 @@
*/
package org.springframework.data.cassandra.core.cql.session;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.scheduler.Scheduler;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletionStage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.cassandra.ReactiveResultSet;
import org.springframework.data.cassandra.ReactiveSession;
import org.springframework.util.Assert;
import com.datastax.oss.driver.api.core.CqlIdentifier;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.context.DriverContext;
@@ -47,6 +35,16 @@ import com.datastax.oss.driver.api.core.cql.Row;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.cql.Statement;
import com.datastax.oss.driver.api.core.metadata.Metadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.scheduler.Scheduler;
import org.springframework.data.cassandra.ReactiveResultSet;
import org.springframework.data.cassandra.ReactiveSession;
import org.springframework.util.Assert;
/**
* Default implementation of a {@link ReactiveSession}. This implementation bridges asynchronous {@link CqlSession}
@@ -166,7 +164,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession {
return Mono.fromCompletionStage(() -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing statement [{}]", getCql(statement));
logger.debug(String.format("Executing statement [%s]", getCql(statement)));
}
return this.session.executeAsync(statement);
@@ -195,7 +193,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession {
return Mono.fromCompletionStage(() -> {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}]", getCql(statement));
logger.debug(String.format("Preparing statement [%s]", getCql(statement)));
}
return this.session.prepareAsync(statement);