feat: Add IterableAsyncQueue.

Author: kinyoklion

lib/shared/common/build.gradle.kts

@@ -33,8 +33,8 @@ base {
 java {
     withJavadocJar()
     withSourcesJar()
-    sourceCompatibility = JavaVersion.VERSION_1_7
-    targetCompatibility = JavaVersion.VERSION_1_7
+    sourceCompatibility = JavaVersion.VERSION_1_8
+    targetCompatibility = JavaVersion.VERSION_1_8
 }
 
 // See Dependencies.kt in buildSrc for the purpose of "privateImplementation"

lib/shared/common/src/main/java/com/launchdarkly/sdk/collections/IterableAsyncQueue.java

@@ -0,0 +1,64 @@
+package com.launchdarkly.sdk.collections;
+
+import java.util.LinkedList;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A thread-safe unbounded queue that provides asynchronous consumption via {@link CompletableFuture}.
+ * <p>
+ * This queue supports multiple concurrent producers and consumers. Items are delivered in FIFO order.
+ * The {@link #take()} method returns a {@link CompletableFuture} that either completes immediately
+ * if an item is available, or completes later when an item is added via {@link #put(Object)}.
+ * <p>
+ * When multiple consumers are waiting (i.e., multiple pending {@link #take()} calls), they are
+ * satisfied in FIFO order as items become available.
+ * <p>
+ * Null values are supported.
+ *
+ * @param <T> the type of elements held in this queue
+ */
+class IterableAsyncQueue<T> {
+    private final Object lock = new Object();
+    private final LinkedList<T> queue = new LinkedList<>();
+
+    private final LinkedList<CompletableFuture<T>> pendingFutures = new LinkedList<>();
+
+    /**
+     * Adds an item to the queue.
+     * <p>
+     * If there is a consumer is waiting (a pending {@link #take()} call), the item is delivered
+     * directly to the oldest waiting consumer's future. Otherwise, the item is added to the
+     * queue for later consumption.
+     *
+     * @param item the item to add (maybe null)
+     */
+    public void put(T item) {
+        synchronized (lock) {
+            CompletableFuture<T> nextFuture = pendingFutures.pollFirst();
+            if(nextFuture != null) {
+                nextFuture.complete(item);
+                return;
+            }
+            queue.addLast(item);
+        }
+    }
+    /**
+     * Retrieves and removes an item from the queue, returning a future that completes with the item.
+     * <p>
+     * If the queue contains items, returns an already-completed future with the oldest item.
+     * If the queue is empty, returns a future that will complete when an item becomes available
+     * via {@link #put(Object)}.
+     *
+     * @return a {@link CompletableFuture} that completes with the next item
+     */
+    public CompletableFuture<T> take() {
+        synchronized (lock) {
+            if(!queue.isEmpty()) {
+                return CompletableFuture.completedFuture(queue.removeFirst());
+            }
+            CompletableFuture<T> takeFuture = new CompletableFuture<>();
+            pendingFutures.addLast(takeFuture);
+            return takeFuture;
+        }
+    }
+}