From 2f0f87e9c263a332b5c8e8f049383366131328c0 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 5 Jul 2017 15:03:19 +0200 Subject: [PATCH] DATACASS-474 - Remove reactive Template API methods accepting Publisher of entities. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../core/ReactiveCassandraOperations.java | 58 ----------------- .../core/ReactiveCassandraTemplate.java | 63 ------------------- .../SimpleReactiveCassandraRepository.java | 11 ++-- 3 files changed, 6 insertions(+), 126 deletions(-) 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 0c430cb91..ab1fc0cd3 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 @@ -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 { */ Mono 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. - */ - Flux insert(Publisher 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. - */ - Flux insert(Publisher 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 { */ Mono 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. - */ - Flux update(Publisher 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. - */ - Flux update(Publisher entities, UpdateOptions options) throws DataAccessException; - /** * Remove the given object from the table by id. * @@ -275,25 +236,6 @@ public interface ReactiveCassandraOperations { */ Mono 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. - */ - Flux delete(Publisher 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. - */ - Flux delete(Publisher entities, QueryOptions options) throws DataAccessException; - /** * Execute a {@code TRUNCATE} query to remove all entities of a given class. * 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 8eda48a98..d4c5d8e50 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 @@ -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 Flux insert(Publisher 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 Flux insert(Publisher 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 Flux update(Publisher 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 Flux update(Publisher 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 Flux delete(Publisher 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 Flux delete(Publisher 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) diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java index 46ae61b50..c9a4bab0e 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java @@ -95,7 +95,8 @@ public class SimpleReactiveCassandraRepository 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 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 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 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 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)