Added Batch Send Implementation

Author: joviegas

services/sqs/pom.xml

@@ -106,6 +106,11 @@
             <artifactId>assertj-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
 
     </dependencies>
 </project>

services/sqs/src/main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/DefaultSqsAsyncBatchManager.java

@@ -15,6 +15,7 @@
 
 package software.amazon.awssdk.services.sqs.internal.batchmanager;
 
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ScheduledExecutorService;
 import software.amazon.awssdk.annotations.SdkInternalApi;
@@ -39,8 +40,6 @@
 public final class DefaultSqsAsyncBatchManager implements SqsAsyncBatchManager {
     // TODO : update the validation here while implementing this class in next PR
     private final SqsAsyncClient client;
-    private final  ScheduledExecutorService scheduledExecutor;
-    private final  BatchOverrideConfiguration overrideConfiguration;
 
     private final BatchManager<SendMessageRequest, SendMessageResponse, SendMessageBatchResponse> sendMessageBatchManager;
 
@@ -53,34 +52,68 @@ public final class DefaultSqsAsyncBatchManager implements SqsAsyncBatchManager {
 
     private DefaultSqsAsyncBatchManager(DefaultBuilder builder) {
         this.client = Validate.notNull(builder.client, "client cannot be null");
-        this.scheduledExecutor = Validate.notNull(builder.scheduledExecutor, "scheduledExecutor cannot be null");
-        // TODO : create overrideConfiguration with Default values if null
-        this.overrideConfiguration = builder.overrideConfiguration;
 
-        sendMessageBatchManager = null;
-        deleteMessageBatchManager = null;
-        changeMessageVisibilityBatchManager = null;
+        ScheduledExecutorService scheduledExecutor = builder.scheduledExecutor;
+
+        this.sendMessageBatchManager = BatchManager
+            .requestBatchManagerBuilder(SendMessageRequest.class, SendMessageResponse.class, SendMessageBatchResponse.class)
+            .batchFunction(SqsBatchFunctions.sendMessageBatchAsyncFunction(client))
+            .responseMapper(SqsBatchFunctions.sendMessageResponseMapper())
+            .batchKeyMapper(SqsBatchFunctions.sendMessageBatchKeyMapper())
+            .overrideConfiguration(sendMessageConfig(builder.overrideConfiguration)).scheduledExecutor(scheduledExecutor)
+            .build();
+        this.deleteMessageBatchManager = BatchManager
+            .requestBatchManagerBuilder(DeleteMessageRequest.class, DeleteMessageResponse.class, DeleteMessageBatchResponse.class)
+            .batchFunction(SqsBatchFunctions.deleteMessageBatchAsyncFunction(client))
+            .responseMapper(SqsBatchFunctions.deleteMessageResponseMapper())
+            .batchKeyMapper(SqsBatchFunctions.deleteMessageBatchKeyMapper())
+            .overrideConfiguration(deleteMessageConfig(builder.overrideConfiguration)).scheduledExecutor(scheduledExecutor)
+            .build();
+        this.changeMessageVisibilityBatchManager = BatchManager
+            .requestBatchManagerBuilder(ChangeMessageVisibilityRequest.class, ChangeMessageVisibilityResponse.class,
+                     ChangeMessageVisibilityBatchResponse.class)
+            .batchFunction(SqsBatchFunctions.changeMessageVisibilityBatchAsyncFunction(client))
+            .responseMapper(SqsBatchFunctions.changeMessageVisibilityResponseMapper())
+            .batchKeyMapper(SqsBatchFunctions.changeMessageVisibilityBatchKeyMapper())
+            .overrideConfiguration(changeMessageVisibilityConfig(builder.overrideConfiguration))
+            .scheduledExecutor(scheduledExecutor).build();
         receiveMessageBatchManager = null;
     }
 
+
+    @SdkInternalApi
+    public DefaultSqsAsyncBatchManager(
+        SqsAsyncClient client,
+        BatchManager<SendMessageRequest, SendMessageResponse, SendMessageBatchResponse> sendMessageBatchManager,
+        BatchManager<DeleteMessageRequest, DeleteMessageResponse, DeleteMessageBatchResponse> deleteMessageBatchManager,
+        BatchManager<ChangeMessageVisibilityRequest, ChangeMessageVisibilityResponse,
+            ChangeMessageVisibilityBatchResponse> changeMessageVisibilityBatchManager) {
+        this.sendMessageBatchManager = sendMessageBatchManager;
+        this.deleteMessageBatchManager = deleteMessageBatchManager;
+        this.changeMessageVisibilityBatchManager = changeMessageVisibilityBatchManager;
+        receiveMessageBatchManager = null;
+
+        this.client = client;
+    }
+
     @Override
     public CompletableFuture<SendMessageResponse> sendMessage(SendMessageRequest request) {
-        return SqsAsyncBatchManager.super.sendMessage(request);
+        return sendMessageBatchManager.batchRequest(request);
     }
 
     @Override
     public CompletableFuture<DeleteMessageResponse> deleteMessage(DeleteMessageRequest request) {
-        return SqsAsyncBatchManager.super.deleteMessage(request);
+        return deleteMessageBatchManager.batchRequest(request);
     }
 
     @Override
     public CompletableFuture<ChangeMessageVisibilityResponse> changeMessageVisibility(ChangeMessageVisibilityRequest request) {
-        return SqsAsyncBatchManager.super.changeMessageVisibility(request);
+        return changeMessageVisibilityBatchManager.batchRequest(request);
     }
 
     @Override
     public CompletableFuture<ReceiveMessageResponse> receiveMessage(ReceiveMessageRequest request) {
-        return SqsAsyncBatchManager.super.receiveMessage(request);
+        return receiveMessageBatchManager.batchRequest(request);
     }
 
     public static SqsAsyncBatchManager.Builder builder() {
@@ -89,6 +122,45 @@ public static SqsAsyncBatchManager.Builder builder() {
 
     @Override
     public void close() {
+        sendMessageBatchManager.close();
+        deleteMessageBatchManager.close();
+        changeMessageVisibilityBatchManager.close();
+    }
+
+    private BatchOverrideConfiguration sendMessageConfig(BatchOverrideConfiguration overrideConfiguration) {
+        BatchOverrideConfiguration.Builder config = BatchOverrideConfiguration.builder();
+        if (overrideConfiguration == null) {
+            config.maxBatchItems(10);
+            config.maxBatchOpenInMs(Duration.ofMillis(200));
+        } else {
+            config.maxBatchItems(overrideConfiguration.maxBatchItems().orElse(10));
+            config.maxBatchOpenInMs(overrideConfiguration.maxBatchOpenInMs().orElse(Duration.ofMillis(200)));
+        }
+        return config.build();
+    }
+
+    private BatchOverrideConfiguration deleteMessageConfig(BatchOverrideConfiguration overrideConfiguration) {
+        BatchOverrideConfiguration.Builder config = BatchOverrideConfiguration.builder();
+        if (overrideConfiguration == null) {
+            config.maxBatchItems(10);
+            config.maxBatchOpenInMs(Duration.ofMillis(200));
+        } else {
+            config.maxBatchItems(overrideConfiguration.maxBatchItems().orElse(10));
+            config.maxBatchOpenInMs(overrideConfiguration.maxBatchOpenInMs().orElse(Duration.ofMillis(200)));
+        }
+        return config.build();
+    }
+
+    private BatchOverrideConfiguration changeMessageVisibilityConfig(BatchOverrideConfiguration overrideConfiguration) {
+        BatchOverrideConfiguration.Builder config = BatchOverrideConfiguration.builder();
+        if (overrideConfiguration == null) {
+            config.maxBatchItems(10);
+            config.maxBatchOpenInMs(Duration.ofMillis(200));
+        } else {
+            config.maxBatchItems(overrideConfiguration.maxBatchItems().orElse(10));
+            config.maxBatchOpenInMs(overrideConfiguration.maxBatchOpenInMs().orElse(Duration.ofMillis(200)));
+        }
+        return config.build();
     }
 
     public static final class DefaultBuilder implements SqsAsyncBatchManager.Builder {

services/sqs/src/main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/RequestsBatchManager.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.services.sqs.internal.batchmanager;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.services.sqs.batchmanager.BatchOverrideConfiguration;
+import software.amazon.awssdk.services.sqs.internal.batchmanager.core.BatchAndSend;
+import software.amazon.awssdk.services.sqs.internal.batchmanager.core.BatchKeyMapper;
+import software.amazon.awssdk.services.sqs.internal.batchmanager.core.BatchManager;
+import software.amazon.awssdk.services.sqs.internal.batchmanager.core.BatchResponseMapper;
+
+
+// TODO:  This class will be implemented in future PR , for now just added it to show how we have separate Batch managers for
+// Request And Response
+@SdkInternalApi
+public final class ResponseBatchManager<RequestT, ResponseT, BatchResponseT>
+    implements BatchManager<RequestT, ResponseT, BatchResponseT> {
+
+
+    private ResponseBatchManager(DefaultBuilder<RequestT, ResponseT, BatchResponseT> builder) {
+        // TODO:  Add implementation logic for sending a request in separate PR
+    }
+
+    public static <ResponseT, BatchResponseT, RequestT> Builder<RequestT, ResponseT, BatchResponseT> builder() {
+        return new DefaultBuilder<>();
+
+    }
+
+
+    @Override
+    public CompletableFuture<ResponseT> batchRequest(RequestT request) {
+        // TODO:  Add implementation logic for sending a request in separate PR
+        return null;
+    }
+
+    @Override
+    public void close() {
+        // TODO:  Add implementation logic for sending a request in separate PR
+    }
+
+    public static final class DefaultBuilder<RequestT, ResponseT, BatchResponseT> implements Builder<RequestT, ResponseT,
+        BatchResponseT> {
+
+        public ResponseBatchManager<RequestT, ResponseT,
+            BatchResponseT> build() {
+            return new ResponseBatchManager<>(this);
+        }
+
+        @Override
+        public Builder<RequestT, ResponseT,
+            BatchResponseT> overrideConfiguration(BatchOverrideConfiguration overrideConfiguration) {
+            return this;
+        }
+
+        @Override
+        public Builder<RequestT, ResponseT, BatchResponseT> scheduledExecutor(ScheduledExecutorService scheduledExecutor) {
+            return this;
+        }
+
+        @Override
+        public Builder<RequestT, ResponseT,
+            BatchResponseT> batchFunction(BatchAndSend<RequestT, BatchResponseT> batchFunction) {
+            return this;
+        }
+
+        @Override
+        public Builder<RequestT, ResponseT,
+            BatchResponseT> responseMapper(BatchResponseMapper<BatchResponseT, ResponseT> responseMapper) {
+            return this;
+        }
+
+        @Override
+        public Builder<RequestT, ResponseT, BatchResponseT> batchKeyMapper(BatchKeyMapper<RequestT> batchKeyMapper) {
+            return this;
+        }
+    }
+}

services/sqs/src/main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ResponseBatchManager.java

@@ -21,7 +21,6 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
 import java.util.stream.Collectors;
-import software.amazon.awssdk.annotations.Generated;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
 import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
@@ -53,7 +52,6 @@
 import software.amazon.awssdk.services.sqs.model.SqsException;
 import software.amazon.awssdk.utils.Either;
 
-@Generated("software.amazon.awssdk:codegen")
 @SdkInternalApi
 public final class SqsBatchFunctions {
     private SqsBatchFunctions() {

services/sqs/src/main/java/software/amazon/awssdk/services/sqs/internal/batchmanager/ResponsesBatchManager.java

@@ -18,7 +18,6 @@
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import software.amazon.awssdk.annotations.SdkInternalApi;
-import software.amazon.awssdk.annotations.SdkProtectedApi;
 
 /**
  * Takes a list of identified requests in addition to a batchKey and batches the requests into a batch request.