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

DATACASS-574 - Add reactive batch operations.

Browse Source 
Original pull request: #134.
This commit is contained in:
Oleh Dokuka
2018-07-20 01:03:35 +03:00
committed by Mark Paluch
parent c61683cae5
commit d7a48f2529
5 changed files with 949 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

183
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchOperations.java Normal file
View File

@@ -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.
* <p>
* 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.
* <p>
* 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<WriteResult>} for the bulk operation.
* @throws IllegalStateException if the batch is executed after it was executed already.
*/
Mono<WriteResult> 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<? extends Iterable<?>> 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<? extends Iterable<?>> 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<? extends Iterable<?>> 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<? extends Iterable<?>> 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<? extends Iterable<?>> entities);
}

284
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchTemplate.java Normal file
View File

@@ -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<Mono<Collection<? extends BuiltStatement>>> 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<WriteResult> 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<? extends Iterable<?>> 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<? extends Iterable<?>> 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<? extends BuiltStatement> doInsert(Iterable<?> entities, WriteOptions options) {
ArrayList<Insert> 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<? extends Iterable<?>> 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<? extends Iterable<?>> 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<? extends BuiltStatement> doUpdate(Iterable<?> entities, WriteOptions options) {
ArrayList<Update> 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<? extends Iterable<?>> entities) {
assertNotExecuted();
Assert.notNull(entities, "Entities must not be null");
batchMonos.add(entities.map(this::doDelete));
return this;
}
Collection<? extends BuiltStatement> doDelete(Iterable<?> entities) {
ArrayList<Delete> 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();
}
}

8
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java
View File

@@ -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}.
*

8
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java
View File

@@ -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()
*/

466
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraBatchTemplateIntegrationTests.java Normal file
View File

@@ -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<WriteResult> execution = batchOperations.insert(walter)
.insert(mike)
.execute();
Mono<Group> 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<Group> 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<ReactiveResultSet> 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<ReactiveResultSet> 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<Group> 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<Group> 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<Group> 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<ReactiveResultSet> 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<ReactiveResultSet> 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<FlatGroup> 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<FlatGroup> 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<Group> 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<Group> 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<Group> 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<ReactiveResultSet> 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();
}
}