Added overloads to saveAll and deleteAll APIs with both Publisher and Iterable (Azure#37532)

* Updated params of reactive bulk API to Iterable:

* Added additional APIs to match saveAll and deleteAll APIs from spring framework

* Removed unused imports

Author: kushagraThapar

eng/code-quality-reports/src/main/resources/revapi/revapi.json

@@ -109,6 +109,16 @@
           "new": "method <S extends T, T> reactor.core.publisher.Flux<S> com.azure.spring.data.cosmos.core.ReactiveCosmosOperations::insertAll(com.azure.spring.data.cosmos.repository.support.CosmosEntityInformation<T, ?>, reactor.core.publisher.Flux<S>)",
           "justification": "Spring interfaces are allowed to add methods."
         },
+        {
+          "code": "java.method.addedToInterface",
+          "new": "method <S extends T, T> reactor.core.publisher.Mono<java.lang.Void> com.azure.spring.data.cosmos.core.ReactiveCosmosOperations::deleteEntities(com.azure.spring.data.cosmos.repository.support.CosmosEntityInformation<T, ?>, java.lang.Iterable<S>)",
+          "justification": "Spring interfaces are allowed to add methods."
+        },
+        {
+          "code": "java.method.addedToInterface",
+          "new": "method <S extends T, T> reactor.core.publisher.Flux<S> com.azure.spring.data.cosmos.core.ReactiveCosmosOperations::insertAll(com.azure.spring.data.cosmos.repository.support.CosmosEntityInformation<T, ?>, java.lang.Iterable<S>)",
+          "justification": "Spring interfaces are allowed to add methods."
+        },
         {
           "regex": true,
           "code": "java\\.annotation\\.(added|attributeValueChanged)",

sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/ReactiveCosmosOperations.java

@@ -147,6 +147,17 @@ Mono<CosmosContainerProperties> replaceContainerProperties(String containerName,
      */
     <T> Mono<T> insert(String containerName, T objectToSave);
 
+    /**
+     * Insert all items with bulk.
+     *
+     * @param entityInformation must not be {@literal null}
+     * @param entities must not be {@literal null}
+     * @param <T> type class of domain type
+     * @param <S> type class of domain type
+     * @return Flux of result
+     */
+    <S extends T, T> Flux<S> insertAll(CosmosEntityInformation<T, ?> entityInformation, Iterable<S> entities);
+
     /**
      * Insert all items with bulk.
      *
@@ -220,6 +231,17 @@ Mono<CosmosContainerProperties> replaceContainerProperties(String containerName,
      */
     <T> Mono<Void> deleteEntity(String containerName, T entity);
 
+    /**
+     * Delete all items with bulk.
+     *
+     * @param entityInformation must not be {@literal null}
+     * @param entities must not be {@literal null}
+     * @param <T> type class of domain type
+     * @param <S> type class of domain type
+     * @return void Mono
+     */
+    <S extends T, T> Mono<Void> deleteEntities(CosmosEntityInformation<T, ?> entityInformation, Iterable<S> entities);
+
     /**
      * Delete all items with bulk.
      *

sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/ReactiveCosmosTemplate.java

@@ -473,7 +473,20 @@ public <T> Mono<T> insert(String containerName, T objectToSave) {
      * Insert all items with bulk.
      *
      * @param entityInformation the CosmosEntityInformation
-     * @param entities the Iterable entities to be deleted
+     * @param entities the Iterable entities to be inserted
+     * @param <T> type class of domain type
+     * @param <S> type class of domain type
+     * @return Flux of result
+     */
+    public <S extends T, T> Flux<S> insertAll(CosmosEntityInformation<T, ?> entityInformation, Iterable<S> entities) {
+        return insertAll(entityInformation, Flux.fromIterable(entities));
+    }
+
+    /**
+     * Insert all items with bulk.
+     *
+     * @param entityInformation the CosmosEntityInformation
+     * @param entities the Flux of entities to be inserted
      * @param <T> type class of domain type
      * @param <S> type class of domain type
      * @return Flux of result
@@ -498,19 +511,19 @@ public <S extends T, T> Flux<S> insertAll(CosmosEntityInformation<T, ?> entityIn
         cosmosBulkExecutionOptions.setInitialMicroBatchSize(1);
 
         return (Flux<S>) this.getCosmosAsyncClient()
-            .getDatabase(this.getDatabaseName())
-            .getContainer(containerName)
-            .executeBulkOperations(cosmosItemOperationsFlux, cosmosBulkExecutionOptions)
-            .publishOn(Schedulers.parallel())
-            .onErrorResume(throwable ->
-                CosmosExceptionUtils.exceptionHandler("Failed to insert item(s)", throwable,
-                    this.responseDiagnosticsProcessor))
-            .flatMap(r -> {
-                CosmosUtils.fillAndProcessResponseDiagnostics(this.responseDiagnosticsProcessor,
-                    r.getResponse().getCosmosDiagnostics(), null);
-                JsonNode responseItem = r.getResponse().getItem(JsonNode.class);
-                return responseItem != null ? Flux.just(toDomainObject(domainType, responseItem)) : Flux.empty();
-            });
+                             .getDatabase(this.getDatabaseName())
+                             .getContainer(containerName)
+                             .executeBulkOperations(cosmosItemOperationsFlux, cosmosBulkExecutionOptions)
+                             .publishOn(Schedulers.parallel())
+                             .onErrorResume(throwable ->
+                                 CosmosExceptionUtils.exceptionHandler("Failed to insert item(s)", throwable,
+                                     this.responseDiagnosticsProcessor))
+                             .flatMap(r -> {
+                                 CosmosUtils.fillAndProcessResponseDiagnostics(this.responseDiagnosticsProcessor,
+                                     r.getResponse().getCosmosDiagnostics(), null);
+                                 JsonNode responseItem = r.getResponse().getItem(JsonNode.class);
+                                 return responseItem != null ? Flux.just(toDomainObject(domainType, responseItem)) : Flux.empty();
+                             });
     }
 
     /**
@@ -674,6 +687,19 @@ public <T> Mono<Void> deleteEntity(String containerName, T entity) {
         return deleteItem(originalItem, containerName, domainType).then();
     }
 
+    /**
+     * Delete all items with bulk.
+     *
+     * @param entityInformation the CosmosEntityInformation
+     * @param entities the Iterable entities to be deleted
+     * @param <T> type class of domain type
+     * @param <S> type class of domain type
+     * @return void Mono
+     */
+    public <S extends T, T> Mono<Void> deleteEntities(CosmosEntityInformation<T, ?> entityInformation, Iterable<S> entities) {
+        return deleteEntities(entityInformation, Flux.fromIterable(entities));
+    }
+
     /**
      * Delete all items with bulk.
      *
@@ -704,13 +730,13 @@ public <S extends T, T> Mono<Void> deleteEntities(CosmosEntityInformation<T, ?>
         cosmosBulkExecutionOptions.setInitialMicroBatchSize(1);
 
         return this.getCosmosAsyncClient()
-            .getDatabase(this.getDatabaseName())
-            .getContainer(containerName)
-            .executeBulkOperations(cosmosItemOperationFlux, cosmosBulkExecutionOptions)
-            .publishOn(Schedulers.parallel())
-            .onErrorResume(throwable ->
-                CosmosExceptionUtils.exceptionHandler("Failed to delete item(s)", throwable,
-                    this.responseDiagnosticsProcessor)).then();
+                   .getDatabase(this.getDatabaseName())
+                   .getContainer(containerName)
+                   .executeBulkOperations(cosmosItemOperationFlux, cosmosBulkExecutionOptions)
+                   .publishOn(Schedulers.parallel())
+                   .onErrorResume(throwable ->
+                       CosmosExceptionUtils.exceptionHandler("Failed to delete item(s)", throwable,
+                           this.responseDiagnosticsProcessor)).then();
     }
 
     /**

sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/repository/support/SimpleReactiveCosmosRepository.java

@@ -130,7 +130,7 @@ public <S extends T> Flux<S> saveAll(Iterable<S> entities) {
         Assert.notNull(entities, "The given Iterable of entities must not be null!");
 
         if (entityInformation.getPartitionKeyFieldName() != null) {
-            return cosmosOperations.insertAll(this.entityInformation, Flux.fromIterable(entities));
+            return cosmosOperations.insertAll(this.entityInformation, entities);
         } else {
             return Flux.fromIterable(entities).flatMap(this::save);
         }
@@ -253,7 +253,7 @@ public Mono<Void> deleteAll(Iterable<? extends T> entities) {
         Assert.notNull(entities, "The given Iterable of entities must not be null!");
 
         if (entityInformation.getPartitionKeyFieldName() != null) {
-            return cosmosOperations.deleteEntities(this.entityInformation, Flux.fromIterable(entities));
+            return cosmosOperations.deleteEntities(this.entityInformation, entities);
         } else {
             return Flux.fromIterable(entities).flatMap(this::delete).then();
         }
@@ -267,9 +267,9 @@ public Mono<Void> deleteAll(Publisher<? extends T> entityStream) {
         if (entityInformation.getPartitionKeyFieldName() != null) {
             return cosmosOperations.deleteEntities(this.entityInformation, Flux.from(entityStream));
         } else {
-            return Flux.from(entityStream)//
-                .map(entityInformation::getRequiredId)//
-                .flatMap(this::deleteById)//
+            return Flux.from(entityStream)
+                .map(entityInformation::getRequiredId)
+                .flatMap(this::deleteById)
                 .then();
         }
     }