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

DATACASS-474 - Remove reactive Template API methods accepting Publisher of entities.

Browse Source 
Removing insert(…), update(…) and delete(…) methods accepting Publisher because these methods hide that Cassandra does not support multi-row writes. We don't want to set expectations to support a streaming feature that isn't natively supported.
This commit is contained in:
Mark Paluch
2017-07-05 15:03:19 +02:00
committed by John Blum
parent 60eeae9d31
commit 2f0f87e9c2
3 changed files with 6 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,7 +18,6 @@ package org.springframework.data.cassandra.core;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import org.springframework.data.cassandra.core.convert.CassandraConverter;
import org.springframework.data.cassandra.core.query.Query;
@@ -189,25 +188,6 @@ public interface ReactiveCassandraOperations {
*/
<T> Mono<T> insert(T entity, InsertOptions options) throws DataAccessException;
/**
* Insert the given entities and emit the entity if the insert was applied.
*
* @param entities The entities to insert, must not be {@literal null}.
* @return the inserted entities.
* @throws DataAccessException if there is any problem issuing the execution.
*/
<T> Flux<T> insert(Publisher<? extends T> entities) throws DataAccessException;
/**
* Insert the given entities applying {@link WriteOptions} and emit the entity if the insert was applied.
*
* @param entities The entities to insert, must not be {@literal null}.
* @param options may be {@literal null}.
* @return the inserted entities. Does not emit items for which the {@code INSERT} operation was not applied.
* @throws DataAccessException if there is any problem issuing the execution.
*/
<T> Flux<T> insert(Publisher<? extends T> entities, InsertOptions options) throws DataAccessException;
/**
* Update the given entity and emit the entity if the update was applied.
*
@@ -227,25 +207,6 @@ public interface ReactiveCassandraOperations {
*/
<T> Mono<T> update(T entity, UpdateOptions options) throws DataAccessException;
/**
* Update the given entities and emit the entity if the update was applied.
*
* @param entities The entities to update, must not be {@literal null}.
* @return the updated entities.
* @throws DataAccessException if there is any problem issuing the execution.
*/
<T> Flux<T> update(Publisher<? extends T> entities) throws DataAccessException;
/**
* Update the given entities applying {@link WriteOptions} and emit the entity if the update was applied.
*
* @param entities The entities to update.
* @param options may be {@literal null}.
* @return the updated entities. Does not emit items for which the {@code UPDATE} operation was not applied.
* @throws DataAccessException if there is any problem issuing the execution.
*/
<T> Flux<T> update(Publisher<? extends T> entities, UpdateOptions options) throws DataAccessException;
/**
* Remove the given object from the table by id.
*
@@ -275,25 +236,6 @@ public interface ReactiveCassandraOperations {
*/
<T> Mono<T> delete(T entity, QueryOptions options) throws DataAccessException;
/**
* Delete the given entities and emit the entity if the delete was applied.
*
* @param entities must not be {@literal null}.
* @return the deleted entities.
* @throws DataAccessException if there is any problem issuing the execution.
*/
<T> Flux<T> delete(Publisher<? extends T> entities) throws DataAccessException;
/**
* Delete the given entities applying {@link QueryOptions} and emit the entity if the delete was applied.
*
* @param entities must not be {@literal null}.
* @param options may be {@literal null}.
* @return the deleted entities.
* @throws DataAccessException if there is any problem issuing the execution.
*/
<T> Flux<T> delete(Publisher<? extends T> entities, QueryOptions options) throws DataAccessException;
/**
* Execute a {@code TRUNCATE} query to remove all entities of a given class.
*

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

@@ -405,27 +405,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
return getReactiveCqlOperations().execute(new InsertCallback()).next();
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#insert(org.reactivestreams.Publisher)
*/
@Override
public <T> Flux<T> insert(Publisher<? extends T> entities) {
return insert(entities, null);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#insert(org.reactivestreams.Publisher, org.springframework.data.cassandra.core.InsertOptions)
*/
@Override
public <T> Flux<T> insert(Publisher<? extends T> entities, InsertOptions options) {
Assert.notNull(entities, "Entity publisher must not be null");
return Flux.from(entities).flatMap(entity -> insert(entity, options));
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#update(java.lang.Object)
@@ -463,27 +442,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
return getReactiveCqlOperations().execute(new UpdateCallback()).next();
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#update(org.reactivestreams.Publisher)
*/
@Override
public <T> Flux<T> update(Publisher<? extends T> entities) {
return update(entities, null);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#update(org.reactivestreams.Publisher, org.springframework.data.cassandra.core.UpdateOptions)
*/
@Override
public <T> Flux<T> update(Publisher<? extends T> entities, UpdateOptions options) {
Assert.notNull(entities, "Entity publisher must not be null");
return Flux.from(entities).flatMap(entity -> update(entity, options));
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#deleteById(java.lang.Object, java.lang.Class)
@@ -540,27 +498,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
return getReactiveCqlOperations().execute(new DeleteCallback()).next();
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#delete(org.reactivestreams.Publisher)
*/
@Override
public <T> Flux<T> delete(Publisher<? extends T> entities) {
return delete(entities, null);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#delete(org.reactivestreams.Publisher, org.springframework.data.cql.core.QueryOptions)
*/
@Override
public <T> Flux<T> delete(Publisher<? extends T> entities, QueryOptions options) {
Assert.notNull(entities, "Entity publisher must not be null");
return Flux.from(entities).flatMap(entity -> delete(entity, options));
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#truncate(java.lang.Class)

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

@@ -95,7 +95,8 @@ 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 -> operations.getReactiveCqlOperations().execute(createFullInsert(entity)).map(it -> entity));
return Flux.from(entityStream)
.flatMap(entity -> operations.getReactiveCqlOperations().execute(createFullInsert(entity)).map(it -> entity));
}
/* (non-Javadoc)
@@ -117,7 +118,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entities, "The given Iterable of entities must not be null");
return operations.insert(Flux.fromIterable(entities));
return Flux.fromIterable(entities).flatMap(operations::insert);
}
/* (non-Javadoc)
@@ -128,7 +129,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entityStream, "The given Publisher of entities must not be null");
return operations.insert(entityStream);
return Flux.from(entityStream).flatMap(operations::insert);
}
/* (non-Javadoc)
@@ -256,7 +257,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entities, "The given Iterable of entities must not be null");
return operations.delete(Flux.fromIterable(entities)).then();
return Flux.fromIterable(entities).flatMap(operations::delete).then();
}
/* (non-Javadoc)
@@ -267,7 +268,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entityStream, "The given Publisher of entities must not be null");
return operations.delete(entityStream).then();
return Flux.from(entityStream).flatMap(operations::delete).then();
}
/* (non-Javadoc)