Auto-created pull request into develop from feature/support-java-21-virtual-threads

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/concurrent/RingBufferBlockingQueue.java

@@ -29,186 +29,177 @@
 import java.util.stream.IntStream;
 
 import lombok.Getter;
+import lombok.Locked;
 import lombok.Setter;
 import lombok.SneakyThrows;
 
 public class RingBufferBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E> {
-
+  
   private static final int DEFAULT_CAPACITY = 2048;
-
+  
   private final AtomicReferenceArray<Entry<E>> buffer;
-
+  
   private final int capacity;
-
+  
   private final AtomicLong writeSequence = new AtomicLong(-1);
-
+  
   private final AtomicLong readSequence = new AtomicLong(0);
-
+  
   private final AtomicInteger size = new AtomicInteger(0);
-
+  
   private final ReentrantLock reentrantLock;
-
+  
   private final Condition waitingConsumer;
-
+  
   private final Condition waitingProducer;
-
+  
   public RingBufferBlockingQueue(final int capacity) {
     this.capacity = capacity;
-    this.buffer = new AtomicReferenceArray<>(capacity);
-    this.reentrantLock = new ReentrantLock(true);
-    this.waitingConsumer = this.reentrantLock.newCondition();
-    this.waitingProducer = this.reentrantLock.newCondition();
-    IntStream.range(0, capacity).forEach(idx -> this.buffer.set(idx, new Entry<>()));
+    buffer = new AtomicReferenceArray<>(capacity);
+    reentrantLock = new ReentrantLock(true);
+    waitingConsumer = reentrantLock.newCondition();
+    waitingProducer = reentrantLock.newCondition();
+    IntStream.range(0, capacity).forEach(idx -> buffer.set(idx, new Entry<>()));
   }
-
+  
   public RingBufferBlockingQueue() {
     this(RingBufferBlockingQueue.DEFAULT_CAPACITY);
   }
-
+  
   private long avoidSequenceOverflow(final long sequence) {
     return (sequence < Long.MAX_VALUE ? sequence : wrap(sequence));
   }
-
+  
   private int wrap(final long sequence) {
-    return Math.toIntExact(sequence % this.capacity);
+    return Math.toIntExact(sequence % capacity);
   }
-  
+
   public int capacity() {
-    return this.capacity;
+    return capacity;
   }
-  
+
   @Override
   public int size() {
-    return this.size.get();
+    return size.get();
   }
-
+  
   @Override
   public boolean isEmpty() {
-    return this.size.get() == 0;
+    return size.get() == 0;
   }
-
+  
   public boolean isFull() {
-    return this.size.get() >= this.capacity;
+    return size.get() >= capacity;
   }
-
+  
   public long writeSequence() {
-    return this.writeSequence.get();
+    return writeSequence.get();
   }
-
+  
   public long readSequence() {
-    return this.readSequence.get();
+    return readSequence.get();
   }
-
+  
   @Override
   public E peek() {
-    return isEmpty() ? null : this.buffer.get(wrap(this.readSequence.get())).getValue();
+    return isEmpty() ? null : buffer.get(wrap(readSequence.get())).getValue();
   }
-
+  
   @Override
   @SneakyThrows
+  @Locked("reentrantLock")
   public void put(final E element) {
-    try {
-      reentrantLock.lock();
-
-      while (isFull()) {
-        waitingProducer.await();
-      }
+    while (isFull()) {
+      waitingProducer.await();
+    }
 
-      final long prevWriteSeq = writeSequence.get();
-      final long nextWriteSeq = avoidSequenceOverflow(prevWriteSeq) + 1;
+    final long prevWriteSeq = writeSequence.get();
+    final long nextWriteSeq = avoidSequenceOverflow(prevWriteSeq) + 1;
 
-      buffer.get(wrap(nextWriteSeq)).setValue(element);
+    buffer.get(wrap(nextWriteSeq)).setValue(element);
 
-      writeSequence.compareAndSet(prevWriteSeq, nextWriteSeq);
+    writeSequence.compareAndSet(prevWriteSeq, nextWriteSeq);
 
-      size.incrementAndGet();
+    size.incrementAndGet();
 
-      waitingConsumer.signal();
-    } finally {
-      reentrantLock.unlock();
-    }
+    waitingConsumer.signal();
   }
-
+  
   @Override
   @SneakyThrows
+  @Locked("reentrantLock")
   public E take() {
-    try {
-      reentrantLock.lock();
-
-      while (isEmpty()) {
-        waitingConsumer.await();
-      }
-
-      final long prevReadSeq = readSequence.get();
-      final long nextReadSeq = avoidSequenceOverflow(prevReadSeq) + 1;
-
-      final E nextValue = buffer.get(wrap(prevReadSeq)).getValue();
-
-      buffer.get(wrap(prevReadSeq)).setValue(null);
-
-      readSequence.compareAndSet(prevReadSeq, nextReadSeq);
-
-      size.decrementAndGet();
-
-      waitingProducer.signal();
-
-      return nextValue;
-    } finally {
-      reentrantLock.unlock();
+    while (isEmpty()) {
+      waitingConsumer.await();
     }
+
+    final long prevReadSeq = readSequence.get();
+    final long nextReadSeq = avoidSequenceOverflow(prevReadSeq) + 1;
+
+    final E nextValue = buffer.get(wrap(prevReadSeq)).getValue();
+
+    buffer.get(wrap(prevReadSeq)).setValue(null);
+
+    readSequence.compareAndSet(prevReadSeq, nextReadSeq);
+
+    size.decrementAndGet();
+
+    waitingProducer.signal();
+
+    return nextValue;
   }
-
+  
   @Override
   public boolean offer(final E element) {
     throw new UnsupportedOperationException();
   }
-
+  
   @Override
   public boolean offer(final E element, final long timeout, final TimeUnit unit) throws InterruptedException {
     throw new UnsupportedOperationException();
   }
-
+  
   @Override
   public E poll() {
     throw new UnsupportedOperationException();
   }
-
+  
   @Override
   public E poll(final long timeout, final TimeUnit unit) throws InterruptedException {
     throw new UnsupportedOperationException();
   }
-
+  
   @Override
   public Iterator<E> iterator() {
     throw new UnsupportedOperationException();
   }
-
+  
   @Override
   public boolean add(final E element) {
     throw new UnsupportedOperationException();
   }
-
+  
   @Override
   public int remainingCapacity() {
     throw new UnsupportedOperationException();
   }
-
+  
   @Override
   public int drainTo(final Collection<? super E> collection) {
     throw new UnsupportedOperationException();
   }
-
+  
   @Override
   public int drainTo(final Collection<? super E> collection, final int maxElements) {
     throw new UnsupportedOperationException();
   }
-
+  
   @Getter
   @Setter
   static class Entry<E> {
-
+    
     private E value;
-
+    
   }
-
+  
 }

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsConsumer.java

@@ -17,6 +17,7 @@
 package com.amazon.sns.messaging.lib.core;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Objects;
@@ -29,6 +30,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
 import java.util.function.BiFunction;
 import java.util.function.UnaryOperator;
 
@@ -41,6 +43,8 @@
 import com.amazon.sns.messaging.lib.core.RequestEntryInternalFactory.RequestEntryInternal;
 import com.amazon.sns.messaging.lib.model.PublishRequestBuilder;
 import com.amazon.sns.messaging.lib.model.RequestEntry;
+import com.amazon.sns.messaging.lib.model.ResponseFailEntry;
+import com.amazon.sns.messaging.lib.model.ResponseSuccessEntry;
 import com.amazon.sns.messaging.lib.model.TopicProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -63,7 +67,7 @@ abstract class AbstractAmazonSnsConsumer<C, R, O, E> implements Runnable {
 
   private final RequestEntryInternalFactory requestEntryInternalFactory;
 
-  protected final ConcurrentMap<String, ListenableFutureRegistry> pendingRequests;
+  protected final ConcurrentMap<String, ListenableFuture<ResponseSuccessEntry, ResponseFailEntry>> pendingRequests;
 
   private final BlockingQueue<RequestEntry<E>> topicRequests;
 
@@ -75,7 +79,7 @@ protected AbstractAmazonSnsConsumer(
       final C amazonSnsClient,
       final TopicProperty topicProperty,
       final ObjectMapper objectMapper,
-      final ConcurrentMap<String, ListenableFutureRegistry> pendingRequests,
+      final ConcurrentMap<String, ListenableFuture<ResponseSuccessEntry, ResponseFailEntry>> pendingRequests,
       final BlockingQueue<RequestEntry<E>> topicRequests,
       final ExecutorService executorService,
       final UnaryOperator<R> publishDecorator) {
@@ -216,18 +220,11 @@ private Optional<R> createBatch(final BlockingQueue<RequestEntry<E>> requests) {
   @SneakyThrows
   public CompletableFuture<Void> await() {
     return CompletableFuture.runAsync(() -> {
-      while (
-        MapUtils.isNotEmpty(this.pendingRequests) ||
-        CollectionUtils.isNotEmpty(this.topicRequests)) {
-        sleep(topicProperty.getLinger());
+      while (MapUtils.isNotEmpty(this.pendingRequests) || CollectionUtils.isNotEmpty(this.topicRequests)) {
+        LockSupport.parkNanos(Duration.ofMillis(topicProperty.getLinger()).toNanos());
       }
     });
   }
 
-  @SneakyThrows
-  private static void sleep(final long millis) {
-    Thread.sleep(millis);
-  }
-
 }
 // @formatter:on

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/AbstractAmazonSnsProducer.java

@@ -18,7 +18,6 @@
 
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -40,15 +39,15 @@ abstract class AbstractAmazonSnsProducer<E> {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(AbstractAmazonSnsProducer.class);
 
-  private final ConcurrentMap<String, ListenableFutureRegistry> pendingRequests;
+  private final ConcurrentMap<String, ListenableFuture<ResponseSuccessEntry, ResponseFailEntry>> pendingRequests;
 
   private final BlockingQueue<RequestEntry<E>> topicRequests;
 
   private final ExecutorService executorService;
 
   @SneakyThrows
   public ListenableFuture<ResponseSuccessEntry, ResponseFailEntry> send(final RequestEntry<E> requestEntry) {
-    return CompletableFuture.supplyAsync(() -> enqueueRequest(requestEntry), executorService).get();
+    return enqueueRequest(requestEntry);
   }
 
   @SneakyThrows
@@ -65,7 +64,7 @@ public void shutdown() {
 
   @SneakyThrows
   private ListenableFuture<ResponseSuccessEntry, ResponseFailEntry> enqueueRequest(final RequestEntry<E> requestEntry) {
-    final ListenableFutureRegistry trackPendingRequest = new ListenableFutureRegistry();
+    final ListenableFuture<ResponseSuccessEntry, ResponseFailEntry> trackPendingRequest = new ListenableFutureImpl();
     pendingRequests.put(requestEntry.getId(), trackPendingRequest);
     topicRequests.put(requestEntry);
     return trackPendingRequest;

amazon-sns-java-messaging-lib-template/src/main/java/com/amazon/sns/messaging/lib/core/ListenableFuture.java

@@ -18,6 +18,9 @@
 
 import java.util.function.Consumer;
 
+import com.amazon.sns.messaging.lib.model.ResponseFailEntry;
+import com.amazon.sns.messaging.lib.model.ResponseSuccessEntry;
+
 // @formatter:off
 public interface ListenableFuture<S, F> {
 
@@ -27,5 +30,9 @@ default void addCallback(final Consumer<? super S> successCallback) {
     addCallback(successCallback, result -> { });
   }
 
+  void success(final ResponseSuccessEntry entry);
+
+  void fail(final ResponseFailEntry entry);
+
 }
 // @formatter:on