diff --git a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsConsumer.java b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsConsumer.java index 7024288..3f37fa2 100644 --- a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsConsumer.java +++ b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsConsumer.java @@ -68,11 +68,6 @@ abstract class AbstractAmazonSqsConsumer implements Runnable, Amazon */ private static final Integer KB = 1024; - /** - * Maximum batch size threshold of 1024 KB imposed by Amazon SQS. - */ - private static final Integer BATCH_SIZE_BYTES_THRESHOLD = 1024 * KB; - /** Class logger. */ private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAmazonSqsConsumer.class); @@ -123,7 +118,7 @@ protected AbstractAmazonSqsConsumer( this.queueProperty = Objects.requireNonNull(queueProperty, "queueProperty cannot be null"); this.amazonSqsClient = Objects.requireNonNull(amazonSqsClient, "amazonSqsClient cannot be null"); this.executorService = Objects.requireNonNull(executorService, "executorService cannot be null"); - this.requestEntryInternalFactory = RequestEntryInternalFactory.build(jsonMapper); + requestEntryInternalFactory = RequestEntryInternalFactory.build(jsonMapper); this.pendingRequests = pendingRequests; this.queueRequests = queueRequests; this.publishDecorator = publishDecorator; @@ -265,7 +260,7 @@ private boolean maxBatchSizeReached(final BlockingQueue> request * @return true if the request can be added */ private boolean canAddToBatch(final int batchSizeBytes, final int requestEntriesSize, final RequestEntry request) { - return (batchSizeBytes < BATCH_SIZE_BYTES_THRESHOLD) + return (batchSizeBytes < queueProperty.getMaxMessageSize()) && (requestEntriesSize < queueProperty.getMaxBatchSize()) && Objects.nonNull(request); } @@ -277,7 +272,7 @@ private boolean canAddToBatch(final int batchSizeBytes, final int requestEntries * @return true if the batch is still within the size limit */ private boolean canAddPayload(final int batchSizeBytes) { - return batchSizeBytes <= BATCH_SIZE_BYTES_THRESHOLD; + return batchSizeBytes <= queueProperty.getMaxMessageSize(); } /** @@ -304,8 +299,9 @@ private Optional createBatch(final BlockingQueue> requests) t final Integer messageSize = messageBodySize + messageAttributesSize; - if (messageSize > BATCH_SIZE_BYTES_THRESHOLD) { - throw PoisonRequestEntryException.fromMaximumAllowedMessage("The maximum allowed message size exceeding 1024KB (1,048,576 bytes)."); + if (messageSize > queueProperty.getMaxMessageSize()) { + final String message = String.format("The maximum allowed message size exceeding %dKB (%,d bytes).", queueProperty.getMaxMessageSize() / KB, queueProperty.getMaxMessageSize()); + throw PoisonRequestEntryException.fromMaximumAllowedMessage(message); } if (canAddPayload(batchSizeBytes.get() + messageSize)) { diff --git a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsProducer.java b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsProducer.java index 9d248b5..cd2a85c 100644 --- a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsProducer.java +++ b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsProducer.java @@ -73,11 +73,14 @@ public ListenableFuture send(final Requ } else { final ListenableFutureImpl listenableFutureImpl = new ListenableFutureImpl(Runnable::run); + final String message = String.format("Producer is currently in %s mode; no further messages will be accepted.", state.get().name()); + listenableFutureImpl.fail(ResponseFailEntry.builder() .withCode("000") .withId(requestEntry.getId()) - .withMessage(String.format("Producer is currently in %s mode; no further messages will be accepted.", state.get().name())) + .withMessage(message) .withSenderFault(true) + .withThrowable(new IllegalStateException(message)) .build()); return listenableFutureImpl; diff --git a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractMessageAttributes.java b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractMessageAttributes.java index 5918ea6..e199e17 100644 --- a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractMessageAttributes.java +++ b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/AbstractMessageAttributes.java @@ -17,6 +17,7 @@ package com.amazon.sqs.messaging.lib.core; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -44,21 +45,29 @@ abstract class AbstractMessageAttributes { */ protected static final String BINARY = "Binary"; + protected static final Integer BINARY_SIZE = BINARY.getBytes(StandardCharsets.UTF_8).length; + /** * Data type constant for string message attributes. */ protected static final String STRING = "String"; + protected static final Integer STRING_SIZE = STRING.getBytes(StandardCharsets.UTF_8).length; + /** * Data type constant for number message attributes. */ protected static final String NUMBER = "Number"; + protected static final Integer NUMBER_SIZE = NUMBER.getBytes(StandardCharsets.UTF_8).length; + /** * Data type constant for string array message attributes. */ protected static final String STRING_ARRAY = "String.Array"; + protected static final Integer STRING_ARRAY_SIZE = STRING_ARRAY.getBytes(StandardCharsets.UTF_8).length; + /** * Converts a map of message headers into typed message attributes. * @@ -112,7 +121,7 @@ protected static String stringArray(final List values) { ); } - return "[ " + String.join(", ", collect) + " ]"; + return String.format("[%s]", String.join(",", collect)); } /** diff --git a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/RequestEntryInternalFactory.java b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/RequestEntryInternalFactory.java index 3ca96cd..e1b423f 100644 --- a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/RequestEntryInternalFactory.java +++ b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/core/RequestEntryInternalFactory.java @@ -21,6 +21,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.function.ToIntFunction; import java.util.stream.Collectors; import com.amazon.sqs.messaging.lib.exception.PoisonRequestEntryException; @@ -177,12 +178,15 @@ static class MessageAttributesInternal extends AbstractMessageAttributes> computeSize = clazz -> + NUMBER.concat(".").concat(clazz.getName()).getBytes(StandardCharsets.UTF_8).length; + /** * {@inheritDoc} */ @Override public Integer getEnumMessageAttribute(final Enum value) { - return value.name().getBytes(StandardCharsets.UTF_8).length; + return STRING_SIZE + value.name().getBytes(StandardCharsets.UTF_8).length; } /** @@ -190,7 +194,7 @@ public Integer getEnumMessageAttribute(final Enum value) { */ @Override public Integer getStringMessageAttribute(final String value) { - return value.getBytes(StandardCharsets.UTF_8).length; + return STRING_SIZE + value.getBytes(StandardCharsets.UTF_8).length; } /** @@ -198,7 +202,7 @@ public Integer getStringMessageAttribute(final String value) { */ @Override public Integer getNumberMessageAttribute(final Number value) { - return value.toString().getBytes(StandardCharsets.UTF_8).length; + return value.toString().getBytes(StandardCharsets.UTF_8).length + computeSize.applyAsInt(value.getClass()); } /** @@ -206,7 +210,7 @@ public Integer getNumberMessageAttribute(final Number value) { */ @Override public Integer getBinaryMessageAttribute(final ByteBuffer value) { - return value.remaining(); + return BINARY_SIZE + value.remaining(); } /** @@ -214,7 +218,7 @@ public Integer getBinaryMessageAttribute(final ByteBuffer value) { */ @Override public Integer getStringArrayMessageAttribute(final List values) { - return stringArray(values).getBytes(StandardCharsets.UTF_8).length; + return STRING_ARRAY_SIZE + stringArray(values).getBytes(StandardCharsets.UTF_8).length; } } diff --git a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/exception/PoisonRequestEntryException.java b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/exception/PoisonRequestEntryException.java index 0855200..fdf6eeb 100644 --- a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/exception/PoisonRequestEntryException.java +++ b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/exception/PoisonRequestEntryException.java @@ -1,9 +1,26 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License 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 com.amazon.sqs.messaging.lib.exception; /** * Base exception indicating that a request entry could not be published because - * it is invalid, either exceeding the maximum allowed message size or failing to - * serialize. Concrete subtypes are created via the {@code from*} factory methods. + * it is invalid, either exceeding the maximum allowed message size or failing + * to serialize. Concrete subtypes are created via the {@code from*} factory + * methods. */ public abstract class PoisonRequestEntryException extends Exception { diff --git a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/model/QueueProperty.java b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/model/QueueProperty.java index ac34a65..95b0e7a 100644 --- a/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/model/QueueProperty.java +++ b/amazon-sqs-java-messaging-lib-template/src/main/java/com/amazon/sqs/messaging/lib/model/QueueProperty.java @@ -51,8 +51,16 @@ @AllArgsConstructor(access = AccessLevel.PRIVATE) public class QueueProperty { + private static final int KB = 1024; + private static final long DEFAULT_LINGER = 10L; + private static final int DEFAULT_MESSAGE_SIZE = 256 * KB; + + private static final int MIN_MESSAGE_SIZE = KB; + + private static final int MAX_MESSAGE_SIZE = 1024 * KB; + /** * Whether the queue is a FIFO queue. */ @@ -78,6 +86,11 @@ public class QueueProperty { */ private final int maxBatchSize; + /** + * The maximum message size in bytes. + */ + private final int maxMessageSize; + /** * Validates {@link QueueProperty} instances against the SQS constraints for * pool size, queue URL format, linger, batch size, and FIFO consistency. @@ -110,6 +123,10 @@ public void rules() { .must(greaterThanOrEqual(DEFAULT_LINGER)) .withMessage("'linger' must be greater than or equal to 10 (ten)"); + ruleFor("maxMessageSize", QueueProperty::getMaxMessageSize) + .must(betweenInclusive(MIN_MESSAGE_SIZE, MAX_MESSAGE_SIZE)) + .withMessage("'maxMessageSize' must be in the range of 1Kb (1,024 bytes) to 1024Kb (1,048,576 bytes)"); + ruleFor("maxBatchSize", QueueProperty::getMaxBatchSize) .must(betweenInclusive(1, 10)) .withMessage("'maxBatchSize' must be in the range of 1 (one) to 10 (ten)"); @@ -148,6 +165,14 @@ public static class QueuePropertyBuilder { */ private boolean linger$set; + /** + * Tracks whether {@code maxMessageSize(int)} was explicitly invoked. + * + *

This flag allows applying {@link QueueProperty#DEFAULT_MESSAGE_SIZE} only when + * no explicit value was provided through the builder. + */ + private boolean maxMessageSize$set; + /** * Sets the batching linger time in milliseconds. * @@ -160,6 +185,18 @@ public QueuePropertyBuilder linger(final long linger) { return this; } + /** + * Sets the maximum message size in bytes. + * + * @param maxMessageSize the maximum message size in bytes + * @return this builder + */ + public QueuePropertyBuilder maxMessageSize(final int maxMessageSize) { + this.maxMessageSize = maxMessageSize; + maxMessageSize$set = true; + return this; + } + /** * Builds the {@link QueueProperty}, applying the default linger value if none * was explicitly set and validating the result. @@ -169,8 +206,9 @@ public QueuePropertyBuilder linger(final long linger) { */ public QueueProperty build() { final long linger = linger$set ? this.linger : DEFAULT_LINGER; + final int maxMessageSize = maxMessageSize$set ? this.maxMessageSize : DEFAULT_MESSAGE_SIZE; - final QueueProperty queueProperty = new QueueProperty(fifo, maximumPoolSize, queueUrl, linger, maxBatchSize); + final QueueProperty queueProperty = new QueueProperty(fifo, maximumPoolSize, queueUrl, linger, maxBatchSize, maxMessageSize); final ValidationResult validationResult = QueuePropertyValidator.INSTANCE.validate(queueProperty); diff --git a/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsConsumerTest.java b/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsConsumerTest.java index 755ddb3..f10f8aa 100644 --- a/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsConsumerTest.java +++ b/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/AbstractAmazonSqsConsumerTest.java @@ -40,9 +40,11 @@ import java.util.Collections; import java.util.LinkedList; import java.util.List; +import java.util.Queue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; @@ -115,6 +117,7 @@ void setUp() { when(queueProperty.getLinger()).thenReturn(LINGER_MS); when(queueProperty.getMaxBatchSize()).thenReturn(MAX_BATCH_SIZE); when(queueProperty.isFifo()).thenReturn(false); + when(queueProperty.getMaxMessageSize()).thenReturn(256 * TestableAmazonSqsConsumer.KB); } @Nested @@ -603,7 +606,7 @@ void testCanAddPayloadAllowsEntryExactlyAtSizeThreshold() throws Exception { when(queueProperty.isFifo()).thenReturn(true); context(consumer -> { - final String payloadAtThreshold = buildPayloadOfBytes(TestableAmazonSqsConsumer.batchSizeBytesThreshold()); + final String payloadAtThreshold = buildPayloadOfBytes(queueProperty.getMaxMessageSize()); queueRequests.put(buildRequestEntry(payloadAtThreshold)); await() @@ -619,7 +622,7 @@ void testCanAddPayloadRejectsEntryExceedingSizeThreshold() throws Exception { when(queueProperty.isFifo()).thenReturn(true); context(consumer -> { - final String oversizedPayload = buildPayloadOfBytes(TestableAmazonSqsConsumer.batchSizeBytesThreshold() + 1); + final String oversizedPayload = buildPayloadOfBytes(queueProperty.getMaxMessageSize() + 1); queueRequests.put(buildRequestEntry(oversizedPayload)); await() @@ -636,7 +639,7 @@ void testCanAddPayloadStopsAccumulatingWhenBatchExceedsThreshold() throws Except when(queueProperty.getMaxBatchSize()).thenReturn(10); context(consumer -> { - final int halfThreshold = TestableAmazonSqsConsumer.batchSizeBytesThreshold() / 2; + final int halfThreshold = queueProperty.getMaxMessageSize() / 2; queueRequests.put(buildRequestEntry(buildPayloadOfBytes(halfThreshold))); queueRequests.put(buildRequestEntry(buildPayloadOfBytes(halfThreshold))); queueRequests.put(buildRequestEntry("small-overflow")); @@ -655,7 +658,7 @@ void testCanAddPayloadPublishesFirstEntryAloneWhenItFillsThreshold() throws Exce when(queueProperty.getMaxBatchSize()).thenReturn(10); context(consumer -> { - final int fullThreshold = TestableAmazonSqsConsumer.batchSizeBytesThreshold(); + final int fullThreshold = queueProperty.getMaxMessageSize(); queueRequests.put(buildRequestEntry(buildPayloadOfBytes(fullThreshold))); queueRequests.put(buildRequestEntry("second-entry")); @@ -691,7 +694,7 @@ void testCanAddPayloadSplitsBatchWhenCumulativeSizeExceedsThreshold() throws Exc when(queueProperty.getMaxBatchSize()).thenReturn(10); context(consumer -> { - final int chunkSize = (TestableAmazonSqsConsumer.batchSizeBytesThreshold() / 3) + 1; + final int chunkSize = (queueProperty.getMaxMessageSize() / 3) + 1; queueRequests.put(buildRequestEntry(buildPayloadOfBytes(chunkSize))); queueRequests.put(buildRequestEntry(buildPayloadOfBytes(chunkSize))); queueRequests.put(buildRequestEntry(buildPayloadOfBytes(chunkSize))); @@ -709,7 +712,7 @@ void testCanAddPayloadDoesNotPublishEmptyBatchWhenAllEntriesExceedThreshold() th when(queueProperty.isFifo()).thenReturn(true); context(consumer -> { - final String oversizedPayload = buildPayloadOfBytes(TestableAmazonSqsConsumer.batchSizeBytesThreshold() + 100); + final String oversizedPayload = buildPayloadOfBytes(queueProperty.getMaxMessageSize() + 100); queueRequests.put(buildRequestEntry(oversizedPayload)); await() @@ -729,7 +732,7 @@ void testPoisonRequestEntryRemovesFromPendingRequestsAndFailsListenableFuture() when(queueProperty.isFifo()).thenReturn(true); final String poisonId = "poison-id"; - final String oversizedPayload = buildPayloadOfBytes(TestableAmazonSqsConsumer.batchSizeBytesThreshold() + 100); + final String oversizedPayload = buildPayloadOfBytes(queueProperty.getMaxMessageSize() + 100); final RequestEntry poisonEntry = RequestEntry.builder() .withId(poisonId) .withValue(oversizedPayload) @@ -779,7 +782,7 @@ private void context(final UnaryOperator trackingDecorator, final TryCon static class TestableAmazonSqsConsumer extends AbstractAmazonSqsConsumer implements AutoCloseable { - private static final int BATCH_SIZE_BYTES_THRESHOLD = 1024 * 1024; + private static final int KB = 1024; private final AtomicInteger publishCallCount = new AtomicInteger(0); private final AtomicInteger handleErrorCallCount = new AtomicInteger(0); @@ -787,7 +790,7 @@ static class TestableAmazonSqsConsumer extends AbstractAmazonSqsConsumer publishedBatchSizes = Collections.synchronizedList(new LinkedList<>()); + private final Queue publishedBatchSizes = new ConcurrentLinkedQueue<>();; TestableAmazonSqsConsumer( final Object amazonSqsClient, final QueueProperty queueProperty, @@ -848,17 +851,13 @@ void setThrowOnPublish(final boolean throwOnPublish) { } List getPublishedBatchSizes() { - return publishedBatchSizes; + return new LinkedList<>(publishedBatchSizes); } int getTotalPublishedEntries() { return publishedBatchSizes.stream().mapToInt(Integer::intValue).sum(); } - static int batchSizeBytesThreshold() { - return BATCH_SIZE_BYTES_THRESHOLD; - } - @Override public void close() { shutdown(); diff --git a/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/AbstractMessageAttributesTest.java b/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/AbstractMessageAttributesTest.java index e2054ea..1f5c1be 100644 --- a/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/AbstractMessageAttributesTest.java +++ b/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/AbstractMessageAttributesTest.java @@ -161,7 +161,7 @@ void testStringArrayFormatsStringsWithQuotes() { final String result = AbstractMessageAttributes.stringArray(list); - assertThat(result, is("[ \"foo\", \"bar\" ]")); + assertThat(result, is("[\"foo\",\"bar\"]")); } @Test @@ -173,14 +173,14 @@ void testStringArrayFiltersOutNonStringElements() { final String result = AbstractMessageAttributes.stringArray(list); - assertThat(result, is("[ \"valid\", \"also-valid\" ]")); + assertThat(result, is("[\"valid\",\"also-valid\"]")); } @Test void testStringArrayReturnsEmptyBracketsForEmptyList() { final String result = AbstractMessageAttributes.stringArray(Collections.emptyList()); - assertThat(result, is("[ ]")); + assertThat(result, is("[]")); } @Test @@ -192,7 +192,7 @@ void testStringArrayReturnsEmptyBracketsWhenNoStringElementsPresent() { final String result = AbstractMessageAttributes.stringArray(list); - assertThat(result, is("[ ]")); + assertThat(result, is("[]")); } } \ No newline at end of file diff --git a/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/JsonMapperFactoryTest.java b/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/JsonMapperFactoryTest.java index 81e8a4a..e1c286c 100644 --- a/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/JsonMapperFactoryTest.java +++ b/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/JsonMapperFactoryTest.java @@ -1,3 +1,19 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License 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 com.amazon.sqs.messaging.lib.core; import static org.hamcrest.MatcherAssert.assertThat; diff --git a/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/RequestEntryInternalFactoryTest.java b/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/RequestEntryInternalFactoryTest.java index 2615fc9..b90aa0f 100644 --- a/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/RequestEntryInternalFactoryTest.java +++ b/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/core/RequestEntryInternalFactoryTest.java @@ -387,7 +387,7 @@ void testMessageAttributesSizeWithStringAttributeCountsKeyAndValue() { final Integer result = factory.messageAttributesSize(entry); - assertThat(result, equalTo(8)); + assertThat(result, equalTo(14)); } @Test @@ -400,7 +400,7 @@ void testMessageAttributesSizeWithMultipleAttributesSumsAll() { final Integer result = factory.messageAttributesSize(entry); - assertThat(result, equalTo(11)); + assertThat(result, equalTo(23)); } @Test @@ -424,7 +424,7 @@ void testMessageAttributesSizeWithEnumAttribute() { final Integer result = factory.messageAttributesSize(entry); - assertThat(result, equalTo(10)); + assertThat(result, equalTo(16)); } @Test @@ -437,7 +437,7 @@ void testMessageAttributesSizeWithBinaryAttribute() { final Integer result = factory.messageAttributesSize(entry); - assertThat(result, equalTo(7)); + assertThat(result, equalTo(13)); } @Test @@ -568,49 +568,49 @@ void testSingletonInstanceIsSameObject() { void testGetEnumMessageAttributeReturnsNameLength() { final Integer result = instance.getEnumMessageAttribute(SampleEnum.VALUE_ONE); - assertThat(result, equalTo("VALUE_ONE".length())); + assertThat(result, equalTo(15)); } @Test void testGetEnumMessageAttributeShortName() { final Integer result = instance.getEnumMessageAttribute(SampleEnum.A); - assertThat(result, equalTo(1)); + assertThat(result, equalTo(7)); } @Test void testGetStringMessageAttributeReturnsLength() { final Integer result = instance.getStringMessageAttribute("hello"); - assertThat(result, equalTo(5)); + assertThat(result, equalTo(11)); } @Test void testGetStringMessageAttributeEmptyString() { final Integer result = instance.getStringMessageAttribute(""); - assertThat(result, equalTo(0)); + assertThat(result, equalTo(6)); } @Test void testGetNumberMessageAttributeInteger() { final Integer result = instance.getNumberMessageAttribute(12345); - assertThat(result, equalTo(5)); + assertThat(result, equalTo(29)); } @Test void testGetNumberMessageAttributeFloat() { final Integer result = instance.getNumberMessageAttribute(3.14f); - assertThat(result, equalTo(String.valueOf(3.14f).length())); + assertThat(result, equalTo(26)); } @Test void testGetNumberMessageAttributeNegativeNumber() { final Integer result = instance.getNumberMessageAttribute(-99); - assertThat(result, equalTo(3)); + assertThat(result, equalTo(27)); } @Test @@ -620,7 +620,7 @@ void testGetBinaryMessageAttributeReturnsRemaining() { final Integer result = instance.getBinaryMessageAttribute(buffer); - assertThat(result, equalTo(3)); + assertThat(result, equalTo(9)); } @Test @@ -629,7 +629,7 @@ void testGetBinaryMessageAttributeEmptyBuffer() { final Integer result = instance.getBinaryMessageAttribute(buffer); - assertThat(result, equalTo(0)); + assertThat(result, equalTo(6)); } @Test diff --git a/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/model/QueuePropertyTest.java b/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/model/QueuePropertyTest.java index aac4e4e..3f87e28 100644 --- a/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/model/QueuePropertyTest.java +++ b/amazon-sqs-java-messaging-lib-template/src/test/java/com/amazon/sqs/messaging/lib/model/QueuePropertyTest.java @@ -1,3 +1,19 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License 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 com.amazon.sqs.messaging.lib.model; import static org.hamcrest.MatcherAssert.assertThat; diff --git a/amazon-sqs-java-messaging-lib-v1/src/main/java/com/amazon/sqs/messaging/lib/core/AmazonSqsConsumerImpl.java b/amazon-sqs-java-messaging-lib-v1/src/main/java/com/amazon/sqs/messaging/lib/core/AmazonSqsConsumerImpl.java index 5a2359b..585f1d3 100644 --- a/amazon-sqs-java-messaging-lib-v1/src/main/java/com/amazon/sqs/messaging/lib/core/AmazonSqsConsumerImpl.java +++ b/amazon-sqs-java-messaging-lib-v1/src/main/java/com/amazon/sqs/messaging/lib/core/AmazonSqsConsumerImpl.java @@ -36,6 +36,7 @@ import com.amazon.sqs.messaging.lib.model.ResponseSuccessEntry; import com.amazonaws.AmazonServiceException; import com.amazonaws.services.sqs.AmazonSQS; +import com.amazonaws.services.sqs.model.AmazonSQSException; import com.amazonaws.services.sqs.model.SendMessageBatchRequest; import com.amazonaws.services.sqs.model.SendMessageBatchRequestEntry; import com.amazonaws.services.sqs.model.SendMessageBatchResult; @@ -141,13 +142,20 @@ public void handleResponse(final SendMessageBatchResult publishBatchResult) { ); publishBatchResult.getFailed().forEach(entry -> - Optional.ofNullable(pendingRequests.remove(entry.getId())).ifPresent(listenableFuture -> - listenableFuture.fail(ResponseFailEntry.builder() - .withId(entry.getId()) - .withCode(entry.getCode()) - .withMessage(entry.getMessage()) - .withSenderFault(entry.getSenderFault()) - .build()) + Optional.ofNullable(pendingRequests.remove(entry.getId())).ifPresent(listenableFuture -> { + final AmazonServiceException throwable = new AmazonSQSException(entry.getMessage()); + throwable.setErrorCode(entry.getCode()); + throwable.setErrorMessage(entry.getMessage()); + throwable.setServiceName("SQS"); + + listenableFuture.fail(ResponseFailEntry.builder() + .withId(entry.getId()) + .withCode(entry.getCode()) + .withMessage(entry.getMessage()) + .withSenderFault(entry.getSenderFault()) + .withThrowable(throwable) + .build()); + } ) ); } diff --git a/amazon-sqs-java-messaging-lib-v1/src/main/java/com/amazon/sqs/messaging/lib/core/MessageAttributes.java b/amazon-sqs-java-messaging-lib-v1/src/main/java/com/amazon/sqs/messaging/lib/core/MessageAttributes.java index 02c5d02..916a7a4 100644 --- a/amazon-sqs-java-messaging-lib-v1/src/main/java/com/amazon/sqs/messaging/lib/core/MessageAttributes.java +++ b/amazon-sqs-java-messaging-lib-v1/src/main/java/com/amazon/sqs/messaging/lib/core/MessageAttributes.java @@ -49,7 +49,7 @@ protected MessageAttributeValue getStringMessageAttribute(final String value) { */ @Override protected MessageAttributeValue getNumberMessageAttribute(final Number value) { - return new MessageAttributeValue().withDataType(NUMBER + "." + value.getClass().getName()).withStringValue(value.toString()); + return new MessageAttributeValue().withDataType(NUMBER.concat(".").concat(value.getClass().getName())).withStringValue(value.toString()); } /** diff --git a/amazon-sqs-java-messaging-lib-v1/src/test/java/com/amazon/sqs/messaging/lib/core/MessageAttributesTest.java b/amazon-sqs-java-messaging-lib-v1/src/test/java/com/amazon/sqs/messaging/lib/core/MessageAttributesTest.java index e26e59c..3d762d7 100644 --- a/amazon-sqs-java-messaging-lib-v1/src/test/java/com/amazon/sqs/messaging/lib/core/MessageAttributesTest.java +++ b/amazon-sqs-java-messaging-lib-v1/src/test/java/com/amazon/sqs/messaging/lib/core/MessageAttributesTest.java @@ -89,7 +89,7 @@ void testSuccessStringArrayHeader() { assertThat(attributes.containsKey("stringArray"), is(true)); assertThat(attributes.get("stringArray").getDataType(), is("String.Array")); - assertThat(attributes.get("stringArray").getStringValue(), is("[ \"123\", \"456\" ]")); + assertThat(attributes.get("stringArray").getStringValue(), is("[\"123\",\"456\"]")); } @Test diff --git a/amazon-sqs-java-messaging-lib-v2/src/main/java/com/amazon/sqs/messaging/lib/core/AmazonSqsConsumerImpl.java b/amazon-sqs-java-messaging-lib-v2/src/main/java/com/amazon/sqs/messaging/lib/core/AmazonSqsConsumerImpl.java index b93ebd9..14ee4cd 100644 --- a/amazon-sqs-java-messaging-lib-v2/src/main/java/com/amazon/sqs/messaging/lib/core/AmazonSqsConsumerImpl.java +++ b/amazon-sqs-java-messaging-lib-v2/src/main/java/com/amazon/sqs/messaging/lib/core/AmazonSqsConsumerImpl.java @@ -35,11 +35,13 @@ import com.amazon.sqs.messaging.lib.model.ResponseFailEntry; import com.amazon.sqs.messaging.lib.model.ResponseSuccessEntry; +import software.amazon.awssdk.awscore.exception.AwsErrorDetails; import software.amazon.awssdk.awscore.exception.AwsServiceException; import software.amazon.awssdk.services.sqs.SqsClient; import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequest; import software.amazon.awssdk.services.sqs.model.SendMessageBatchRequestEntry; import software.amazon.awssdk.services.sqs.model.SendMessageBatchResponse; +import software.amazon.awssdk.services.sqs.model.SqsException; // @formatter:off /** @@ -143,13 +145,26 @@ public void handleResponse(final SendMessageBatchResponse publishBatchResult) { ); publishBatchResult.failed().forEach(entry -> - Optional.ofNullable(pendingRequests.remove(entry.id())).ifPresent(listenableFuture -> - listenableFuture.fail(ResponseFailEntry.builder() - .withId(entry.id()) - .withCode(entry.code()) - .withMessage(entry.message()) - .withSenderFault(entry.senderFault()) - .build()) + Optional.ofNullable(pendingRequests.remove(entry.id())).ifPresent(listenableFuture -> { + final AwsErrorDetails awsErrorDetails = AwsErrorDetails.builder() + .errorCode(entry.code()) + .errorMessage(entry.message()) + .serviceName("SQS") + .build(); + + final AwsServiceException throwable = SqsException.builder() + .awsErrorDetails(awsErrorDetails) + .message(entry.message()) + .build(); + + listenableFuture.fail(ResponseFailEntry.builder() + .withId(entry.id()) + .withCode(entry.code()) + .withMessage(entry.message()) + .withSenderFault(entry.senderFault()) + .withThrowable(throwable) + .build()); + } ) ); } diff --git a/amazon-sqs-java-messaging-lib-v2/src/main/java/com/amazon/sqs/messaging/lib/core/MessageAttributes.java b/amazon-sqs-java-messaging-lib-v2/src/main/java/com/amazon/sqs/messaging/lib/core/MessageAttributes.java index 0b4ccb7..6e02efe 100644 --- a/amazon-sqs-java-messaging-lib-v2/src/main/java/com/amazon/sqs/messaging/lib/core/MessageAttributes.java +++ b/amazon-sqs-java-messaging-lib-v2/src/main/java/com/amazon/sqs/messaging/lib/core/MessageAttributes.java @@ -50,7 +50,7 @@ protected MessageAttributeValue getStringMessageAttribute(final String value) { */ @Override protected MessageAttributeValue getNumberMessageAttribute(final Number value) { - return MessageAttributeValue.builder().dataType(NUMBER + "." + value.getClass().getName()).stringValue(value.toString()).build(); + return MessageAttributeValue.builder().dataType(NUMBER.concat(".").concat(value.getClass().getName())).stringValue(value.toString()).build(); } /** diff --git a/amazon-sqs-java-messaging-lib-v2/src/test/java/com/amazon/sqs/messaging/lib/core/MessageAttributesTest.java b/amazon-sqs-java-messaging-lib-v2/src/test/java/com/amazon/sqs/messaging/lib/core/MessageAttributesTest.java index d2165dd..5a2af07 100644 --- a/amazon-sqs-java-messaging-lib-v2/src/test/java/com/amazon/sqs/messaging/lib/core/MessageAttributesTest.java +++ b/amazon-sqs-java-messaging-lib-v2/src/test/java/com/amazon/sqs/messaging/lib/core/MessageAttributesTest.java @@ -90,7 +90,7 @@ void testSuccessStringArrayHeader() { assertThat(attributes.containsKey("stringArray"), is(true)); assertThat(attributes.get("stringArray").dataType(), is("String.Array")); - assertThat(attributes.get("stringArray").stringValue(), is("[ \"123\", \"456\" ]")); + assertThat(attributes.get("stringArray").stringValue(), is("[\"123\",\"456\"]")); } @Test diff --git a/pom.xml b/pom.xml index bc6c677..e590690 100644 --- a/pom.xml +++ b/pom.xml @@ -25,7 +25,6 @@ 1.18.42 1.3.14 - 4.11.0 5.10.2 2.16.1 4.3.0 @@ -101,16 +100,19 @@ amazon-sqs-java-messaging-lib-template ${project.version} + com.github.mvallim amazon-sqs-java-messaging-lib-v1 ${project.version} + com.github.mvallim amazon-sqs-java-messaging-lib-v2 ${project.version} + com.fasterxml.jackson jackson-bom @@ -118,40 +120,62 @@ pom import + + + org.junit + junit-bom + ${junit.jupiter.version} + pom + import + + + + org.mockito + mockito-bom + ${mockito.version} + pom + import + + + + io.micrometer + micrometer-bom + ${micrometer.version} + pom + import + + + + org.testcontainers + testcontainers-bom + ${testcontainers.version} + pom + import + + org.apache.fory fory-json ${fory.version} + org.slf4j slf4j-api ${slf4j-api.version} + org.apache.commons commons-collections4 ${commons-collections4.version} + org.apache.commons commons-lang3 ${commons-lang3.version} - - io.micrometer - micrometer-bom - ${micrometer.version} - pom - import - - - org.testcontainers - testcontainers-bom - ${testcontainers.version} - pom - import - @@ -203,21 +227,18 @@ org.junit.jupiter junit-jupiter-engine - ${junit.jupiter.version} test org.junit.jupiter junit-jupiter-api - ${junit.jupiter.version} test org.junit.jupiter junit-jupiter-params - ${junit.jupiter.version} test @@ -231,21 +252,12 @@ org.mockito mockito-junit-jupiter - ${mockito.version} test org.mockito mockito-core - ${mockito.version} - test - - - - org.mockito - mockito-inline - ${mockito.version} test @@ -372,6 +384,17 @@ + + org.apache.maven.plugins + maven-dependency-plugin + + + + properties + + + + org.apache.maven.plugins maven-clean-plugin @@ -398,7 +421,7 @@ org.apache.maven.plugins maven-surefire-plugin - ${jacoco.argLine} -Dfile.encoding=${project.build.sourceEncoding} -Xms1024m -Xmx1024m -Dnet.bytebuddy.experimental=true + ${jacoco.argLine} ${mockito.argLine} -Dfile.encoding=${project.build.sourceEncoding} -Xms1024m -Xmx1024m @@ -468,51 +491,79 @@ 8 + 1.8 + 8 + 4.11.0 + + + + + org.mockito + mockito-inline + test + + 11 + 11 + 11 + 5.23.0 + -XX:+EnableDynamicAgentLoading -Xshare:off 17 + 17 + 17 + 5.23.0 + -XX:+EnableDynamicAgentLoading -Xshare:off 21 + 21 + 21 - + 5.23.0 + -XX:+EnableDynamicAgentLoading -Xshare:off + 25 + 25 + 25 + 5.23.0 + -XX:+EnableDynamicAgentLoading -Xshare:off