Skip to content

Commit 3f763f6

Browse files
leahzhao-sqsleahzhao-sqs
authored
Merge pull request #178 from stefan-sq/stefan/fix-v2-move-message-task
Delegate ReDrive APIs to the underlying SQS client
2 parents 38ac70d + df4d896 commit 3f763f6

4 files changed

Lines changed: 170 additions & 0 deletions

File tree

‎src/main/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedAsyncClientBase.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,8 @@
44
import software.amazon.awssdk.services.sqs.SqsAsyncClient;
55
import software.amazon.awssdk.services.sqs.model.AddPermissionRequest;
66
import software.amazon.awssdk.services.sqs.model.AddPermissionResponse;
7+
import software.amazon.awssdk.services.sqs.model.CancelMessageMoveTaskRequest;
8+
import software.amazon.awssdk.services.sqs.model.CancelMessageMoveTaskResponse;
79
import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchRequest;
810
import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchResponse;
911
import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityRequest;
@@ -22,6 +24,8 @@
2224
import software.amazon.awssdk.services.sqs.model.GetQueueUrlResponse;
2325
import software.amazon.awssdk.services.sqs.model.ListDeadLetterSourceQueuesRequest;
2426
import software.amazon.awssdk.services.sqs.model.ListDeadLetterSourceQueuesResponse;
27+
import software.amazon.awssdk.services.sqs.model.ListMessageMoveTasksRequest;
28+
import software.amazon.awssdk.services.sqs.model.ListMessageMoveTasksResponse;
2529
import software.amazon.awssdk.services.sqs.model.ListQueueTagsRequest;
2630
import software.amazon.awssdk.services.sqs.model.ListQueueTagsResponse;
2731
import software.amazon.awssdk.services.sqs.model.ListQueuesRequest;
@@ -38,6 +42,8 @@
3842
import software.amazon.awssdk.services.sqs.model.SendMessageResponse;
3943
import software.amazon.awssdk.services.sqs.model.SetQueueAttributesRequest;
4044
import software.amazon.awssdk.services.sqs.model.SetQueueAttributesResponse;
45+
import software.amazon.awssdk.services.sqs.model.StartMessageMoveTaskRequest;
46+
import software.amazon.awssdk.services.sqs.model.StartMessageMoveTaskResponse;
4147
import software.amazon.awssdk.services.sqs.model.TagQueueRequest;
4248
import software.amazon.awssdk.services.sqs.model.TagQueueResponse;
4349
import software.amazon.awssdk.services.sqs.model.UntagQueueRequest;
@@ -226,6 +232,33 @@ public CompletableFuture<UntagQueueResponse> untagQueue(final UntagQueueRequest
226232
return amazonSqsToBeExtended.untagQueue(untagQueueRequest);
227233
}
228234

235+
/**
236+
* {@inheritDoc}
237+
*/
238+
@Override
239+
public CompletableFuture<StartMessageMoveTaskResponse> startMessageMoveTask(
240+
StartMessageMoveTaskRequest startMessageMoveTaskRequest) {
241+
return amazonSqsToBeExtended.startMessageMoveTask(startMessageMoveTaskRequest);
242+
}
243+
244+
/**
245+
* {@inheritDoc}
246+
*/
247+
@Override
248+
public CompletableFuture<ListMessageMoveTasksResponse> listMessageMoveTasks(
249+
ListMessageMoveTasksRequest listMessageMoveTasksRequest) {
250+
return amazonSqsToBeExtended.listMessageMoveTasks(listMessageMoveTasksRequest);
251+
}
252+
253+
/**
254+
* {@inheritDoc}
255+
*/
256+
@Override
257+
public CompletableFuture<CancelMessageMoveTaskResponse> cancelMessageMoveTask(
258+
CancelMessageMoveTaskRequest cancelMessageMoveTaskRequest) {
259+
return amazonSqsToBeExtended.cancelMessageMoveTask(cancelMessageMoveTaskRequest);
260+
}
261+
229262
/**
230263
* {@inheritDoc}
231264
*/

