diff --git a/README.md b/README.md index 135f6a9..76d9b3d 100644 --- a/README.md +++ b/README.md @@ -105,7 +105,16 @@ repositories { | **`linger`** | **int** | refers to the time to wait before sending messages out to SNS. | | **`maxBatchSize`** | **int** | refers to the maximum amount of data to be collected before sending the batch. | -**NOTICE**: the buffer of message store in memory is calculate using **`maximumPoolSize`** * **`maxBatchSize`** huge values demand huge memory. +> [!NOTE] +> The buffer of message store in memory is calculated using **`maximumPoolSize`** * **`maxBatchSize`**; huge values demand huge memory. +> +> **Note on effective capacity:** the default queue implementation (`RingBufferBlockingQueue`) internally +> rounds its capacity up to the next power of two, to allow fast bitwise index calculation. This means the +> *actual* allocated capacity may be up to ~2x the value computed above — e.g. `maximumPoolSize=10` and +> `maxBatchSize=10` yields a requested capacity of 100, but the queue actually allocates 128 slots. If you +> need to budget memory precisely, use `RingBufferBlockingQueue#remainingCapacity()` (or a `LinkedBlockingQueue` +> via the [Custom `BlockingQueue`](#custom-blockingqueue) option below, which does not round up) rather than +> relying on the `maximumPoolSize * maxBatchSize` formula as an exact figure. #### Custom `BlockingQueue` diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/concurrent/RingBufferBlockingQueue.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/concurrent/RingBufferBlockingQueue.java index 1125e98..eaa6ea3 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/concurrent/RingBufferBlockingQueue.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/concurrent/RingBufferBlockingQueue.java @@ -34,6 +34,13 @@ * blocking {@link #put(Object)} and {@link #take()} operations. Other * {@link BlockingQueue} methods throw {@link UnsupportedOperationException}. * + *

Capacity rounding: the requested capacity is rounded up to the next + * power of two (see {@link #RingBufferBlockingQueue(int)}), so that slot indices + * can be computed with a bitwise AND ({@code sequence & indexMask}) instead of a + * modulo operation. As a result, the effective capacity — reported by + * {@link #remainingCapacity()} plus {@link #size()} — may be up to twice the + * value that was requested. + * * @param the type of elements held in this queue */ @SuppressWarnings({ "unchecked", "java:S3078", "java:S1948" }) @@ -83,8 +90,12 @@ public class RingBufferBlockingQueue extends AbstractQueue implements Bloc /** * Creates a ring buffer with the specified capacity. * - * @param capacity the maximum number of elements the queue can hold; must be - * positive + * @param capacity the requested capacity; must be positive. This value is + * rounded up to the next power of two internally — the queue's + * actual capacity (see {@link #remainingCapacity()}) may end up + * up to 2x larger than the value passed here. This trade-off + * enables the ring buffer to use a bitwise mask for index + * calculation instead of a modulo operation on every put/take. * @throws IllegalArgumentException if {@code capacity <= 0} */ public RingBufferBlockingQueue(final int capacity) { diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumer.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumer.java index 7efec64..a1bc29e 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumer.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumer.java @@ -23,11 +23,14 @@ import java.util.Optional; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiFunction; import java.util.function.UnaryOperator; @@ -263,7 +266,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 < AbstractAmazonSnsConsumer.BATCH_SIZE_BYTES_THRESHOLD) + return (batchSizeBytes < BATCH_SIZE_BYTES_THRESHOLD) && (requestEntriesSize < topicProperty.getMaxBatchSize()) && Objects.nonNull(request); } @@ -275,7 +278,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 <= AbstractAmazonSnsConsumer.BATCH_SIZE_BYTES_THRESHOLD; + return batchSizeBytes <= BATCH_SIZE_BYTES_THRESHOLD; } /** @@ -349,5 +352,44 @@ public CompletableFuture await() { }); } + /** + * Returns a {@link CompletableFuture} that completes once all pending requests have + * been processed (i.e., both the pending requests map and the topic requests queue are empty), + * bounded by the given timeout. + *

+ * Internally reuses {@link #await()} and waits on it via {@link CompletableFuture#get(long, TimeUnit)} + * on a separate thread, so the calling thread is never blocked directly. If the timeout elapses + * before all pending requests are drained, the returned future completes exceptionally with a + * {@link CompletionException} wrapping a {@link java.util.concurrent.TimeoutException}. + *

+ * Note that the underlying drain triggered by {@link #await()} is not cancelled when the timeout + * elapses; it keeps running in the background until the pending requests and topic requests queue + * are actually empty. + * + * @param timeout the maximum time to wait for all pending requests to be processed + * @return a future that completes when all requests are drained, or completes exceptionally + * if {@code timeout} elapses first + * @throws NullPointerException if {@code timeout} is {@code null} + */ + @Override + public CompletableFuture await(final Duration timeout) { + Objects.requireNonNull(timeout, "timeout cannot be null"); + + final CompletableFuture pending = await(); + + return CompletableFuture.runAsync(() -> { + try { + pending.get(timeout.toNanos(), TimeUnit.NANOSECONDS); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new CompletionException(ex); + } catch (final ExecutionException ex) { + throw new CompletionException(ex.getCause()); + } catch (final TimeoutException ex) { + throw new CompletionException(ex); + } + }); + } + } // @formatter:on diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducer.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducer.java index e7eead1..2d6ed0d 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducer.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducer.java @@ -16,10 +16,18 @@ package com.amazon.sns.messaging.lib.core; +import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.amazon.sns.messaging.lib.concurrent.ThreadFactoryProvider; import com.amazon.sns.messaging.lib.model.RequestEntry; import com.amazon.sns.messaging.lib.model.ResponseFailEntry; import com.amazon.sns.messaging.lib.model.ResponseSuccessEntry; @@ -39,6 +47,9 @@ @RequiredArgsConstructor(access = AccessLevel.PROTECTED) abstract class AbstractAmazonSnsProducer implements AmazonSnsProducer { + /** Class logger. */ + private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAmazonSnsProducer.class); + /** The producer lifecycle state, initially {@link State#RUNNING}. */ private final AtomicReference state = new AtomicReference<>(State.RUNNING); @@ -48,6 +59,8 @@ abstract class AbstractAmazonSnsProducer implements AmazonSnsProducer { /** The blocking queue for buffering requests before batch processing. */ private final BlockingQueue> topicRequests; + private final ExecutorService callbackExecutor = Executors.newCachedThreadPool(ThreadFactoryProvider.getThreadFactory()); + /** * Sends a request entry by enqueuing it for batch processing. * @@ -59,7 +72,7 @@ public ListenableFuture send(final Requ if (State.RUNNING.equals(state.get())) { return enqueueRequest(requestEntry); } else { - final ListenableFutureImpl listenableFutureImpl = new ListenableFutureImpl(); + final ListenableFutureImpl listenableFutureImpl = new ListenableFutureImpl(Runnable::run); listenableFutureImpl.fail(ResponseFailEntry.builder() .withCode("000") @@ -78,8 +91,24 @@ public ListenableFuture send(final Requ * accepted once shutdown. */ @Override - public void shutdown() { + public void shutdown(final Runnable runnable) { state.compareAndSet(State.RUNNING, State.SHUTDOWN); + + runnable.run(); + + try { + LOGGER.warn("Shutdown producer {}", getClass().getSimpleName()); + + callbackExecutor.shutdown(); + if (!callbackExecutor.awaitTermination(60, TimeUnit.SECONDS)) { + LOGGER.warn("Producer executor service did not terminate in the specified time."); + final List droppedTasks = callbackExecutor.shutdownNow(); + LOGGER.warn("Producer executor service was abruptly shut down. {} tasks will not be executed.", droppedTasks.size()); + } + } catch (final InterruptedException ex) { + LOGGER.error(ex.getMessage(), ex); + Thread.currentThread().interrupt(); + } } /** @@ -92,7 +121,7 @@ public void shutdown() { @SneakyThrows private ListenableFuture enqueueRequest(final RequestEntry requestEntry) { try { - final ListenableFuture trackPendingRequest = new ListenableFutureImpl(); + final ListenableFuture trackPendingRequest = new ListenableFutureImpl(callbackExecutor); pendingRequests.put(requestEntry.getId(), trackPendingRequest); topicRequests.put(requestEntry); return trackPendingRequest; diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsTemplate.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsTemplate.java index 880b487..01d957c 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsTemplate.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsTemplate.java @@ -16,6 +16,7 @@ package com.amazon.sns.messaging.lib.core; +import java.time.Duration; import java.util.Objects; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; @@ -72,8 +73,7 @@ public ListenableFuture send(final Requ * Shuts down both the producer and consumer gracefully. */ public void shutdown() { - amazonSnsProducer.shutdown(); - amazonSnsConsumer.shutdown(); + amazonSnsProducer.shutdown(amazonSnsConsumer::shutdown); } /** @@ -85,6 +85,20 @@ public CompletableFuture await() { return amazonSnsConsumer.await(); } + /** + * Returns a future that completes once all pending requests are drained and processed, + * bounded by the given timeout. + * + * @param timeout the maximum time to wait for all pending requests to be processed + * @return a {@link CompletableFuture} that completes when the consumer has finished, or + * completes exceptionally with a {@link java.util.concurrent.TimeoutException} + * if {@code timeout} elapses first + * @throws NullPointerException if {@code timeout} is {@code null} + */ + public CompletableFuture await(final Duration timeout) { + return amazonSnsConsumer.await(timeout); + } + /** * Creates an {@link AmazonSnsThreadPoolExecutor} configured for the given topic property. * For FIFO topics, a single-threaded pool is used to guarantee order. diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumer.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumer.java index 1741aa7..b4ddc1a 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumer.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsConsumer.java @@ -16,6 +16,7 @@ package com.amazon.sns.messaging.lib.core; +import java.time.Duration; import java.util.concurrent.CompletableFuture; /** @@ -66,4 +67,18 @@ public interface AmazonSnsConsumer { */ public CompletableFuture await(); + /** + * Returns a {@link CompletableFuture} that completes once all pending requests + * have been processed, bounded by the given timeout. If the timeout elapses + * before all pending requests are drained, the returned future completes + * exceptionally with a {@link java.util.concurrent.TimeoutException}. + * + * @param timeout the maximum time to wait for all pending requests to be + * processed + * @return a future that completes when all requests are drained, or completes + * exceptionally if {@code timeout} elapses first + * @throws NullPointerException if {@code timeout} is {@code null} + */ + public CompletableFuture await(final Duration timeout); + } diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsProducer.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsProducer.java index 526584e..592ef76 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsProducer.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AmazonSnsProducer.java @@ -42,7 +42,7 @@ public interface AmazonSnsProducer { /** * Shuts down the producer, preventing any further messages from being accepted. */ - public void shutdown(); + public void shutdown(final Runnable runnable); } // @formatter:on diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/ListenableFuture.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/ListenableFuture.java index d4bc6d6..a6fdb9a 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/ListenableFuture.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/ListenableFuture.java @@ -18,6 +18,9 @@ import static java.util.function.Function.identity; +import java.time.Duration; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeoutException; import java.util.function.Consumer; import com.amazon.sns.messaging.lib.model.ResponseFailEntry; @@ -64,5 +67,32 @@ default void addCallback(final Consumer successCallback) { */ void fail(final ResponseFailEntry entry); + /** + * Waits, if necessary, for this future to complete, then returns its success result. + *

+ * Mirrors the contract of {@link java.util.concurrent.Future#get()}: if the future + * completed with a failure, this throws {@link ExecutionException} instead of returning. + * The exception's cause is the {@link Throwable} carried by the failure result, if any + * (see {@link ResponseFailEntry#getThrowable()}), or a new exception built from the + * failure result's message otherwise. + * + * @return the success result + * @throws InterruptedException if the current thread is interrupted while waiting + * @throws ExecutionException if the future completed with a failure + */ + S get() throws InterruptedException, ExecutionException; + + /** + * Waits, if necessary, for at most the given timeout for this future to complete, then + * returns its success result. See {@link #get()} for failure semantics. + * + * @param timeout the maximum time to wait + * @return the success result + * @throws InterruptedException if the current thread is interrupted while waiting + * @throws ExecutionException if the future completed with a failure + * @throws TimeoutException if the timeout elapses before the future completes + */ + S get(final Duration timeout) throws InterruptedException, ExecutionException, TimeoutException; + } // @formatter:on diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/ListenableFutureImpl.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/ListenableFutureImpl.java index 443b138..d0ad7c9 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/ListenableFutureImpl.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/ListenableFutureImpl.java @@ -16,111 +16,157 @@ package com.amazon.sns.messaging.lib.core; +import static br.com.fluentvalidator.predicate.LogicalPredicate.not; import static java.util.function.Function.identity; -import java.util.LinkedList; +import java.time.Duration; +import java.util.Objects; import java.util.Optional; -import java.util.Queue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.function.Consumer; -import org.apache.commons.collections4.CollectionUtils; - import com.amazon.sns.messaging.lib.model.ResponseFailEntry; import com.amazon.sns.messaging.lib.model.ResponseSuccessEntry; -import lombok.AccessLevel; import lombok.Getter; // @formatter:off /** - * Default implementation of {@link ListenableFuture}. Supports state tracking (NEW, SUCCESS, FAILURE) - * and thread-safe callback registration and notification. + * Default implementation of {@link ListenableFuture}. Supports state tracking + * (NEW, SUCCESS, FAILURE), thread-safe callback registration and notification, + * and blocking retrieval of the result via {@link #get()} / + * {@link #get(Duration)}. + *

+ * All state transitions and reads are guarded by {@link #mutex}. Completion + * ({@link #success(ResponseSuccessEntry)} or {@link #fail(ResponseFailEntry)}) + * both notifies any registered callbacks synchronously and wakes up any thread + * blocked in {@link #get()} / {@link #get(Duration)} via + * {@link Object#notifyAll()}. */ class ListenableFutureImpl implements ListenableFuture { - private final Object mutex = new Object(); - - @Getter(value = AccessLevel.PACKAGE) - private State state = State.NEW; - - @Getter(value = AccessLevel.PACKAGE) - private ResponseSuccessEntry successResult; - - @Getter(value = AccessLevel.PACKAGE) - private ResponseFailEntry failureResult; + /** + * Runs registered callbacks off of the calling (consumer) thread; see class + * Javadoc. + */ + private final Executor callbackExecutor; - private final Queue> successCallback = new LinkedList<>(); + /** + * Backing future. Completed with a success result, or exceptionally with a + * {@link FailureSignal}. + */ + private final CompletableFuture delegate = new CompletableFuture<>(); - private final Queue> failureCallback = new LinkedList<>(); + /** + * Creates a new future whose callbacks are dispatched on the given executor. + * + * @param callbackExecutor the executor used to run success/failure callbacks; + * must not be one of the library's own publish/consumer + * executors, to avoid starving them (see class Javadoc) + */ + ListenableFutureImpl(final Executor callbackExecutor) { + this.callbackExecutor = Objects.requireNonNull(callbackExecutor, "callbackExecutor cannot be null"); + } @Override public void addCallback(final Consumer successCallback, final Consumer failureCallback) { - synchronized (mutex) { - final Consumer success = Optional.ofNullable(successCallback).orElse(identity()::apply); - final Consumer failure = Optional.ofNullable(failureCallback).orElse(identity()::apply); - - switch (state) { - case NEW: - this.successCallback.add(success); - this.failureCallback.add(failure); - break; - case SUCCESS: - notifySuccess(success); - break; - case FAILURE: - notifyFailure(failure); - break; + final Consumer success = Optional.ofNullable(successCallback).orElse(identity()::apply); + final Consumer failure = Optional.ofNullable(failureCallback).orElse(identity()::apply); + + delegate.whenCompleteAsync((result, throwable) -> { + if (Objects.isNull(throwable)) { + success.accept(result); + } else { + failure.accept(unwrap(throwable)); } - } + }, callbackExecutor); } @Override public void success(final ResponseSuccessEntry entry) { - synchronized (mutex) { - state = State.SUCCESS; - successResult = entry; - - while (CollectionUtils.isNotEmpty(successCallback)) { - notifySuccess(successCallback.poll()); - } + if (not(delegate::complete).test(entry)) { + throw new IllegalStateException("ListenableFuture already completed."); } } @Override public void fail(final ResponseFailEntry entry) { - synchronized (mutex) { - state = State.FAILURE; - failureResult = entry; + if (not(delegate::completeExceptionally).test(new FailureSignal(entry))) { + throw new IllegalStateException("ListenableFuture already completed."); + } + } - while (CollectionUtils.isNotEmpty(failureCallback)) { - notifyFailure(failureCallback.poll()); - } + @Override + public ResponseSuccessEntry get() throws InterruptedException, ExecutionException { + try { + return delegate.get(); + } catch (final ExecutionException ex) { + throw new ExecutionException(unwrapCause(ex.getCause())); + } + } + + @Override + public ResponseSuccessEntry get(final Duration timeout) throws InterruptedException, ExecutionException, TimeoutException { + Objects.requireNonNull(timeout, "timeout cannot be null"); + + try { + return delegate.get(timeout.toNanos(), TimeUnit.NANOSECONDS); + } catch (final ExecutionException ex) { + throw new ExecutionException(unwrapCause(ex.getCause())); } } /** - * Notifies a single success callback with the stored result. + * Unwraps a {@link FailureSignal} into the {@link ResponseFailEntry} it + * carries. * - * @param callback the callback to invoke + * @param throwable the throwable passed to + * {@link CompletableFuture#whenCompleteAsync}; either a + * {@link FailureSignal} directly, or (in composed/chained + * usages) a {@link CompletionException} wrapping one + * @return the original failure result */ - private void notifySuccess(final Consumer callback) { - callback.accept(successResult); + private static ResponseFailEntry unwrap(final Throwable throwable) { + final Throwable cause = throwable instanceof CompletionException ? throwable.getCause() : throwable; + return FailureSignal.class.cast(cause).getEntry(); } /** - * Notifies a single failure callback with the stored result. + * Returns the cause that {@link #get()} / {@link #get(Duration)} should report: + * the original {@link Throwable} carried by the {@link ResponseFailEntry} if + * present, or the entry itself (via {@link FailureSignal}) as a fallback. * - * @param callback the callback to invoke + * @param cause the cause of the {@link ExecutionException} thrown by the + * backing future, expected to be a {@link FailureSignal} + * @return the throwable to expose as the {@link ExecutionException}'s cause */ - private void notifyFailure(final Consumer callback) { - callback.accept(failureResult); + private static Throwable unwrapCause(final Throwable cause) { + final ResponseFailEntry entry = FailureSignal.class.cast(cause).getEntry(); + return Optional.ofNullable(entry.getThrowable()).orElseGet(() -> new IllegalStateException(entry.getMessage())); } /** - * The lifecycle states of a {@link ListenableFuture}. + * Wraps a {@link ResponseFailEntry} so it can be used to complete the backing + * {@link CompletableFuture} exceptionally via + * {@link CompletableFuture#completeExceptionally(Throwable)}. */ - enum State { - NEW, SUCCESS, FAILURE + private static final class FailureSignal extends RuntimeException { + + private static final long serialVersionUID = -1790175325395257266L; + + @Getter + private final transient ResponseFailEntry entry; + + FailureSignal(final ResponseFailEntry entry) { + super(entry.getMessage()); + this.entry = entry; + } + } } diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactory.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactory.java index a2ac06c..ef0aebe 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactory.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/RequestEntryInternalFactory.java @@ -144,9 +144,9 @@ static class RequestEntryInternal { private final String deduplicationId; /** - * Returns the size of the serialized payload in bytes. + * Returns the size of the binary payload in bytes. * - * @return the payload size + * @return the payload size in bytes */ public int size() { return value.capacity(); @@ -163,6 +163,10 @@ public String getMessage() { } + /** + * Internal implementation of {@link AbstractMessageAttributes} that calculates + * attribute size values for batching decisions. + */ @SuppressWarnings("java:S6548") @NoArgsConstructor(access = AccessLevel.PRIVATE) static class MessageAttributesInternal extends AbstractMessageAttributes { diff --git a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/metrics/AbstractAmazonSnsConsumerMetricsDecorator.java b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/metrics/AbstractAmazonSnsConsumerMetricsDecorator.java index 836e689..5c7e9de 100644 --- a/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/metrics/AbstractAmazonSnsConsumerMetricsDecorator.java +++ b/amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/metrics/AbstractAmazonSnsConsumerMetricsDecorator.java @@ -16,6 +16,7 @@ package com.amazon.sns.messaging.lib.metrics; +import java.time.Duration; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicInteger; @@ -178,4 +179,11 @@ public CompletableFuture await() { return delegate.await(); } + /** + * {@inheritDoc} + */ + @Override + public CompletableFuture await(final Duration timeout) { + return delegate.await(timeout); + } } diff --git a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducerTest.java b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducerTest.java index eba4111..5d0de1f 100644 --- a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducerTest.java +++ b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducerTest.java @@ -64,7 +64,7 @@ void setUp() { @AfterEach void tearDown() { if (Objects.nonNull(producer)) { - producer.shutdown(); + producer.shutdown(() -> {}); } } @@ -74,7 +74,7 @@ void testSendReturnsShutdownState() throws InterruptedException { final RequestEntry entry = requestEntry(); - producer.shutdown(); + producer.shutdown(() -> {}); final ListenableFuture future = producer.send(entry); diff --git a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsTemplateTest.java b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsTemplateTest.java index fa5c81f..47b0ac3 100644 --- a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsTemplateTest.java +++ b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsTemplateTest.java @@ -22,6 +22,8 @@ import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.notNullValue; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -31,9 +33,11 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingDeque; import java.util.function.UnaryOperator; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -64,15 +68,23 @@ class AbstractAmazonSnsTemplateTest { private AbstractAmazonSnsTemplate template; + private ExecutorService callbackExecutor; + @BeforeEach void setUp() { + callbackExecutor = Executors.newSingleThreadExecutor(); template = new AbstractAmazonSnsTemplate(producerMock, consumerMock) { }; } + @AfterEach + void tearDown() { + callbackExecutor.shutdownNow(); + } + @Test void testSendDelegatesToProducer() { final RequestEntry requestEntry = RequestEntry.builder().build(); - final ListenableFuture expectedFuture = new ListenableFutureImpl(); + final ListenableFuture expectedFuture = new ListenableFutureImpl(callbackExecutor); when(producerMock.send(requestEntry)).thenReturn(expectedFuture); final ListenableFuture result = template.send(requestEntry); @@ -84,12 +96,19 @@ void testSendDelegatesToProducer() { @Test void testShutdownDelegatesToProducer() { template.shutdown(); - verify(producerMock).shutdown(); + verify(producerMock).shutdown(any()); } @Test void testShutdownDelegatesToConsumer() { + doAnswer(invocation -> { + final Runnable argument = invocation.getArgument(0, Runnable.class); + argument.run(); + return null; + }).when(producerMock).shutdown(any()); + template.shutdown(); + verify(consumerMock).shutdown(); } diff --git a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/ListenableFutureImplTest.java b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/ListenableFutureImplTest.java index 926cf8c..d64c39e 100644 --- a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/ListenableFutureImplTest.java +++ b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/ListenableFutureImplTest.java @@ -16,118 +16,596 @@ package com.amazon.sns.messaging.lib.core; +import static org.hamcrest.CoreMatchers.instanceOf; import static org.hamcrest.CoreMatchers.is; -import static org.hamcrest.CoreMatchers.notNullValue; import static org.hamcrest.CoreMatchers.nullValue; +import static org.hamcrest.CoreMatchers.sameInstance; import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import com.amazon.sns.messaging.lib.core.ListenableFutureImpl.State; import com.amazon.sns.messaging.lib.model.ResponseFailEntry; import com.amazon.sns.messaging.lib.model.ResponseSuccessEntry; // @formatter:off class ListenableFutureImplTest { + private static final Duration CALLBACK_TIMEOUT = Duration.ofSeconds(5); + + private ExecutorService callbackExecutor; + + @BeforeEach + void setUp() { + callbackExecutor = Executors.newSingleThreadExecutor(); + } + + @AfterEach + void tearDown() { + callbackExecutor.shutdownNow(); + } + + @Test + void testSuccessWithCallbacksBefore() throws InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference captured = new AtomicReference<>(); + + final Consumer successCallback = entry -> { + captured.set(entry); + latch.countDown(); + }; + final Consumer failureCallback = entry -> { throw new AssertionError("should not be called"); }; + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + listenableFutureRegistry.addCallback(successCallback, failureCallback); + + final ResponseSuccessEntry entry = mock(ResponseSuccessEntry.class); + listenableFutureRegistry.success(entry); + + assertTrue(latch.await(CALLBACK_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), "success callback was not invoked in time"); + assertThat(captured.get(), sameInstance(entry)); + } + + @Test + void testFailWithCallbacksBefore() throws InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference captured = new AtomicReference<>(); + + final Consumer successCallback = entry -> { throw new AssertionError("should not be called"); }; + final Consumer failureCallback = entry -> { + captured.set(entry); + latch.countDown(); + }; + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + listenableFutureRegistry.addCallback(successCallback, failureCallback); + + final ResponseFailEntry entry = mock(ResponseFailEntry.class); + listenableFutureRegistry.fail(entry); + + assertTrue(latch.await(CALLBACK_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), "failure callback was not invoked in time"); + assertThat(captured.get(), sameInstance(entry)); + } + @Test - void testSuccessWithCallbacksBefore() { - final Consumer successCallback = entry -> assertThat(entry, notNullValue()); - final Consumer failureCallback = entry -> assertThat(entry, notNullValue()); + void testSuccessWithCallbacksAfter() throws InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference captured = new AtomicReference<>(); + + final Consumer successCallback = entry -> { + captured.set(entry); + latch.countDown(); + }; + final Consumer failureCallback = entry -> { throw new AssertionError("should not be called"); }; - final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(); + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + final ResponseSuccessEntry entry = mock(ResponseSuccessEntry.class); + listenableFutureRegistry.success(entry); listenableFutureRegistry.addCallback(successCallback, failureCallback); - listenableFutureRegistry.success(mock(ResponseSuccessEntry.class)); - listenableFutureRegistry.fail(mock(ResponseFailEntry.class)); + assertTrue(latch.await(CALLBACK_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), "success callback was not invoked in time"); + assertThat(captured.get(), sameInstance(entry)); } @Test - void testSuccessWithCallbacksAfter() { - final Consumer successCallback = entry -> assertThat(entry, notNullValue()); - final Consumer failureCallback = entry -> assertThat(entry, notNullValue()); + void testFailWithCallbacksAfter() throws InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference captured = new AtomicReference<>(); + + final Consumer successCallback = entry -> { throw new AssertionError("should not be called"); }; + final Consumer failureCallback = entry -> { + captured.set(entry); + latch.countDown(); + }; - final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(); + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); - listenableFutureRegistry.success(mock(ResponseSuccessEntry.class)); - listenableFutureRegistry.fail(mock(ResponseFailEntry.class)); + final ResponseFailEntry entry = mock(ResponseFailEntry.class); + listenableFutureRegistry.fail(entry); listenableFutureRegistry.addCallback(successCallback, failureCallback); + + assertTrue(latch.await(CALLBACK_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), "failure callback was not invoked in time"); + assertThat(captured.get(), sameInstance(entry)); } @Test - void testSuccessWithCallbackSuccessBefore() { - final Consumer successCallback = entry -> assertThat(entry, notNullValue()); + void testSuccessWithCallbackSuccessBefore() throws InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference captured = new AtomicReference<>(); - final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(); + final Consumer successCallback = entry -> { + captured.set(entry); + latch.countDown(); + }; + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); listenableFutureRegistry.addCallback(successCallback, null); - listenableFutureRegistry.success(mock(ResponseSuccessEntry.class)); + final ResponseSuccessEntry entry = mock(ResponseSuccessEntry.class); + listenableFutureRegistry.success(entry); + + assertTrue(latch.await(CALLBACK_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), "success callback was not invoked in time"); + assertThat(captured.get(), sameInstance(entry)); } @Test - void testSuccessWithCallbackSuccessAfter() { - final Consumer successCallback = entry -> assertThat(entry, notNullValue()); + void testSuccessWithCallbackSuccessAfter() throws InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference captured = new AtomicReference<>(); - final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(); + final Consumer successCallback = entry -> { + captured.set(entry); + latch.countDown(); + }; - listenableFutureRegistry.success(mock(ResponseSuccessEntry.class)); + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + final ResponseSuccessEntry entry = mock(ResponseSuccessEntry.class); + listenableFutureRegistry.success(entry); listenableFutureRegistry.addCallback(successCallback, null); + + assertTrue(latch.await(CALLBACK_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), "success callback was not invoked in time"); + assertThat(captured.get(), sameInstance(entry)); } @Test - void testSuccessWithCallbackFailBefore() { - final Consumer failureCallback = entry -> assertThat(entry, notNullValue()); + void testSuccessWithCallbackFailBefore() throws InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference captured = new AtomicReference<>(); + + final Consumer failureCallback = entry -> { + captured.set(entry); + latch.countDown(); + }; - final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(); + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); listenableFutureRegistry.addCallback(null, failureCallback); - listenableFutureRegistry.fail(mock(ResponseFailEntry.class)); + final ResponseFailEntry entry = mock(ResponseFailEntry.class); + listenableFutureRegistry.fail(entry); + + assertTrue(latch.await(CALLBACK_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), "failure callback was not invoked in time"); + assertThat(captured.get(), sameInstance(entry)); } @Test - void testSuccessWithCallbackFailAfter() { - final Consumer failureCallback = entry -> assertThat(entry, notNullValue()); + void testSuccessWithCallbackFailAfter() throws InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference captured = new AtomicReference<>(); + + final Consumer failureCallback = entry -> { + captured.set(entry); + latch.countDown(); + }; - final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(); + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); - listenableFutureRegistry.fail(mock(ResponseFailEntry.class)); + final ResponseFailEntry entry = mock(ResponseFailEntry.class); + listenableFutureRegistry.fail(entry); listenableFutureRegistry.addCallback(null, failureCallback); + + assertTrue(latch.await(CALLBACK_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), "failure callback was not invoked in time"); + assertThat(captured.get(), sameInstance(entry)); } @Test - void testSuccessWithoutCallbacksBefore() { - final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(); + void testSuccessWithoutCallbacksBefore() throws Exception { + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); listenableFutureRegistry.addCallback(null, null); - listenableFutureRegistry.success(mock(ResponseSuccessEntry.class)); + final ResponseSuccessEntry entry = mock(ResponseSuccessEntry.class); + listenableFutureRegistry.success(entry); - assertThat(listenableFutureRegistry.getState(), is(State.SUCCESS)); - assertThat(listenableFutureRegistry.getSuccessResult(), notNullValue()); - assertThat(listenableFutureRegistry.getFailureResult(), nullValue()); + assertThat(listenableFutureRegistry.get(), sameInstance(entry)); } @Test void testSuccessWithoutCallbacksAfter() { - final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(); + final RuntimeException cause = new RuntimeException("boom"); + final ResponseFailEntry entry = ResponseFailEntry.builder() + .withId("id-1") + .withMessage("boom") + .withThrowable(cause) + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); - listenableFutureRegistry.fail(mock(ResponseFailEntry.class)); + listenableFutureRegistry.fail(entry); listenableFutureRegistry.addCallback(null, null); - assertThat(listenableFutureRegistry.getState(), is(State.FAILURE)); - assertThat(listenableFutureRegistry.getFailureResult(), notNullValue()); - assertThat(listenableFutureRegistry.getSuccessResult(), nullValue()); + final ExecutionException ex = assertThrows(ExecutionException.class, listenableFutureRegistry::get); + assertThat(ex.getCause(), sameInstance(cause)); + } + + @Test + void testEnsureNotCompletedThrowsWhenSuccessIsCalledTwice() throws Exception { + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + final ResponseSuccessEntry firstEntry = mock(ResponseSuccessEntry.class); + + listenableFutureRegistry.success(firstEntry); + + final IllegalStateException ex = assertThrows(IllegalStateException.class, () -> + listenableFutureRegistry.success(mock(ResponseSuccessEntry.class))); + + assertThat(ex.getMessage(), is("ListenableFuture already completed.")); + assertThat(listenableFutureRegistry.get(), sameInstance(firstEntry)); + } + + @Test + void testEnsureNotCompletedThrowsWhenFailIsCalledAfterSuccess() throws Exception { + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + final ResponseSuccessEntry firstEntry = mock(ResponseSuccessEntry.class); + + listenableFutureRegistry.success(firstEntry); + + final IllegalStateException ex = assertThrows(IllegalStateException.class, () -> + listenableFutureRegistry.fail(mock(ResponseFailEntry.class))); + + assertThat(ex.getMessage(), is("ListenableFuture already completed.")); + assertThat(listenableFutureRegistry.get(), sameInstance(firstEntry)); + } + + @Test + void testEnsureNotCompletedThrowsWhenFailIsCalledTwice() { + final RuntimeException firstCause = new RuntimeException("first boom"); + final ResponseFailEntry firstEntry = ResponseFailEntry.builder() + .withId("id-1") + .withMessage("first boom") + .withThrowable(firstCause) + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + listenableFutureRegistry.fail(firstEntry); + + final RuntimeException secondCause = new RuntimeException("second boom"); + final ResponseFailEntry secondEntry = ResponseFailEntry.builder() + .withId("id-1") + .withMessage("second boom") + .withThrowable(secondCause) + .build(); + + final IllegalStateException ex = assertThrows(IllegalStateException.class, () -> + listenableFutureRegistry.fail(secondEntry)); + + assertThat(ex.getMessage(), is("ListenableFuture already completed.")); + + final ExecutionException getEx = assertThrows(ExecutionException.class, listenableFutureRegistry::get); + assertThat(getEx.getCause(), sameInstance(firstCause)); + } + + @Test + void testEnsureNotCompletedThrowsWhenSuccessIsCalledAfterFail() { + final RuntimeException firstCause = new RuntimeException("boom"); + final ResponseFailEntry firstEntry = ResponseFailEntry.builder() + .withId("id-1") + .withMessage("boom") + .withThrowable(firstCause) + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + listenableFutureRegistry.fail(firstEntry); + + final IllegalStateException ex = assertThrows(IllegalStateException.class, () -> + listenableFutureRegistry.success(mock(ResponseSuccessEntry.class))); + + assertThat(ex.getMessage(), is("ListenableFuture already completed.")); + + final ExecutionException getEx = assertThrows(ExecutionException.class, listenableFutureRegistry::get); + assertThat(getEx.getCause(), sameInstance(firstCause)); + } + + @Test + void testGetReturnsImmediatelyWhenAlreadySucceeded() throws Exception { + final ResponseSuccessEntry entry = ResponseSuccessEntry.builder() + .withId("id-1") + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + listenableFutureRegistry.success(entry); + + assertThat(listenableFutureRegistry.get(), sameInstance(entry)); + } + + @Test + void testGetThrowsExecutionExceptionWithOriginalThrowableWhenAlreadyFailed() { + final RuntimeException cause = new RuntimeException("boom"); + + final ResponseFailEntry entry = ResponseFailEntry.builder() + .withId("id-1") + .withMessage("boom") + .withThrowable(cause) + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + listenableFutureRegistry.fail(entry); + + final ExecutionException ex = assertThrows(ExecutionException.class, listenableFutureRegistry::get); + + assertThat(ex.getCause(), sameInstance(cause)); + } + + @Test + void testGetThrowsExecutionExceptionWithFallbackCauseWhenThrowableIsAbsent() { + final ResponseFailEntry entry = ResponseFailEntry.builder() + .withId("id-1") + .withMessage("no throwable available") + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + listenableFutureRegistry.fail(entry); + + final ExecutionException ex = assertThrows(ExecutionException.class, listenableFutureRegistry::get); + + assertThat(ex.getCause(), instanceOf(IllegalStateException.class)); + assertThat(ex.getCause().getMessage(), is("no throwable available")); + } + + @Test + void testGetBlocksUntilSuccessIsCalledFromAnotherThread() { + final ResponseSuccessEntry entry = ResponseSuccessEntry.builder() + .withId("id-1") + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + final CountDownLatch aboutToBlock = new CountDownLatch(1); + final AtomicReference result = new AtomicReference<>(); + final AtomicReference error = new AtomicReference<>(); + + final Thread waiter = new Thread(() -> { + try { + aboutToBlock.countDown(); + result.set(listenableFutureRegistry.get()); + } catch (final Exception ex) { + error.set(ex); + } + }); + + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + waiter.start(); + aboutToBlock.await(); + + Thread.sleep(100); + + listenableFutureRegistry.success(entry); + + waiter.join(); + }); + + assertThat(error.get(), nullValue()); + assertThat(result.get(), sameInstance(entry)); + } + + @Test + void testGetBlocksUntilFailIsCalledFromAnotherThread() { + final RuntimeException cause = new RuntimeException("boom"); + + final ResponseFailEntry entry = ResponseFailEntry.builder() + .withId("id-1") + .withMessage("boom") + .withThrowable(cause) + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + final CountDownLatch aboutToBlock = new CountDownLatch(1); + final AtomicReference error = new AtomicReference<>(); + + final Thread waiter = new Thread(() -> { + try { + aboutToBlock.countDown(); + listenableFutureRegistry.get(); + } catch (final Exception ex) { + error.set(ex); + } + }); + + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + waiter.start(); + aboutToBlock.await(); + + Thread.sleep(100); + + listenableFutureRegistry.fail(entry); + + waiter.join(); + }); + + assertThat(error.get(), instanceOf(ExecutionException.class)); + assertThat(error.get().getCause(), sameInstance(cause)); + } + + @Test + void testGetPropagatesInterruptedExceptionWhenInterruptedWhileWaiting() { + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + final CountDownLatch aboutToBlock = new CountDownLatch(1); + final AtomicReference error = new AtomicReference<>(); + + final Thread waiter = new Thread(() -> { + try { + aboutToBlock.countDown(); + listenableFutureRegistry.get(); + } catch (final Exception ex) { + error.set(ex); + } + }); + + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + waiter.start(); + aboutToBlock.await(); + + Thread.sleep(100); + + waiter.interrupt(); + waiter.join(); + }); + + assertThat(error.get(), instanceOf(InterruptedException.class)); + } + + @Test + void testGetWithDurationReturnsImmediatelyWhenAlreadySucceeded() throws Exception { + final ResponseSuccessEntry entry = ResponseSuccessEntry.builder() + .withId("id-1") + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + listenableFutureRegistry.success(entry); + + assertThat(listenableFutureRegistry.get(Duration.ofSeconds(1)), sameInstance(entry)); + } + + @Test + void testGetWithDurationCompletesBeforeTimeoutElapses() { + final ResponseSuccessEntry entry = ResponseSuccessEntry.builder() + .withId("id-1") + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + final AtomicReference result = new AtomicReference<>(); + final AtomicReference error = new AtomicReference<>(); + + final Thread waiter = new Thread(() -> { + try { + result.set(listenableFutureRegistry.get(Duration.ofSeconds(5))); + } catch (final Exception ex) { + error.set(ex); + } + }); + + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + waiter.start(); + + Thread.sleep(100); + + listenableFutureRegistry.success(entry); + + waiter.join(); + }); + + assertThat(error.get(), nullValue()); + assertThat(result.get(), sameInstance(entry)); + } + + @Test + void testGetWithDurationThrowsTimeoutExceptionWhenNeverCompleted() { + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + assertThrows(TimeoutException.class, () -> listenableFutureRegistry.get(Duration.ofMillis(100))) + ); + + assertDoesNotThrow(() -> listenableFutureRegistry.success(mock(ResponseSuccessEntry.class))); + } + + @Test + void testGetWithDurationThrowsNullPointerExceptionWhenTimeoutIsNull() { + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + assertThrows(NullPointerException.class, () -> listenableFutureRegistry.get(null)); + } + + @Test + void testGetWithDurationThrowsExecutionExceptionWhenAlreadyFailed() { + final RuntimeException cause = new RuntimeException("boom"); + + final ResponseFailEntry entry = ResponseFailEntry.builder() + .withId("id-1") + .withMessage("boom") + .withThrowable(cause) + .build(); + + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + listenableFutureRegistry.fail(entry); + + final ExecutionException ex = assertThrows(ExecutionException.class, () -> + listenableFutureRegistry.get(Duration.ofSeconds(1))); + + assertThat(ex.getCause(), sameInstance(cause)); + } + + @Test + void testGetWithDurationTimesOutSeparatelyFromInterruption() { + final ListenableFutureImpl listenableFutureRegistry = new ListenableFutureImpl(callbackExecutor); + + final AtomicReference error = new AtomicReference<>(); + final CountDownLatch aboutToBlock = new CountDownLatch(1); + + final Thread waiter = new Thread(() -> { + try { + aboutToBlock.countDown(); + listenableFutureRegistry.get(Duration.ofMillis(200)); + } catch (final Exception ex) { + error.set(ex); + } + }); + + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + waiter.start(); + aboutToBlock.await(); + waiter.join(); + }); + + assertThat(error.get(), instanceOf(TimeoutException.class)); } } -// @formatter:on +// @formatter:on \ No newline at end of file diff --git a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/ListenableFutureTest.java b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/ListenableFutureTest.java index 1b7cfa6..6c3019b 100644 --- a/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/ListenableFutureTest.java +++ b/amazon-sns-java-messaging-lib-template/src/test/java/com/amazon/sns/messaging/lib/core/ListenableFutureTest.java @@ -20,10 +20,15 @@ import static org.hamcrest.Matchers.is; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.function.Consumer; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -37,6 +42,8 @@ @ExtendWith(MockitoExtension.class) class ListenableFutureTest { + private static final int CALLBACK_TIMEOUT = 3000; + private ListenableFutureImpl listenableFuture; @Mock(strictness = Strictness.LENIENT) @@ -45,9 +52,17 @@ class ListenableFutureTest { @Mock(strictness = Strictness.LENIENT) private Consumer failureCallback; + private ExecutorService callbackExecutor; + @BeforeEach void setUp() { - listenableFuture = new ListenableFutureImpl(); + callbackExecutor = Executors.newSingleThreadExecutor(); + listenableFuture = new ListenableFutureImpl(callbackExecutor); + } + + @AfterEach + void tearDown() { + callbackExecutor.shutdownNow(); } @Test @@ -57,7 +72,7 @@ void testAddCallbackInvokesSuccessCallbackOnSuccess() { listenableFuture.addCallback(successCallback, failureCallback); listenableFuture.success(entry); - verify(successCallback).accept(entry); + verify(successCallback, timeout(CALLBACK_TIMEOUT)).accept(entry); } @Test @@ -67,7 +82,7 @@ void testAddCallbackInvokesFailureCallbackOnFail() { listenableFuture.addCallback(successCallback, failureCallback); listenableFuture.fail(entry); - verify(failureCallback).accept(entry); + verify(failureCallback, timeout(CALLBACK_TIMEOUT)).accept(entry); } @Test @@ -77,7 +92,7 @@ void testAddCallbackWithSuccessOnlyInvokesSuccessCallbackOnSuccess() { listenableFuture.addCallback(successCallback); listenableFuture.success(entry); - verify(successCallback).accept(entry); + verify(successCallback, timeout(CALLBACK_TIMEOUT)).accept(entry); } @Test @@ -96,7 +111,7 @@ void testSuccessDoesNotInvokeFailureCallback() { listenableFuture.addCallback(successCallback, failureCallback); listenableFuture.success(entry); - org.mockito.Mockito.verifyNoInteractions(failureCallback); + verifyNoInteractions(failureCallback); } @Test @@ -106,7 +121,7 @@ void testFailDoesNotInvokeSuccessCallback() { listenableFuture.addCallback(successCallback, failureCallback); listenableFuture.fail(entry); - org.mockito.Mockito.verifyNoInteractions(successCallback); + verifyNoInteractions(successCallback); } @Test diff --git a/amazon-sns-java-messaging-lib-v1/src/test/java/com/amazon/sns/messaging/lib/core/AmazonSnsTemplateIntegrationTest.java b/amazon-sns-java-messaging-lib-v1/src/test/java/com/amazon/sns/messaging/lib/core/AmazonSnsTemplateIntegrationTest.java index 6faa3f1..a17da9f 100644 --- a/amazon-sns-java-messaging-lib-v1/src/test/java/com/amazon/sns/messaging/lib/core/AmazonSnsTemplateIntegrationTest.java +++ b/amazon-sns-java-messaging-lib-v1/src/test/java/com/amazon/sns/messaging/lib/core/AmazonSnsTemplateIntegrationTest.java @@ -40,8 +40,8 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import org.testcontainers.containers.localstack.LocalStackContainer; -import org.testcontainers.containers.localstack.LocalStackContainer.Service; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; import org.testcontainers.utility.DockerImageName; @@ -79,12 +79,10 @@ class AmazonSnsTemplateIntegrationTest { @Container - static LocalStackContainer localstack = new LocalStackContainer(DockerImageName.parse("localstack/localstack:3.4.0")) - .withEnv("LOCALSTACK_HOST", "localhost") - .withEnv("SQS_ENDPOINT_STRATEGY", "off") + static GenericContainer ministack = new GenericContainer<>(DockerImageName.parse("ministackorg/ministack:1.4.0")) .withReuse(true) .withExposedPorts(4566) - .withServices(Service.SNS, Service.SQS) + .waitingFor(Wait.forLogMessage(".*Running on.*", 1)) .withCreateContainerCmdModifier(cmd -> cmd.getHostConfig() .withPortBindings(PortBinding.parse("4566:4566")) ); @@ -104,13 +102,13 @@ class AmazonSnsTemplateIntegrationTest { @BeforeAll static void setupClient() { snsClient = AmazonSNSClientBuilder.standard() - .withEndpointConfiguration(new EndpointConfiguration(localstack.getEndpoint().toString(), localstack.getRegion())) - .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(localstack.getAccessKey(), localstack.getSecretKey()))) + .withEndpointConfiguration(new EndpointConfiguration("http://localhost:4566", "sa-east-1")) + .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("fakeAccessKey", "fakeSecretKey"))) .build(); sqsClient = AmazonSQSClientBuilder.standard() - .withEndpointConfiguration(new EndpointConfiguration(localstack.getEndpoint().toString(), localstack.getRegion())) - .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(localstack.getAccessKey(), localstack.getSecretKey()))) + .withEndpointConfiguration(new EndpointConfiguration("http://localhost:4566", "sa-east-1")) + .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("fakeAccessKey", "fakeSecretKey"))) .build(); standardTopicArn = snsClient.createTopic("it-standard-topic").getTopicArn(); @@ -158,8 +156,8 @@ static void tearDownClient() { sqsClient.shutdown(); } - if (Objects.nonNull(localstack)) { - localstack.close(); + if (Objects.nonNull(ministack)) { + ministack.close(); } } @@ -220,9 +218,9 @@ private void countDownLatch(final Integer count, final Consumer void testSendSingleMessage() { final String messageBody = "hello-sqs-" + UUID.randomUUID(); - countDownLatch(1, countDownLatch -> { + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); + countDownLatch(1, countDownLatch -> { final String id = UUID.randomUUID().toString(); @@ -231,8 +229,6 @@ void testSendSingleMessage() { .withValue(messageBody) .build()); - template.await().thenRun(template::shutdown).join(); - future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), is(id)); @@ -248,17 +244,19 @@ void testSendSingleMessage() { final Message message = result.getMessages().get(0); assertThat(message.getBody(), is(messageBody)); assertThat(message.getMessageAttributes().keySet(), hasSize(0)); + + template.await().thenRun(template::shutdown).join(); } @Test void testSendMultipleMessages() { final int messageCount = 500; + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 50L, 10, 10); + countDownLatch(messageCount, countDownLatch -> { final List> futures = new ArrayList<>(); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 50L, 10, 10); - IntStream.range(0, messageCount).forEach(i -> { futures.add( template.send(RequestEntry.builder() @@ -268,8 +266,6 @@ void testSendMultipleMessages() { ); }); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -290,17 +286,19 @@ void testSendMultipleMessages() { assertThat(message.getBody(), containsString("msg-")); assertThat(message.getMessageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test void testSendMessagesExceedingBatchSize() { final int messageCount = 25; + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 50L, 10, 10); + countDownLatch(messageCount, countDownLatch -> { final List> futures = new ArrayList<>(); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 50L, 10, 10); - IntStream.range(0, messageCount).forEach(i -> { futures.add(template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) @@ -308,8 +306,6 @@ void testSendMessagesExceedingBatchSize() { .build())); }); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -330,18 +326,19 @@ void testSendMessagesExceedingBatchSize() { assertThat(message.getBody(), containsString("batch-test-")); assertThat(message.getMessageAttributes().keySet(), hasSize(0)); }); - } + template.await().thenRun(template::shutdown).join(); + } @Test void testSendMessagesWithLinger() { final int messageCount = 20; + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 200L, 10, 5); + countDownLatch(messageCount, countDownLatch -> { final List> futures = new ArrayList<>(); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 200L, 10, 5); - IntStream.range(0, messageCount).forEach(i -> { futures.add(template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) @@ -349,8 +346,6 @@ void testSendMessagesWithLinger() { .build())); }); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -371,27 +366,27 @@ void testSendMessagesWithLinger() { assertThat(message.getBody(), containsString("linger-test-")); assertThat(message.getMessageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test void testSendMessageWithgetMessageAttributes() { final String messageBody = "attr-test-" + UUID.randomUUID(); + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); + countDownLatch(1, countDownLatch -> { final Map messageHeaders = new HashMap<>(); messageHeaders.put("string-attr", "hello"); messageHeaders.put("number-attr", 42); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); - final ListenableFuture future = template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) .withValue(messageBody) .withMessageHeaders(messageHeaders) .build()); - template.await().thenRun(template::shutdown).join(); - future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -413,22 +408,23 @@ void testSendMessageWithgetMessageAttributes() { assertThat(message.getMessageAttributes().get("string-attr").getStringValue(), is("hello")); assertThat(message.getMessageAttributes().get("number-attr").getStringValue(), is("42")); }); + + template.await().thenRun(template::shutdown).join(); } @Test void testSendLargeMessage() { final String largeBody = RandomStringUtils.secure().nextAlphabetic(262_144); + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 200L, 5, 5); + countDownLatch(1, countDownLatch -> { - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 200L, 5, 5); final ListenableFuture future = template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) .withValue(largeBody) .build()); - template.await().thenRun(template::shutdown).join(); - future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -449,10 +445,14 @@ void testSendLargeMessage() { assertThat(message.getBody(), is(largeBody)); assertThat(message.getMessageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test void testSendMessageExceedingMaxSize() { + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); + countDownLatch(1, countDownLatch -> { final String oversizedBody = RandomStringUtils.secure().nextAlphabetic((1024 * 256) + 1); @@ -461,12 +461,8 @@ void testSendMessageExceedingMaxSize() { .withValue(oversizedBody) .build(); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); - final ListenableFuture future = template.send(entry); - template.await().thenRun(template::shutdown).join(); - future.addCallback(null, failureResult -> { assertThat(failureResult.getCode(), is("000")); assertThat(failureResult.getId(), is(entry.getId())); @@ -479,14 +475,17 @@ void testSendMessageExceedingMaxSize() { final List messages = receiveMessage(standardQueueUrl, 10, 5).getMessages(); assertThat(messages, hasSize(0)); + + template.await().thenRun(template::shutdown).join(); } @Test void testShutdownDrainsPendingMessages() { final int messageCount = 5; + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 10_000L, 10, 5); + countDownLatch(messageCount, countDownLatch -> { - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 10_000L, 10, 5); final List> futures = new ArrayList<>(); @@ -497,8 +496,6 @@ void testShutdownDrainsPendingMessages() { .build())); }); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -519,20 +516,21 @@ void testShutdownDrainsPendingMessages() { assertThat(message.getBody(), containsString("drain-test-")); assertThat(message.getMessageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test void testTemplateLifecycle() { + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); + countDownLatch(1, countDownLatch -> { - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); final ListenableFuture future = template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) .withValue("lifecycle-" + UUID.randomUUID()) .build()); - template.await().thenRun(template::shutdown).join(); - future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -553,6 +551,8 @@ void testTemplateLifecycle() { assertThat(message.getBody(), containsString("lifecycle-")); assertThat(message.getMessageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test @@ -561,8 +561,9 @@ void testSendSingleFifoMessage() { final String id = UUID.randomUUID().toString(); final String groupId = id; + final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 100L, 10, 1); + countDownLatch(1, countDownLatch -> { - final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 100L, 10, 1); final ListenableFuture future = template.send(RequestEntry.builder() .withId(id) @@ -570,8 +571,6 @@ void testSendSingleFifoMessage() { .withGroupId(groupId) .build()); - template.await().thenRun(template::shutdown).join(); - future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), is(id)); @@ -594,6 +593,8 @@ void testSendSingleFifoMessage() { assertThat(message.getAttributes().get(MessageSystemAttributeName.MessageGroupId.toString()), is(groupId)); assertThat(message.getMessageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test @@ -601,11 +602,11 @@ void testSendFifoMessagesWithOrdering() { final int messageCount = 100; final String groupId = UUID.randomUUID().toString(); + final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 50L, 10, 1); + countDownLatch(1, countDownLatch -> { final List> futures = new ArrayList<>(); - final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 50L, 10, 1); - IntStream.range(0, messageCount).forEach(i -> { futures.add(template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) @@ -614,8 +615,6 @@ void testSendFifoMessagesWithOrdering() { .build())); }); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -638,6 +637,8 @@ void testSendFifoMessagesWithOrdering() { assertThat(message.getAttributes().get(MessageSystemAttributeName.MessageGroupId.toString()), is(groupId)); assertThat(message.getMessageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test @@ -646,11 +647,11 @@ void testSendFifoMessageWithDeduplication() { final String groupId = UUID.randomUUID().toString(); final String messageBody = "dedup-test-" + UUID.randomUUID(); + final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 100L, 10, 1); + countDownLatch(1, countDownLatch -> { final List> futures = new ArrayList<>(); - final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 100L, 10, 1); - futures.add(template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) .withValue(messageBody) @@ -665,8 +666,6 @@ void testSendFifoMessageWithDeduplication() { .withDeduplicationId(deduplicationId) .build())); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -690,6 +689,8 @@ void testSendFifoMessageWithDeduplication() { assertThat(message.getAttributes().get(MessageSystemAttributeName.MessageDeduplicationId.toString()), is(deduplicationId)); assertThat(message.getMessageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } } diff --git a/amazon-sns-java-messaging-lib-v2/src/test/java/com/amazon/sns/messaging/lib/core/AmazonSnsTemplateIntegrationTest.java b/amazon-sns-java-messaging-lib-v2/src/test/java/com/amazon/sns/messaging/lib/core/AmazonSnsTemplateIntegrationTest.java index 0f87c1d..c265344 100644 --- a/amazon-sns-java-messaging-lib-v2/src/test/java/com/amazon/sns/messaging/lib/core/AmazonSnsTemplateIntegrationTest.java +++ b/amazon-sns-java-messaging-lib-v2/src/test/java/com/amazon/sns/messaging/lib/core/AmazonSnsTemplateIntegrationTest.java @@ -42,8 +42,8 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import org.testcontainers.containers.localstack.LocalStackContainer; -import org.testcontainers.containers.localstack.LocalStackContainer.Service; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; import org.testcontainers.utility.DockerImageName; @@ -73,12 +73,10 @@ class AmazonSnsTemplateIntegrationTest { @Container - static LocalStackContainer localstack = new LocalStackContainer(DockerImageName.parse("localstack/localstack:3.4.0")) - .withEnv("LOCALSTACK_HOST", "localhost") - .withEnv("SQS_ENDPOINT_STRATEGY", "off") + static GenericContainer ministack = new GenericContainer<>(DockerImageName.parse("ministackorg/ministack:1.4.0")) .withReuse(true) .withExposedPorts(4566) - .withServices(Service.SNS, Service.SQS) + .waitingFor(Wait.forLogMessage(".*Running on.*", 1)) .withCreateContainerCmdModifier(cmd -> cmd.getHostConfig() .withPortBindings(PortBinding.parse("4566:4566")) ); @@ -98,15 +96,15 @@ class AmazonSnsTemplateIntegrationTest { @BeforeAll static void setupClient() { snsClient = SnsClient.builder() - .endpointOverride(URI.create(localstack.getEndpoint().toString())) - .region(Region.of(localstack.getRegion())) - .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create(localstack.getAccessKey(), localstack.getSecretKey()))) + .endpointOverride(URI.create("http://localhost:4566")) + .region(Region.of("sa-east")) + .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("fakeAccessKey", "fakeSecretKey"))) .build(); sqsClient = SqsClient.builder() - .endpointOverride(localstack.getEndpoint()) - .region(Region.of(localstack.getRegion())) - .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create(localstack.getAccessKey(), localstack.getSecretKey()))) + .endpointOverride(URI.create("http://localhost:4566")) + .region(Region.of("sa-east")) + .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("fakeAccessKey", "fakeSecretKey"))) .build(); standardTopicArn = snsClient.createTopic(request -> request.name("it-standard-topic")).topicArn(); @@ -158,8 +156,8 @@ static void tearDownClient() { sqsClient.close(); } - if (Objects.nonNull(localstack)) { - localstack.close(); + if (Objects.nonNull(ministack)) { + ministack.close(); } } @@ -220,9 +218,9 @@ private void countDownLatch(final Integer count, final Consumer void testSendSingleMessage() { final String messageBody = "hello-sqs-" + UUID.randomUUID(); - countDownLatch(1, countDownLatch -> { + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); + countDownLatch(1, countDownLatch -> { final String id = UUID.randomUUID().toString(); @@ -231,7 +229,6 @@ void testSendSingleMessage() { .withValue(messageBody) .build()); - template.await().thenRun(template::shutdown).join(); future.addCallback(result -> { assertThat(result, notNullValue()); @@ -248,17 +245,19 @@ void testSendSingleMessage() { final Message message = result.messages().get(0); assertThat(message.body(), is(messageBody)); assertThat(message.messageAttributes().keySet(), hasSize(0)); + + template.await().thenRun(template::shutdown).join(); } @Test void testSendMultipleMessages() { final int messageCount = 500; + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 50L, 10, 10); + countDownLatch(messageCount, countDownLatch -> { final List> futures = new ArrayList<>(); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 50L, 10, 10); - IntStream.range(0, messageCount).forEach(i -> { futures.add( template.send(RequestEntry.builder() @@ -268,8 +267,6 @@ void testSendMultipleMessages() { ); }); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -290,17 +287,19 @@ void testSendMultipleMessages() { assertThat(message.body(), containsString("msg-")); assertThat(message.messageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test void testSendMessagesExceedingBatchSize() { final int messageCount = 25; + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 50L, 10, 10); + countDownLatch(messageCount, countDownLatch -> { final List> futures = new ArrayList<>(); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 50L, 10, 10); - IntStream.range(0, messageCount).forEach(i -> { futures.add(template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) @@ -308,8 +307,6 @@ void testSendMessagesExceedingBatchSize() { .build())); }); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -330,18 +327,19 @@ void testSendMessagesExceedingBatchSize() { assertThat(message.body(), containsString("batch-test-")); assertThat(message.messageAttributes().keySet(), hasSize(0)); }); - } + template.await().thenRun(template::shutdown).join(); + } @Test void testSendMessagesWithLinger() { final int messageCount = 20; + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 200L, 10, 5); + countDownLatch(messageCount, countDownLatch -> { final List> futures = new ArrayList<>(); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 200L, 10, 5); - IntStream.range(0, messageCount).forEach(i -> { futures.add(template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) @@ -349,8 +347,6 @@ void testSendMessagesWithLinger() { .build())); }); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -371,27 +367,27 @@ void testSendMessagesWithLinger() { assertThat(message.body(), containsString("linger-test-")); assertThat(message.messageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test void testSendMessageWithgetMessageAttributes() { final String messageBody = "attr-test-" + UUID.randomUUID(); + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); + countDownLatch(1, countDownLatch -> { final Map messageHeaders = new HashMap<>(); messageHeaders.put("string-attr", "hello"); messageHeaders.put("number-attr", 42); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); - final ListenableFuture future = template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) .withValue(messageBody) .withMessageHeaders(messageHeaders) .build()); - template.await().thenRun(template::shutdown).join(); - future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -413,22 +409,23 @@ void testSendMessageWithgetMessageAttributes() { assertThat(message.messageAttributes().get("string-attr").stringValue(), is("hello")); assertThat(message.messageAttributes().get("number-attr").stringValue(), is("42")); }); + + template.await().thenRun(template::shutdown).join(); } @Test void testSendLargeMessage() { final String largeBody = RandomStringUtils.secure().nextAlphabetic(262_144); + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 200L, 5, 5); + countDownLatch(1, countDownLatch -> { - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 200L, 5, 5); final ListenableFuture future = template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) .withValue(largeBody) .build()); - template.await().thenRun(template::shutdown).join(); - future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -449,10 +446,15 @@ void testSendLargeMessage() { assertThat(message.body(), is(largeBody)); assertThat(message.messageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test void testSendMessageExceedingMaxSize() { + + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); + countDownLatch(1, countDownLatch -> { final String oversizedBody = RandomStringUtils.secure().nextAlphabetic((1024 * 256) + 1); @@ -461,12 +463,8 @@ void testSendMessageExceedingMaxSize() { .withValue(oversizedBody) .build(); - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); - final ListenableFuture future = template.send(entry); - template.await().thenRun(template::shutdown).join(); - future.addCallback(null, failureResult -> { assertThat(failureResult.getCode(), is("000")); assertThat(failureResult.getId(), is(entry.getId())); @@ -479,14 +477,17 @@ void testSendMessageExceedingMaxSize() { final List messages = receiveMessage(standardQueueUrl, 10, 5).messages(); assertThat(messages, hasSize(0)); + + template.await().thenRun(template::shutdown).join(); } @Test void testShutdownDrainsPendingMessages() { final int messageCount = 5; + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 10_000L, 10, 5); + countDownLatch(messageCount, countDownLatch -> { - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 10_000L, 10, 5); final List> futures = new ArrayList<>(); @@ -497,8 +498,6 @@ void testShutdownDrainsPendingMessages() { .build())); }); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -519,20 +518,21 @@ void testShutdownDrainsPendingMessages() { assertThat(message.body(), containsString("drain-test-")); assertThat(message.messageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test void testTemplateLifecycle() { + final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); + countDownLatch(1, countDownLatch -> { - final AmazonSnsTemplate template = createTemplate(standardTopicArn, false, 100L, 10, 5); final ListenableFuture future = template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) .withValue("lifecycle-" + UUID.randomUUID()) .build()); - template.await().thenRun(template::shutdown).join(); - future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -553,6 +553,8 @@ void testTemplateLifecycle() { assertThat(message.body(), containsString("lifecycle-")); assertThat(message.messageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test @@ -561,17 +563,15 @@ void testSendSingleFifoMessage() { final String id = UUID.randomUUID().toString(); final String groupId = id; - countDownLatch(1, countDownLatch -> { - final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 100L, 10, 1); + final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 100L, 10, 1); + countDownLatch(1, countDownLatch -> { final ListenableFuture future = template.send(RequestEntry.builder() .withId(id) .withValue(messageBody) .withGroupId(groupId) .build()); - template.await().thenRun(template::shutdown).join(); - future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), is(id)); @@ -594,6 +594,8 @@ void testSendSingleFifoMessage() { assertThat(message.attributes().get(MessageSystemAttributeName.MESSAGE_GROUP_ID), is(groupId)); assertThat(message.messageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test @@ -601,11 +603,11 @@ void testSendFifoMessagesWithOrdering() { final int messageCount = 100; final String groupId = UUID.randomUUID().toString(); + final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 50L, 10, 1); + countDownLatch(1, countDownLatch -> { final List> futures = new ArrayList<>(); - final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 50L, 10, 1); - IntStream.range(0, messageCount).forEach(i -> { futures.add(template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) @@ -614,8 +616,6 @@ void testSendFifoMessagesWithOrdering() { .build())); }); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -638,6 +638,8 @@ void testSendFifoMessagesWithOrdering() { assertThat(message.attributes().get(MessageSystemAttributeName.MESSAGE_GROUP_ID), is(groupId)); assertThat(message.messageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } @Test @@ -646,11 +648,11 @@ void testSendFifoMessageWithDeduplication() { final String groupId = UUID.randomUUID().toString(); final String messageBody = "dedup-test-" + UUID.randomUUID(); + final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 100L, 10, 1); + countDownLatch(1, countDownLatch -> { final List> futures = new ArrayList<>(); - final AmazonSnsTemplate template = createTemplate(fifoTopicArn, true, 100L, 10, 1); - futures.add(template.send(RequestEntry.builder() .withId(UUID.randomUUID().toString()) .withValue(messageBody) @@ -665,8 +667,6 @@ void testSendFifoMessageWithDeduplication() { .withDeduplicationId(deduplicationId) .build())); - template.await().thenRun(template::shutdown).join(); - futures.forEach(future -> future.addCallback(result -> { assertThat(result, notNullValue()); assertThat(result.getId(), notNullValue()); @@ -690,6 +690,8 @@ void testSendFifoMessageWithDeduplication() { assertThat(message.attributes().get(MessageSystemAttributeName.MESSAGE_DEDUPLICATION_ID), is(deduplicationId)); assertThat(message.messageAttributes().keySet(), hasSize(0)); }); + + template.await().thenRun(template::shutdown).join(); } }