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

DATACASS-462 - Polishing.

Browse Source 
Update issue references, insert tests data properly and add removeById(Publisher).

Related ticket: DATACMNS-1063.

Related Pull Request: #106
This commit is contained in:
Christoph Strobl
2017-06-13 11:08:31 +02:00
parent 3f5d2589fb
commit 01cd2dfc24
2 changed files with 64 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java
View File

@@ -38,6 +38,7 @@ import com.datastax.driver.core.querybuilder.Select;
* Reactive repository base implementation for Cassandra.
*
* @author Mark Paluch
* @author Christoph Strobl
* @since 2.0
*/
public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassandraRepository<T, ID> {
@@ -94,27 +95,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entityStream, "The given Publisher of entities must not be null");
return Flux.from(entityStream).flatMap(entity -> {
return operations.getReactiveCqlOperations().execute(createFullInsert(entity)).map(it -> entity);
});
}
private <S extends T> Insert createFullInsert(S entity) {
CassandraConverter converter = operations.getConverter();
CassandraPersistentEntity<?> persistentEntity = converter.getMappingContext()
.getRequiredPersistentEntity(entity.getClass());
Map<String, Object> toInsert = new LinkedHashMap<>();
converter.write(entity, toInsert, persistentEntity);
Insert insert = QueryBuilder.insertInto(persistentEntity.getTableName().toCql());
for (Entry<String, Object> entry : toInsert.entrySet()) {
insert.value(entry.getKey(), entry.getValue());
}
return insert;
return Flux.from(entityStream).flatMap(entity -> operations.getReactiveCqlOperations().execute(createFullInsert(entity)).map(it -> entity));
}
/* (non-Javadoc)
@@ -245,6 +226,17 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
return operations.deleteById(id, entityInformation.getJavaType()).then();
}
/* (non-Javadoc)
* @see org.springframework.data.repository.reactive.ReactiveCrudRepository#deleteById(org.reactivestreams.Publisher)
*/
@Override
public Mono<Void> deleteById(Publisher<ID> publisher) {
Assert.notNull(publisher, "The given id must not be null");
return Mono.from(publisher).flatMap(id -> operations.deleteById(id, entityInformation.getJavaType())).then();
}
/* (non-Javadoc)
* @see org.springframework.data.repository.reactive.ReactiveCrudRepository#delete(java.lang.Object)
*/
@@ -285,4 +277,22 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
public Mono<Void> deleteAll() {
return operations.truncate(entityInformation.getJavaType());
}
private <S extends T> Insert createFullInsert(S entity) {
CassandraConverter converter = operations.getConverter();
CassandraPersistentEntity<?> persistentEntity = converter.getMappingContext()
.getRequiredPersistentEntity(entity.getClass());
Map<String, Object> toInsert = new LinkedHashMap<>();
converter.write(entity, toInsert, persistentEntity);
Insert insert = QueryBuilder.insertInto(persistentEntity.getTableName().toCql());
for (Entry<String, Object> entry : toInsert.entrySet()) {
insert.value(entry.getKey(), entry.getValue());
}
return insert;
}
}

36
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java
View File

@@ -44,6 +44,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* Integration tests for {@link SimpleReactiveCassandraRepository}.
*
* @author Mark Paluch
* @author Christoph Strobl
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@@ -127,8 +128,8 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK
StepVerifier.create(repository.existsById(Mono.just(dave.getId()))).expectNext(true).verifyComplete();
}
@Test // DATACASS-335
public void existsByFluxOfIdShouldReturnTrueForExistingObject() {
@Test // DATACASS-462
public void existsByIdUsingFluxShouldReturnTrueForExistingObject() {
insertTestData();
@@ -163,7 +164,7 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK
}
@Test // DATACASS-462
public void findByIdByFluxOfIdShouldReturnTrueForExistingObject() {
public void findByIdUsingFluxShouldReturnTrueForExistingObject() {
insertTestData();
@@ -321,14 +322,39 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK
@Test // DATACASS-335
public void deleteByIdShouldRemoveEntity() {
insertTestData();
StepVerifier.create(repository.deleteById(dave.getId())).verifyComplete();
StepVerifier.create(repository.findById(dave.getId())).expectNextCount(0).verifyComplete();
}
@Test // DATACASS-462
public void deleteByIdUsingMonoShouldRemoveEntity() {
insertTestData();
StepVerifier.create(repository.deleteById(Mono.just(dave.getId()))).verifyComplete();
StepVerifier.create(repository.existsById(dave.getId())).expectNext(false).verifyComplete();
}
@Test // DATACASS-462
public void deleteByIdUsingFluxShouldRemoveFirstEntity() {
insertTestData();
StepVerifier.create(repository.deleteById(Flux.just(dave.getId(), oliver.getId()))).verifyComplete();
StepVerifier.create(repository.existsById(dave.getId())).expectNext(false).verifyComplete();
StepVerifier.create(repository.existsById(oliver.getId())).expectNext(true).verifyComplete();
}
@Test // DATACASS-335
public void deleteShouldRemoveEntity() {
insertTestData();
StepVerifier.create(repository.delete(dave)).verifyComplete();
StepVerifier.create(repository.findById(dave.getId())).expectNextCount(0).verifyComplete();
@@ -337,6 +363,8 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK
@Test // DATACASS-335
public void deleteIterableOfEntitiesShouldRemoveEntities() {
insertTestData();
StepVerifier.create(repository.deleteAll(Arrays.asList(dave, boyd))).verifyComplete();
StepVerifier.create(repository.findById(boyd.getId())).expectNextCount(0).verifyComplete();
@@ -345,6 +373,8 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK
@Test // DATACASS-335
public void deletePublisherOfEntitiesShouldRemoveEntities() {
insertTestData();
StepVerifier.create(repository.deleteAll(Flux.just(dave, boyd))).verifyComplete();
StepVerifier.create(repository.findById(boyd.getId())).expectNextCount(0).verifyComplete();