‎src/main/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedClientBase.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
import software.amazon.awssdk.services.sqs.model.AddPermissionResponse;
2626
import software.amazon.awssdk.services.sqs.model.BatchEntryIdsNotDistinctException;
2727
import software.amazon.awssdk.services.sqs.model.BatchRequestTooLongException;
28+
import software.amazon.awssdk.services.sqs.model.CancelMessageMoveTaskRequest;
29+
import software.amazon.awssdk.services.sqs.model.CancelMessageMoveTaskResponse;
2830
import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchRequest;
2931
import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchResponse;
3032
import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityRequest;
@@ -48,6 +50,8 @@
4850
import software.amazon.awssdk.services.sqs.model.InvalidMessageContentsException;
4951
import software.amazon.awssdk.services.sqs.model.ListDeadLetterSourceQueuesRequest;
5052
import software.amazon.awssdk.services.sqs.model.ListDeadLetterSourceQueuesResponse;
53+
import software.amazon.awssdk.services.sqs.model.ListMessageMoveTasksRequest;
54+
import software.amazon.awssdk.services.sqs.model.ListMessageMoveTasksResponse;
5155
import software.amazon.awssdk.services.sqs.model.ListQueueTagsRequest;
5256
import software.amazon.awssdk.services.sqs.model.ListQueueTagsResponse;
5357
import software.amazon.awssdk.services.sqs.model.ListQueuesRequest;
@@ -72,6 +76,8 @@
7276
import software.amazon.awssdk.services.sqs.model.SetQueueAttributesRequest;
7377
import software.amazon.awssdk.services.sqs.model.SetQueueAttributesResponse;
7478
import software.amazon.awssdk.services.sqs.model.SqsException;
79+
import software.amazon.awssdk.services.sqs.model.StartMessageMoveTaskRequest;
80+
import software.amazon.awssdk.services.sqs.model.StartMessageMoveTaskResponse;
7581
import software.amazon.awssdk.services.sqs.model.TagQueueRequest;
7682
import software.amazon.awssdk.services.sqs.model.TagQueueResponse;
7783
import software.amazon.awssdk.services.sqs.model.TooManyEntriesInBatchRequestException;
@@ -1112,6 +1118,27 @@ public ListQueuesResponse listQueues() throws AwsServiceException, SdkClientExce
11121118
return amazonSqsToBeExtended.untagQueue(untagQueueRequest);
11131119
}
11141120

1121+
/** {@inheritDoc} */
1122+
@Override
1123+
public StartMessageMoveTaskResponse startMessageMoveTask(
1124+
StartMessageMoveTaskRequest startMessageMoveTaskRequest) {
1125+
return amazonSqsToBeExtended.startMessageMoveTask(startMessageMoveTaskRequest);
1126+
}
1127+
1128+
/** {@inheritDoc} */
1129+
@Override
1130+
public ListMessageMoveTasksResponse listMessageMoveTasks(
1131+
ListMessageMoveTasksRequest listMessageMoveTasksRequest) {
1132+
return amazonSqsToBeExtended.listMessageMoveTasks(listMessageMoveTasksRequest);
1133+
}
1134+
1135+
/** {@inheritDoc} */
1136+
@Override
1137+
public CancelMessageMoveTaskResponse cancelMessageMoveTask(
1138+
CancelMessageMoveTaskRequest cancelMessageMoveTaskRequest) {
1139+
return amazonSqsToBeExtended.cancelMessageMoveTask(cancelMessageMoveTaskRequest);
1140+
}
1141+
11151142
@Override
11161143
public String serviceName() {
11171144
return amazonSqsToBeExtended.serviceName();

‎src/test/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedAsyncClientTest.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,11 +58,17 @@
5858
import software.amazon.awssdk.services.sqs.model.MessageAttributeValue;
5959
import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
6060
import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
61+
import software.amazon.awssdk.services.sqs.model.CancelMessageMoveTaskRequest;
62+
import software.amazon.awssdk.services.sqs.model.CancelMessageMoveTaskResponse;
63+
import software.amazon.awssdk.services.sqs.model.ListMessageMoveTasksRequest;
64+
import software.amazon.awssdk.services.sqs.model.ListMessageMoveTasksResponse;
6165
import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequest;
6266
import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequestEntry;
6367
import software.amazon.awssdk.services.sqs.model.SendMessageBatchResponse;
6468
import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
6569
import software.amazon.awssdk.services.sqs.model.SendMessageResponse;
70+
import software.amazon.awssdk.services.sqs.model.StartMessageMoveTaskRequest;
71+
import software.amazon.awssdk.services.sqs.model.StartMessageMoveTaskResponse;
6672
import software.amazon.awssdk.utils.ImmutableMap;
6773
import software.amazon.payloadoffloading.PayloadS3Pointer;
6874
import software.amazon.payloadoffloading.ServerSideEncryptionFactory;
@@ -703,6 +709,55 @@ public void testReceiveMessage_when_ignorePayloadNotFoundIsFalse_then_messageWit
703709
}
704710
}
705711

