diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraBatchTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraBatchTemplate.java index 4bac9c9e4..65e42efbc 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraBatchTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraBatchTemplate.java @@ -39,13 +39,14 @@ import com.datastax.oss.driver.api.core.cql.SimpleStatement; * @author Mark Paluch * @author John Blum * @author Anup Sabbi + * @author Sam Lightfoot * @since 1.5 */ class CassandraBatchTemplate implements CassandraBatchOperations { private final AtomicBoolean executed = new AtomicBoolean(); - private final BatchStatementBuilder batch = BatchStatement.builder(BatchType.LOGGED); + private final BatchStatementBuilder batch; private final CassandraConverter converter; @@ -61,10 +62,23 @@ class CassandraBatchTemplate implements CassandraBatchOperations { * @param operations must not be {@literal null}. */ CassandraBatchTemplate(CassandraOperations operations) { + this(operations, BatchType.LOGGED); + } + + /** + * Create a new {@link CassandraBatchTemplate} given {@link CassandraOperations} and {@link BatchType}. + * + * @param operations must not be {@literal null}. + * @param batchType must not be {@literal null}. + * @since 3.3.0 + */ + CassandraBatchTemplate(CassandraOperations operations, BatchType batchType) { Assert.notNull(operations, "CassandraOperations must not be null"); + Assert.notNull(batchType, "BatchType must not be null"); this.operations = operations; + this.batch = BatchStatement.builder(batchType); this.converter = operations.getConverter(); this.mappingContext = this.converter.getMappingContext(); this.statementFactory = new StatementFactory(new UpdateMapper(converter)); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java index 710f90433..627b4777e 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java @@ -31,6 +31,7 @@ import org.springframework.data.domain.Slice; import org.springframework.lang.Nullable; import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.cql.BatchType; import com.datastax.oss.driver.api.core.cql.ResultSet; import com.datastax.oss.driver.api.core.cql.Statement; @@ -42,6 +43,7 @@ import com.datastax.oss.driver.api.core.cql.Statement; * @author David Webb * @author Matthew Adams * @author Mark Paluch + * @author Sam Lightfoot * @see CassandraTemplate * @see CqlOperations * @see Statement @@ -52,12 +54,22 @@ public interface CassandraOperations extends FluentCassandraOperations { /** * Returns a new {@link CassandraBatchOperations}. Each {@link CassandraBatchOperations} instance can be executed only - * once so you might want to obtain new {@link CassandraBatchOperations} instances for each batch. + * once, so you might want to obtain new {@link CassandraBatchOperations} instances for each batch. * * @return a new {@link CassandraBatchOperations} associated with the given entity class. */ CassandraBatchOperations batchOps(); + /** + * Returns a new {@link CassandraBatchOperations}. Each {@link CassandraBatchOperations} instance can be executed only + * once, so you might want to obtain new {@link CassandraBatchOperations} instances for each batch. + * + * @param batchType must not be {@literal null}. + * @return a new {@link ReactiveCassandraBatchOperations} associated with the given entity class. + * @since 3.3.0 + */ + CassandraBatchOperations batchOps(BatchType batchType); + /** * Expose the underlying {@link CqlOperations} to allow CQL operations. * diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java index 2edfa8f95..69c10edb5 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java @@ -73,6 +73,7 @@ import com.datastax.oss.driver.api.core.CqlIdentifier; import com.datastax.oss.driver.api.core.CqlSession; import com.datastax.oss.driver.api.core.DriverException; import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.cql.BatchType; import com.datastax.oss.driver.api.core.cql.BoundStatement; import com.datastax.oss.driver.api.core.cql.PreparedStatement; import com.datastax.oss.driver.api.core.cql.ResultSet; @@ -109,6 +110,7 @@ import com.datastax.oss.driver.api.querybuilder.update.Update; * @author Mark Paluch * @author John Blum * @author Lukasz Antoniak + * @author Sam Lightfoot * @see org.springframework.data.cassandra.core.CassandraOperations * @since 2.0 */ @@ -202,6 +204,14 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP return new CassandraBatchTemplate(this); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.CassandraOperations#batchOps(com.datastax.oss.driver.api.core.cql.BatchType) + */ + @Override + public CassandraBatchOperations batchOps(BatchType batchType) { + return new CassandraBatchTemplate(this, batchType); + } + /* (non-Javadoc) * @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher) */ diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchTemplate.java index d9c71b173..fd310bb3b 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchTemplate.java @@ -46,13 +46,14 @@ import com.datastax.oss.driver.api.core.cql.SimpleStatement; * * @author Oleh Dokuka * @author Mark Paluch + * @author Sam Lightfoot * @since 2.1 */ class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations { private final AtomicBoolean executed = new AtomicBoolean(); - private final BatchStatementBuilder batch = BatchStatement.builder(BatchType.LOGGED); + private final BatchStatementBuilder batch; private final CassandraConverter converter; @@ -70,10 +71,23 @@ class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations * @param operations must not be {@literal null}. */ ReactiveCassandraBatchTemplate(ReactiveCassandraOperations operations) { + this(operations, BatchType.LOGGED); + } + + /** + * Create a new {@link CassandraBatchTemplate} given {@link CassandraOperations} and {@link BatchType}. + * + * @param operations must not be {@literal null}. + * @param batchType must not be {@literal null}. + * @since 3.3.0 + */ + ReactiveCassandraBatchTemplate(ReactiveCassandraOperations operations, BatchType batchType) { Assert.notNull(operations, "CassandraOperations must not be null"); + Assert.notNull(batchType, "BatchType must not be null"); this.operations = operations; + this.batch = BatchStatement.builder(batchType); this.converter = operations.getConverter(); this.mappingContext = this.converter.getMappingContext(); this.statementFactory = new StatementFactory(new UpdateMapper(converter)); @@ -155,7 +169,6 @@ class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations public ReactiveCassandraBatchOperations withTimestamp(long timestamp) { assertNotExecuted(); - this.batch.setQueryTimestamp(timestamp); return this; diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java index 256dd9082..a3e9eb4eb 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java @@ -29,6 +29,7 @@ import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.cassandra.core.query.Update; import org.springframework.data.domain.Slice; +import com.datastax.oss.driver.api.core.cql.BatchType; import com.datastax.oss.driver.api.core.cql.Statement; /** @@ -38,6 +39,7 @@ import com.datastax.oss.driver.api.core.cql.Statement; * @author Mark Paluch * @author Hleb Albau * @author Oleh Dokuka + * @author Sam Lightfoot * @since 2.0 * @see ReactiveCassandraTemplate * @see ReactiveCqlOperations @@ -51,7 +53,7 @@ public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOper /** * Returns a new {@link ReactiveCassandraBatchOperations}. Each {@link ReactiveCassandraBatchOperations} instance can - * be executed only once so you might want to obtain new {@link ReactiveCassandraBatchOperations} instances for each + * be executed only once, so you might want to obtain new {@link ReactiveCassandraBatchOperations} instances for each * batch. * * @return a new {@link ReactiveCassandraBatchOperations} associated with the given entity class. @@ -59,6 +61,17 @@ public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOper */ ReactiveCassandraBatchOperations batchOps(); + /** + * Returns a new {@link ReactiveCassandraBatchOperations}. Each {@link ReactiveCassandraBatchOperations} instance can + * be executed only once, so you might want to obtain new {@link ReactiveCassandraBatchOperations} instances for each + * batch. + * + * @param batchType must not be {@literal null}. + * @return a new {@link ReactiveCassandraBatchOperations} associated with the given entity class. + * @since 3.3.0 + */ + ReactiveCassandraBatchOperations batchOps(BatchType batchType); + /** * Expose the underlying {@link ReactiveCqlOperations} to allow CQL operations. * diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java index 1e33de064..cd696a909 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java @@ -69,6 +69,7 @@ import com.datastax.oss.driver.api.core.CqlIdentifier; import com.datastax.oss.driver.api.core.DriverException; import com.datastax.oss.driver.api.core.config.DefaultDriverOption; import com.datastax.oss.driver.api.core.context.DriverContext; +import com.datastax.oss.driver.api.core.cql.BatchType; import com.datastax.oss.driver.api.core.cql.BoundStatement; import com.datastax.oss.driver.api.core.cql.PreparedStatement; import com.datastax.oss.driver.api.core.cql.Row; @@ -105,6 +106,7 @@ import com.datastax.oss.driver.api.querybuilder.update.Update; * @author John Blum * @author Lukasz Antoniak * @author Hleb Albau + * @author Sam Lightfoot * @since 2.0 */ public class ReactiveCassandraTemplate @@ -199,6 +201,14 @@ public class ReactiveCassandraTemplate return new ReactiveCassandraBatchTemplate(this); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#batchOps(com.datastax.oss.driver.api.core.cql.BatchType) + */ + @Override + public ReactiveCassandraBatchOperations batchOps(BatchType batchType) { + return new ReactiveCassandraBatchTemplate(this, batchType); + } + /* (non-Javadoc) * @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher) */