diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchOperations.java new file mode 100644 index 000000000..7f6a9767a --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchOperations.java @@ -0,0 +1,183 @@ +/* + * Copyright 2016-2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.cassandra.core; + +import reactor.core.publisher.Mono; + +import org.springframework.data.cassandra.core.cql.WriteOptions; + +/** + * Reactive Batch operations for insert/update/delete actions on a table. {@link ReactiveCassandraBatchOperations} use logged Cassandra + * {@code BATCH}es for single entities and collections of entities. A {@link ReactiveCassandraBatchOperations} instance cannot + * be modified/used once it was executed. + *

+ * Batches are atomic by default. In the context of a Cassandra batch operation, atomic means that if any of the batch + * succeeds, all of it will. Statement order does not matter within a batch. {@link ReactiveCassandraBatchOperations} applies + * all rows using the same {@link #withTimestamp(long) timestamp} if supplied, otherwise Cassandra will generate a + * timestamp. + *

+ * Multi partition batches should only be used to achieve atomicity for a few writes on different tables. Apart from + * this they should be avoided because they’re too expensive. Single partition batches can be used to get atomicity and + * isolation, they're not much more expensive than normal writes. + * + * @author Oleh Dokuka + * @since 2.0 + */ +public interface ReactiveCassandraBatchOperations { + + /** + * Execute the batch. The batch can be executed only once. + * + * @return the {@link Mono} for the bulk operation. + * @throws IllegalStateException if the batch is executed after it was executed already. + */ + Mono execute(); + + /** + * Apply a given {@code timestamp} to the whole batch. + * + * @param timestamp the timestamp to apply. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + */ + ReactiveCassandraBatchOperations withTimestamp(long timestamp); + + /** + * Add an array of inserts to the batch. + * + * @param entities the entities to insert; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + */ + ReactiveCassandraBatchOperations insert(Object... entities); + + /** + * Add a collection of inserts to the batch. + * + * @param entities the entities to insert; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + */ + ReactiveCassandraBatchOperations insert(Iterable entities); + + /** + * Add a collection of inserts to the batch. + * + * @param entities the entities to insert; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + */ + ReactiveCassandraBatchOperations insert(Mono> entities); + + /** + * Add a collection of inserts with given {@link WriteOptions} to the batch. + * + * @param entities the entities to insert; must not be {@literal null}. + * @param options the WriteOptions to apply; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + * @since 2.0 + */ + ReactiveCassandraBatchOperations insert(Iterable entities, WriteOptions options); + + + /** + * Add a collection of inserts with given {@link WriteOptions} to the batch. + * + * @param entities the entities to insert; must not be {@literal null}. + * @param options the WriteOptions to apply; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + * @since 2.0 + */ + ReactiveCassandraBatchOperations insert(Mono> entities, WriteOptions options); + + /** + * Add an array of updates to the batch. + * + * @param entities the entities to update; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + */ + ReactiveCassandraBatchOperations update(Object... entities); + + /** + * Add a collection of updates to the batch. + * + * @param entities the entities to update; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + */ + ReactiveCassandraBatchOperations update(Iterable entities); + + /** + * Add a collection of updates to the batch. + * + * @param entities the entities to update; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + */ + ReactiveCassandraBatchOperations update(Mono> entities); + + /** + * Add a collection of updates with given {@link WriteOptions} to the batch. + * + * @param entities the entities to update; must not be {@literal null}. + * @param options the WriteOptions to apply; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + * @since 2.0 + */ + ReactiveCassandraBatchOperations update(Iterable entities, WriteOptions options); + + /** + * Add a collection of updates with given {@link WriteOptions} to the batch. + * + * @param entities the entities to update; must not be {@literal null}. + * @param options the WriteOptions to apply; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + * @since 2.0 + */ + ReactiveCassandraBatchOperations update(Mono> entities, WriteOptions options); + + /** + * Add an array of deletes to the batch. + * + * @param entities the entities to delete; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + */ + ReactiveCassandraBatchOperations delete(Object... entities); + + /** + * Add a collection of deletes to the batch. + * + * @param entities the entities to delete; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + */ + ReactiveCassandraBatchOperations delete(Iterable entities); + + /** + * Add a collection of deletes to the batch. + * + * @param entities the entities to delete; must not be {@literal null}. + * @return {@code this} {@link ReactiveCassandraBatchOperations}. + * @throws IllegalStateException if the batch was already executed. + */ + ReactiveCassandraBatchOperations delete(Mono> entities); +} 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 new file mode 100644 index 000000000..122cb6af8 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchTemplate.java @@ -0,0 +1,284 @@ +package org.springframework.data.cassandra.core; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +import com.datastax.driver.core.querybuilder.Batch; +import com.datastax.driver.core.querybuilder.BuiltStatement; +import com.datastax.driver.core.querybuilder.Delete; +import com.datastax.driver.core.querybuilder.Insert; +import com.datastax.driver.core.querybuilder.QueryBuilder; +import com.datastax.driver.core.querybuilder.Update; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.data.cassandra.core.cql.QueryOptions; +import org.springframework.data.cassandra.core.cql.WriteOptions; +import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity; +import org.springframework.data.cassandra.core.mapping.CassandraMappingContext; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; + +public class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations { + + private final AtomicBoolean executed = new AtomicBoolean(); + private final ReactiveCassandraOperations operations; + private final Batch batch; + private final List>> batchMonos; + + public ReactiveCassandraBatchTemplate(ReactiveCassandraOperations cassandraOperations) { + this.operations = cassandraOperations; + this.batch = QueryBuilder.batch(); + this.batchMonos = new ArrayList<>(); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#execute() + */ + @Override + public Mono execute() { + + if (executed.compareAndSet(false, true)) { + + return Flux.merge(batchMonos) + .doOnNext(c -> c.forEach(batch::add)) + .then(operations.getReactiveCqlOperations() + .queryForResultSet(batch)) + .flatMap(resultSet -> + resultSet.rows() + .collectList() + .map(rows -> new WriteResult(resultSet.getAllExecutionInfo(), resultSet.wasApplied(), rows)) + ); + } + + throw new IllegalStateException("This Cassandra Batch was already executed"); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#withTimestamp(long) + */ + @Override + public ReactiveCassandraBatchOperations withTimestamp(long timestamp) { + + assertNotExecuted(); + + batch.using(QueryBuilder.timestamp(timestamp)); + + return this; + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#insert(java.lang.Object[]) + */ + @Override + public ReactiveCassandraBatchOperations insert(Object... entities) { + + Assert.notNull(entities, "Entities must not be null"); + + return insert(Arrays.asList(entities)); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#insert(java.lang.Iterable) + */ + @Override + public ReactiveCassandraBatchOperations insert(Iterable entities) { + return insert(entities, InsertOptions.empty()); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#insert(reactor.core.publisher.Mono) + */ + @Override + public ReactiveCassandraBatchOperations insert(Mono> entities) { + return insert(entities, InsertOptions.empty()); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#insert(java.lang.Iterable, org.springframework.data.cassandra.core.cql.WriteOptions) + */ + @Override + public ReactiveCassandraBatchOperations insert(Iterable entities, WriteOptions options) { + + assertNotExecuted(); + Assert.notNull(entities, "Entities must not be null"); + Assert.notNull(options, "WriteOptions must not be null"); + + batchMonos.add(Mono.just(doInsert(entities, options))); + + return this; + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#insert(reactor.core.publisher.Mono, org.springframework.data.cassandra.core.cql.WriteOptions) + */ + @Override + public ReactiveCassandraBatchOperations insert(Mono> entities, WriteOptions options) { + + assertNotExecuted(); + Assert.notNull(entities, "Entities must not be null"); + Assert.notNull(options, "WriteOptions must not be null"); + + batchMonos.add(entities.map(e -> doInsert(e, options))); + + return this; + } + + Collection doInsert(Iterable entities, WriteOptions options) { + + ArrayList insertQueries = new ArrayList<>(); + CassandraMappingContext mappingContext = operations.getConverter().getMappingContext(); + + for (Object entity : entities) { + + Assert.notNull(entity, "Entity must not be null"); + + BasicCassandraPersistentEntity persistentEntity = mappingContext + .getRequiredPersistentEntity(entity.getClass()); + insertQueries.add(QueryUtils.createInsertQuery(persistentEntity.getTableName().toCql(), entity, options, + operations.getConverter(), persistentEntity)); + } + + return insertQueries; + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#update(java.lang.Object[]) + */ + @Override + public ReactiveCassandraBatchOperations update(Object... entities) { + + Assert.notNull(entities, "Entities must not be null"); + + return update(Arrays.asList(entities)); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#update(java.lang.Iterable) + */ + @Override + public ReactiveCassandraBatchOperations update(Iterable entities) { + return update(entities, UpdateOptions.empty()); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#update(reactor.core.publisher.Mono) + */ + @Override + public ReactiveCassandraBatchOperations update(Mono> entities) { + return update(entities, UpdateOptions.empty()); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#update(java.lang.Iterable, org.springframework.data.cassandra.core.cql.WriteOptions) + */ + @Override + public ReactiveCassandraBatchOperations update(Iterable entities, WriteOptions options) { + + assertNotExecuted(); + Assert.notNull(entities, "Entities must not be null"); + Assert.notNull(options, "WriteOptions must not be null"); + + batchMonos.add(Mono.just(doUpdate(entities, options))); + + return this; + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#update(reactor.core.publisher.Mono, org.springframework.data.cassandra.core.cql.WriteOptions) + */ + @Override + public ReactiveCassandraBatchOperations update(Mono> entities, WriteOptions options) { + + assertNotExecuted(); + Assert.notNull(entities, "Entities must not be null"); + Assert.notNull(options, "WriteOptions must not be null"); + + batchMonos.add(entities.map(e -> doUpdate(e, options))); + + return this; + } + + Collection doUpdate(Iterable entities, WriteOptions options) { + ArrayList updateQueries = new ArrayList<>(); + + for (Object entity : entities) { + + Assert.notNull(entity, "Entity must not be null"); + + updateQueries.add(QueryUtils.createUpdateQuery(getTable(entity), entity, options, operations.getConverter())); + } + + return updateQueries; + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#delete(java.lang.Object[]) + */ + @Override + public ReactiveCassandraBatchOperations delete(Object... entities) { + + Assert.notNull(entities, "Entities must not be null"); + + return delete(Arrays.asList(entities)); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#delete(java.lang.Iterable) + */ + @Override + public ReactiveCassandraBatchOperations delete(Iterable entities) { + + assertNotExecuted(); + Assert.notNull(entities, "Entities must not be null"); + + batchMonos.add(Mono.just(doDelete(entities))); + + return this; + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraBatchOperations#delete(reactor.core.publisher.Mono) + */ + @Override + public ReactiveCassandraBatchOperations delete(Mono> entities) { + + assertNotExecuted(); + Assert.notNull(entities, "Entities must not be null"); + + batchMonos.add(entities.map(this::doDelete)); + + return this; + } + + Collection doDelete(Iterable entities) { + ArrayList deleteQueries = new ArrayList<>(); + + for (Object entity : entities) { + + Assert.notNull(entity, "Entity must not be null"); + + deleteQueries.add(QueryUtils.createDeleteQuery(getTable(entity), entity, QueryOptions.empty(), operations.getConverter())); + } + + return deleteQueries; + } + + private void assertNotExecuted() { + Assert.state(!executed.get(), "This Cassandra Batch was already executed"); + } + + private String getTable(Object entity) { + + Assert.notNull(entity, "Entity must not be null"); + + return operations.getConverter() + .getMappingContext() + .getRequiredPersistentEntity(ClassUtils.getUserClass(entity.getClass())) + .getTableName() + .toCql(); + } +} \ No newline at end of file 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 d6b45ca60..379f157ab 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 @@ -47,6 +47,14 @@ import com.datastax.driver.core.Statement; */ public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOperations { + /** + * 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. + * + * @return a new {@link ReactiveCassandraBatchOperations} associated with the given entity class. + */ + ReactiveCassandraBatchOperations batchOps(); + /** * Returns the underlying {@link CassandraConverter}. * 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 ce7956a25..94ed2c5b6 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 @@ -171,6 +171,14 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A this.projectionFactory = new SpelAwareProxyProjectionFactory(); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#batchOps() + */ + @Override + public ReactiveCassandraBatchOperations batchOps() { + return new ReactiveCassandraBatchTemplate(this); + } + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getConverter() */ diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchTemplateIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchTemplateIntegrationTests.java new file mode 100644 index 000000000..794e362e8 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchTemplateIntegrationTests.java @@ -0,0 +1,466 @@ +/* + * Copyright 2016-2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.cassandra.core; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Random; +import java.util.concurrent.TimeUnit; + +import org.junit.Before; +import org.junit.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; +import reactor.test.StepVerifier; + +import org.springframework.data.cassandra.ReactiveResultSet; +import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; +import org.springframework.data.cassandra.core.cql.ReactiveCqlTemplate; +import org.springframework.data.cassandra.core.cql.WriteOptions; +import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession; +import org.springframework.data.cassandra.domain.FlatGroup; +import org.springframework.data.cassandra.domain.Group; +import org.springframework.data.cassandra.domain.GroupKey; +import org.springframework.data.cassandra.repository.support.SchemaTestUtils; +import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Integration tests for {@link ReactiveCassandraBatchTemplate}. + * + * @author Mark Paluch + * @author Oleh Dokuka + */ +public class ReactiveCassandraBatchTemplateIntegrationTests + extends AbstractKeyspaceCreatingIntegrationTest { + + ReactiveCassandraTemplate template; + + Group walter = new Group(new GroupKey("users", "0x1", "walter")); + Group mike = new Group(new GroupKey("users", "0x1", "mike")); + + @Before + public void setUp() throws Exception { + + MappingCassandraConverter converter = new MappingCassandraConverter(); + CassandraTemplate cassandraTemplate = new CassandraTemplate(this.session, converter); + + DefaultBridgedReactiveSession session = new DefaultBridgedReactiveSession(this.session); + template = new ReactiveCassandraTemplate(new ReactiveCqlTemplate(session), converter); + + SchemaTestUtils.potentiallyCreateTableFor(Group.class, cassandraTemplate); + SchemaTestUtils.potentiallyCreateTableFor(FlatGroup.class, cassandraTemplate); + + SchemaTestUtils.truncate(Group.class, cassandraTemplate); + SchemaTestUtils.truncate(FlatGroup.class, cassandraTemplate); + + this.template.insert(walter) + .then(this.template.insert(mike)) + .block(); + } + + @Test // DATACASS-574 + public void shouldInsertEntities() { + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono execution = batchOperations.insert(walter) + .insert(mike) + .execute(); + + Mono loadedMono = execution.then(template.selectOneById(walter.getId(), Group.class)); + + StepVerifier.create(loadedMono) + .assertNext(loaded -> assertThat(loaded.getId().getUsername()).isEqualTo(walter.getId().getUsername())) + .verifyComplete(); + } + + @Test // DATACASS-574 + @SuppressWarnings("unchecked") + public void shouldInsertEntitiesWithLwt() { + + InsertOptions lwtOptions = InsertOptions.builder().withIfNotExists().build(); + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + + Group previousWalter = new Group(new GroupKey("users", "0x1", "walter")); + previousWalter.setAge(42); + + Flux concatenatedExecution = Flux.concat( + template + .insert(previousWalter) + .then(Mono.fromRunnable(() -> walter.setAge(100))) + .then(Mono.defer(() -> batchOperations + .insert(Collections.singleton(walter), lwtOptions) + .insert(mike) + .execute())), + template.selectOneById(walter.getId(), Group.class), + template.selectOneById(mike.getId(), Group.class) + ); + + StepVerifier.create(concatenatedExecution) + .assertNext(o -> { + WriteResult writeResult = (WriteResult) o; + + assertThat(writeResult.wasApplied()).isFalse(); + assertThat(writeResult.getExecutionInfo()).isNotEmpty(); + assertThat(writeResult.getRows()).isNotEmpty(); + }) + .assertNext(o -> { + Group loadedWalter = (Group) o; + + assertThat(loadedWalter.getId().getUsername()).isEqualTo(walter.getId().getUsername()); + assertThat(loadedWalter.getAge()).isEqualTo(42); + }) + .assertNext(o -> { + Group loadedMike = (Group) o; + + assertThat(loadedMike).isNotNull(); + }) + .verifyComplete(); + + } + + @Test // DATACASS-574 + public void shouldInsertCollectionOfEntities() { + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono loadedMono = batchOperations + .insert(Arrays.asList(walter, mike)) + .execute() + .then(template.selectOneById(walter.getId(), Group.class)); + + StepVerifier.create(loadedMono) + .assertNext(loaded -> assertThat(loaded.getId().getUsername()).isEqualTo(walter.getId().getUsername())) + .verifyComplete(); + } + + @Test // DATACASS-443 + public void shouldInsertCollectionOfEntitiesWithTtl() { + + walter.setEmail("walter@white.com"); + mike.setEmail("mike@sauls.com"); + + int ttl = 30; + WriteOptions options = WriteOptions.builder().ttl(30).build(); + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono resultSet = batchOperations + .insert(Arrays.asList(walter, mike), options) + .execute() + .then(template.getReactiveCqlOperations() + .queryForResultSet("SELECT TTL(email) FROM group;")); + + StepVerifier.create(resultSet.flatMapMany(ReactiveResultSet::availableRows)) + .assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)) + .assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)) + .verifyComplete(); + } + + + @Test // DATACASS-443 + public void shouldInsertMonoCollectionOfEntitiesWithTtl() { + + walter.setEmail("walter@white.com"); + mike.setEmail("mike@sauls.com"); + + int ttl = 30; + WriteOptions options = WriteOptions.builder().ttl(30).build(); + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono resultSet = batchOperations + .insert(Mono.just(Arrays.asList(walter, mike)), options) + .execute() + .then(template.getReactiveCqlOperations() + .queryForResultSet("SELECT TTL(email) FROM group;")); + + StepVerifier.create(resultSet.flatMapMany(ReactiveResultSet::availableRows)) + .assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)) + .assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)) + .verifyComplete(); + } + + @Test // DATACASS-574 + public void shouldUpdateEntities() { + + walter.setEmail("walter@white.com"); + mike.setEmail("mike@sauls.com"); + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono loadedMono = batchOperations + .update(walter) + .update(mike) + .execute() + .then(template.selectOneById(walter.getId(), Group.class)); + + + StepVerifier.create(loadedMono) + .assertNext(loaded -> assertThat(loaded.getEmail()).isEqualTo(walter.getEmail())) + .verifyComplete(); + } + + @Test // DATACASS-574 + public void shouldUpdateMonoEntities() { + + walter.setEmail("walter@white.com"); + mike.setEmail("mike@sauls.com"); + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono loadedMono = batchOperations + .update(walter) + .update(Mono.just(Arrays.asList(mike))) + .execute() + .then(template.selectOneById(walter.getId(), Group.class)); + + + StepVerifier.create(loadedMono) + .assertNext(loaded -> assertThat(loaded.getEmail()).isEqualTo(walter.getEmail())) + .verifyComplete(); + } + + @Test // DATACASS-574 + public void shouldUpdateCollectionOfEntities() { + + walter.setEmail("walter@white.com"); + mike.setEmail("mike@sauls.com"); + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono loadedMono = batchOperations + .update(Arrays.asList(walter, mike)) + .execute() + .then(template.selectOneById(walter.getId(), Group.class)); + + StepVerifier.create(loadedMono) + .assertNext(loaded -> assertThat(loaded.getEmail()).isEqualTo(walter.getEmail())) + .verifyComplete(); + } + + @Test // DATACASS-443 + public void shouldUpdateCollectionOfEntitiesWithTtl() { + + walter.setEmail("walter@white.com"); + mike.setEmail("mike@sauls.com"); + + int ttl = 30; + WriteOptions options = WriteOptions.builder().ttl(ttl).build(); + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono resultSet = batchOperations + .update(Arrays.asList(walter, mike), options) + .execute() + .then(template.getReactiveCqlOperations() + .queryForResultSet("SELECT TTL(email) FROM group;")); + + + StepVerifier.create(resultSet.flatMapMany(ReactiveResultSet::availableRows)) + .assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)) + .assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)) + .verifyComplete(); + } + + @Test // DATACASS-443 + public void shouldUpdateMonoCollectionOfEntitiesWithTtl() { + + walter.setEmail("walter@white.com"); + mike.setEmail("mike@sauls.com"); + + int ttl = 30; + WriteOptions options = WriteOptions.builder().ttl(ttl).build(); + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono resultSet = batchOperations + .update(Arrays.asList(walter), options) + .update(Mono.just(Arrays.asList(mike)), options) + .execute() + .then(template.getReactiveCqlOperations() + .queryForResultSet("SELECT TTL(email) FROM group;")); + + + StepVerifier.create(resultSet.flatMapMany(ReactiveResultSet::availableRows)) + .assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)) + .assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)) + .verifyComplete(); + } + + @Test // DATACASS-574 + public void shouldUpdatesCollectionOfEntities() { + + FlatGroup walter = new FlatGroup("users", "0x1", "walter"); + FlatGroup mike = new FlatGroup("users", "0x1", "mike"); + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono loadedMono = + template.insert(walter) + .then(template.insert(mike)) + .then(Mono.fromRunnable(() -> { + walter.setEmail("walter@white.com"); + mike.setEmail("mike@sauls.com"); + })) + .then(Mono.defer(() -> batchOperations.update(Arrays.asList(walter, mike)) + .execute())) + .then(template.selectOneById(walter, FlatGroup.class)); + + StepVerifier.create(loadedMono) + .assertNext(loaded -> assertThat(loaded.getEmail()).isEqualTo(walter.getEmail())) + .verifyComplete(); + } + + @Test // DATACASS-574 + public void shouldUpdatesMonoCollectionOfEntities() { + + FlatGroup walter = new FlatGroup("users", "0x1", "walter"); + FlatGroup mike = new FlatGroup("users", "0x1", "mike"); + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono loadedMono = + template.insert(walter) + .then(template.insert(mike)) + .then(Mono.fromRunnable(() -> { + walter.setEmail("walter@white.com"); + mike.setEmail("mike@sauls.com"); + })) + .then(Mono.defer(() -> batchOperations.update(Arrays.asList(walter)) + .update(Mono.just(Arrays.asList(mike))) + .execute())) + .then(template.selectOneById(walter, FlatGroup.class)); + + StepVerifier.create(loadedMono) + .assertNext(loaded -> assertThat(loaded.getEmail()).isEqualTo(walter.getEmail())) + .verifyComplete(); + } + + @Test // DATACASS-574 + public void shouldDeleteEntities() { + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + + Mono loadedMono = batchOperations + .delete(walter) + .delete(mike) + .execute() + .then(template.selectOneById(walter.getId(), Group.class)); + + StepVerifier.create(loadedMono) + .verifyComplete(); + } + + @Test // DATACASS-574 + public void shouldDeleteCollectionOfEntities() { + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + + Mono loadedMono = batchOperations + .delete(Arrays.asList(walter, mike)) + .execute() + .then(template.selectOneById(walter.getId(), Group.class)); + + StepVerifier.create(loadedMono) + .verifyComplete(); + } + + @Test // DATACASS-574 + public void shouldDeleteMonoCollectionOfEntities() { + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + + Mono loadedMono = batchOperations + .delete(Mono.just(Arrays.asList(walter, mike))) + .execute() + .then(template.selectOneById(walter.getId(), Group.class)); + + StepVerifier.create(loadedMono) + .verifyComplete(); + } + + @Test // DATACASS-574 + public void shouldApplyTimestampToAllEntities() { + + walter.setEmail("walter@white.com"); + mike.setEmail("mike@sauls.com"); + + long timestamp = (System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1)) * 1000; + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Mono resultSet = batchOperations + .insert(walter) + .insert(mike) + .withTimestamp(timestamp) + .execute() + .then(template.getReactiveCqlOperations() + .queryForResultSet("SELECT writetime(email) FROM group;")); + + StepVerifier.create(resultSet.flatMapMany(ReactiveResultSet::availableRows)) + .assertNext(row -> assertThat(row.getLong(0)).isEqualTo(timestamp)) + .assertNext(row -> assertThat(row.getLong(0)).isEqualTo(timestamp)) + .verifyComplete(); + } + + @Test // DATACASS-574 + public void shouldNotExecuteTwice() { + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + StepVerifier.create( + batchOperations.insert(walter) + .execute() + .then(Mono.fromRunnable(batchOperations::execute)) + ).verifyError(IllegalStateException.class); + } + + @Test // DATACASS-574 + public void shouldNotAllowModificationAfterExecution() { + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + StepVerifier.create( + batchOperations.insert(walter) + .execute() + .then(Mono.fromRunnable(() -> batchOperations.update(new Group()))) + ).verifyError(IllegalStateException.class); + } + + @Test // DATACASS-574 + public void shouldNotAllowModificationAfterExecutionMonoCase() { + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + StepVerifier.create( + batchOperations.insert(Mono.just(Arrays.asList(walter))) + .execute() + .then(Mono.fromRunnable(() -> batchOperations.update(new Group()))) + ).verifyError(IllegalStateException.class); + } + + @Test // DATACASS-574 + public void shouldSupportMultithreadedMerge() { + + ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template); + Random random = new Random(); + + for (int i = 0; i < 100; i++) { + batchOperations.insert(Mono.just(Arrays.asList( + new Group(new GroupKey("users", "0x1", "walter" + random.longs())), + new Group(new GroupKey("users", "0x1", "walter" + random.longs())), + new Group(new GroupKey("users", "0x1", "walter" + random.longs())), + new Group(new GroupKey("users", "0x1", "walter" + random.longs())) + )).publishOn(Schedulers.elastic())); + } + + StepVerifier.create(batchOperations.execute() + .then(template.getReactiveCqlOperations() + .queryForResultSet("SELECT TTL(email) FROM group;")) + .flatMapMany(ReactiveResultSet::availableRows)) + .expectNextCount(402) + .verifyComplete(); + } +}