712+
@Test
713+
public void testStartMessageMoveTaskDelegatesToUnderlyingClient() {
714+
StartMessageMoveTaskRequest request = StartMessageMoveTaskRequest.builder()
715+
.sourceArn("arn:aws:sqs:us-east-1:123456789012:source-dlq")
716+
.build();
717+
StartMessageMoveTaskResponse expectedResponse = StartMessageMoveTaskResponse.builder()
718+
.taskHandle("task-handle")
719+
.build();
720+
when(mockSqsBackend.startMessageMoveTask(isA(StartMessageMoveTaskRequest.class)))
721+
.thenReturn(CompletableFuture.completedFuture(expectedResponse));
722+
723+
StartMessageMoveTaskResponse response = extendedSqsWithDefaultConfig.startMessageMoveTask(request).join();
724+
725+
assertEquals(expectedResponse, response);
726+
verify(mockSqsBackend).startMessageMoveTask(request);
727+
}
728+
729+
@Test
730+
public void testListMessageMoveTasksDelegatesToUnderlyingClient() {
731+
ListMessageMoveTasksRequest request = ListMessageMoveTasksRequest.builder()
732+
.sourceArn("arn:aws:sqs:us-east-1:123456789012:source-dlq")
733+
.build();
734+
ListMessageMoveTasksResponse expectedResponse = ListMessageMoveTasksResponse.builder().build();
735+
when(mockSqsBackend.listMessageMoveTasks(isA(ListMessageMoveTasksRequest.class)))
736+
.thenReturn(CompletableFuture.completedFuture(expectedResponse));
737+
738+
ListMessageMoveTasksResponse response = extendedSqsWithDefaultConfig.listMessageMoveTasks(request).join();
739+
740+
assertEquals(expectedResponse, response);
741+
verify(mockSqsBackend).listMessageMoveTasks(request);
742+
}
743+
744+
@Test
745+
public void testCancelMessageMoveTaskDelegatesToUnderlyingClient() {
746+
CancelMessageMoveTaskRequest request = CancelMessageMoveTaskRequest.builder()
747+
.taskHandle("task-handle")
748+
.build();
749+
CancelMessageMoveTaskResponse expectedResponse = CancelMessageMoveTaskResponse.builder()
750+
.approximateNumberOfMessagesMoved(42L)
751+
.build();
752+
when(mockSqsBackend.cancelMessageMoveTask(isA(CancelMessageMoveTaskRequest.class)))
753+
.thenReturn(CompletableFuture.completedFuture(expectedResponse));
754+
755+
CancelMessageMoveTaskResponse response = extendedSqsWithDefaultConfig.cancelMessageMoveTask(request).join();
756+
757+
assertEquals(expectedResponse, response);
758+
verify(mockSqsBackend).cancelMessageMoveTask(request);
759+
}
760+
706761
private DeleteMessageBatchRequest generateLargeDeleteBatchRequest(List<String> originalReceiptHandles) {
707762
List<DeleteMessageBatchRequestEntry> deleteEntries = IntStream.range(0, originalReceiptHandles.size())
708763
.mapToObj(i -> DeleteMessageBatchRequestEntry.builder()

‎src/test/java/com/amazon/sqs/javamessaging/AmazonSQSExtendedClientTest.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,13 +39,19 @@
3939
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequest;
4040
import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequestEntry;
4141
import software.amazon.awssdk.services.sqs.model.DeleteMessageRequest;
42+
import software.amazon.awssdk.services.sqs.model.CancelMessageMoveTaskRequest;
43+
import software.amazon.awssdk.services.sqs.model.CancelMessageMoveTaskResponse;
44+
import software.amazon.awssdk.services.sqs.model.ListMessageMoveTasksRequest;
45+
import software.amazon.awssdk.services.sqs.model.ListMessageMoveTasksResponse;
4246
import software.amazon.awssdk.services.sqs.model.Message;
4347
import software.amazon.awssdk.services.sqs.model.MessageAttributeValue;
4448
import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
4549
import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
4650
import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequest;
4751
import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequestEntry;
4852
import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
53+
import software.amazon.awssdk.services.sqs.model.StartMessageMoveTaskRequest;
54+
import software.amazon.awssdk.services.sqs.model.StartMessageMoveTaskResponse;
4955
import software.amazon.awssdk.utils.ImmutableMap;
5056
import software.amazon.awssdk.utils.StringInputStream;
5157
import software.amazon.payloadoffloading.PayloadS3Pointer;
@@ -757,6 +763,55 @@ public void testReceiveMessage_when_ignorePayloadNotFoundIsFalse_then_messageWit
757763
}
758764
}
759765

766+
@Test
767+
public void testStartMessageMoveTaskDelegatesToUnderlyingClient() {
768+
StartMessageMoveTaskRequest request = StartMessageMoveTaskRequest.builder()
769+
.sourceArn("arn:aws:sqs:us-east-1:123456789012:source-dlq")
770+
.build();
771+
StartMessageMoveTaskResponse expectedResponse = StartMessageMoveTaskResponse.builder()
772+
.taskHandle("task-handle")
773+
.build();
774+
when(mockSqsBackend.startMessageMoveTask(isA(StartMessageMoveTaskRequest.class)))
775+
.thenReturn(expectedResponse);
776+
777+
StartMessageMoveTaskResponse response = extendedSqsWithDefaultConfig.startMessageMoveTask(request);
778+
779+
assertEquals(expectedResponse, response);
780+
verify(mockSqsBackend).startMessageMoveTask(request);
781+
}
782+
783+
@Test
784+
public void testListMessageMoveTasksDelegatesToUnderlyingClient() {
785+
ListMessageMoveTasksRequest request = ListMessageMoveTasksRequest.builder()
786+
.sourceArn("arn:aws:sqs:us-east-1:123456789012:source-dlq")
787+
.build();
788+
ListMessageMoveTasksResponse expectedResponse = ListMessageMoveTasksResponse.builder().build();
789+
when(mockSqsBackend.listMessageMoveTasks(isA(ListMessageMoveTasksRequest.class)))
790+
.thenReturn(expectedResponse);
791+
792+
ListMessageMoveTasksResponse response = extendedSqsWithDefaultConfig.listMessageMoveTasks(request);
793+
794+
assertEquals(expectedResponse, response);
795+
verify(mockSqsBackend).listMessageMoveTasks(request);
796+
}
797+
798+
@Test
799+
public void testCancelMessageMoveTaskDelegatesToUnderlyingClient() {
800+
CancelMessageMoveTaskRequest request = CancelMessageMoveTaskRequest.builder()
801+
.taskHandle("task-handle")
802+
.build();
803+
CancelMessageMoveTaskResponse expectedResponse = CancelMessageMoveTaskResponse.builder()
804+
.approximateNumberOfMessagesMoved(42L)
805+
.build();
806+
when(mockSqsBackend.cancelMessageMoveTask(isA(CancelMessageMoveTaskRequest.class)))
807+
.thenReturn(expectedResponse);
808+
809+
CancelMessageMoveTaskResponse response = extendedSqsWithDefaultConfig.cancelMessageMoveTask(request);
810+
811+
assertEquals(expectedResponse, response);
812+
verify(mockSqsBackend).cancelMessageMoveTask(request);
813+
}
814+
760815
private DeleteMessageBatchRequest generateLargeDeleteBatchRequest(List<String> originalReceiptHandles) {
761816
List<DeleteMessageBatchRequestEntry> deleteEntries = IntStream.range(0, originalReceiptHandles.size())
762817
.mapToObj(i -> DeleteMessageBatchRequestEntry.builder()

0 commit comments

Comments
 (0)