Java: async getPubSubMessage.

.github/workflows/java.yml

@@ -126,7 +126,7 @@ jobs:
                     - 17
         runs-on: ubuntu-latest
         container: amazonlinux:latest
-        timeout-minutes: 20
+        timeout-minutes: 35
         steps:
             - name: Install git
               run: |

java/client/src/main/java/glide/api/BaseClient.java

@@ -252,6 +252,7 @@
 import glide.connectors.handlers.CallbackDispatcher;
 import glide.connectors.handlers.ChannelHandler;
 import glide.connectors.handlers.MessageHandler;
+import glide.connectors.handlers.PubSubMessageQueue;
 import glide.connectors.resources.Platform;
 import glide.connectors.resources.ThreadPoolResource;
 import glide.connectors.resources.ThreadPoolResourceAllocator;
@@ -267,14 +268,12 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.ArrayUtils;
-import org.apache.commons.lang3.NotImplementedException;
 import response.ResponseOuterClass.ConstantResponse;
 import response.ResponseOuterClass.Response;
 
@@ -300,7 +299,7 @@ public abstract class BaseClient
 
     protected final CommandManager commandManager;
     protected final ConnectionManager connectionManager;
-    protected final ConcurrentLinkedDeque<PubSubMessage> messageQueue;
+    protected final PubSubMessageQueue messageQueue;
     protected final Optional<BaseSubscriptionConfiguration> subscriptionConfiguration;
 
     /** Helper which extracts data from received {@link Response}s from GLIDE. */
@@ -324,7 +323,7 @@ protected BaseClient(ClientBuilder builder) {
     protected static class ClientBuilder {
         private final ConnectionManager connectionManager;
         private final CommandManager commandManager;
-        private final ConcurrentLinkedDeque<PubSubMessage> messageQueue;
+        private final PubSubMessageQueue messageQueue;
         private final Optional<BaseSubscriptionConfiguration> subscriptionConfiguration;
     }
 
@@ -368,7 +367,7 @@ protected static <T extends BaseClient> CompletableFuture<T> CreateClient(
     }
 
     /**
-     * Tries to return a next pubsub message.
+     * Return a next pubsub message if it is present.
      *
      * @throws ConfigurationError If client is not subscribed to any channel or if client configured
      *     with a callback.
@@ -385,16 +384,16 @@ public PubSubMessage tryGetPubSubMessage() {
                     "The operation will never complete since messages will be passed to the configured"
                             + " callback.");
         }
-        return messageQueue.poll();
+        return messageQueue.popSync();
     }
 
     /**
-     * Returns a promise for a next pubsub message.
+     * Returns a promise for a next pubsub message.<br>
+     * Message gets unrecoverable lost if future is cancelled or reference to this future is lost.
      *
-     * @apiNote <b>Not implemented!</b>
      * @throws ConfigurationError If client is not subscribed to any channel or if client configured
      *     with a callback.
-     * @return A <code>Future</code> which resolved with the next incoming message.
+     * @return A {@link CompletableFuture} which will asynchronously hold the next available message.
      */
     public CompletableFuture<PubSubMessage> getPubSubMessage() {
         if (subscriptionConfiguration.isEmpty()) {
@@ -407,8 +406,7 @@ public CompletableFuture<PubSubMessage> getPubSubMessage() {
                     "The operation will never complete since messages will be passed to the configured"
                             + " callback.");
         }
-        throw new NotImplementedException(
-                "This feature will be supported in a future release of the GLIDE java client");
+        return messageQueue.popAsync();
     }
 
     /**

java/client/src/main/java/glide/api/models/PubSubMessage.java

@@ -32,10 +32,10 @@ public PubSubMessage(String message, String channel) {
 
     @Override
     public String toString() {
-        String res = String.format("%s, channel = %s", message, channel);
+        String res = String.format("(%s, channel = %s", message, channel);
         if (pattern.isPresent()) {
             res += ", pattern = " + pattern.get();
         }
-        return res;
+        return res + ")";
     }
 }

java/client/src/main/java/glide/connectors/handlers/MessageHandler.java

@@ -9,7 +9,6 @@
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.stream.Collectors;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
@@ -34,7 +33,7 @@ public class MessageHandler {
     private final BaseResponseResolver responseResolver;
 
     /** A message queue wrapper. */
-    private final ConcurrentLinkedDeque<PubSubMessage> queue = new ConcurrentLinkedDeque<>();
+    private final PubSubMessageQueue queue = new PubSubMessageQueue();
 
     /** Process a push (PUBSUB) message received as a part of {@link Response} from GLIDE. */
     public void handle(Response response) {

java/client/src/main/java/glide/connectors/handlers/PubSubMessageQueue.java

@@ -0,0 +1,66 @@
+/** Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0 */
+package glide.connectors.handlers;
+
+import glide.api.models.PubSubMessage;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * An asynchronous FIFO message queue for {@link PubSubMessage} backed by {@link
+ * ConcurrentLinkedDeque}.
+ */
+public class PubSubMessageQueue {
+    // fields are protected to ease testing
+    /** The queue itself. */
+    protected final ConcurrentLinkedDeque<PubSubMessage> messageQueue = new ConcurrentLinkedDeque<>();
+
+    /**
+     * A promise for the first incoming message. Returned to a user, if message queried in async
+     * manner, but nothing received yet.
+     */
+    protected CompletableFuture<PubSubMessage> firstMessagePromise = new CompletableFuture<>();
+
+    /** A flag whether a user already got a {@link #firstMessagePromise}. */
+    private final AtomicBoolean firstMessagePromiseRequested = new AtomicBoolean(false);
+
+    /** A private object used to synchronize {@link #push} and {@link #popAsync}. */
+    private final Object lock = new Object();
+
+    // TODO Rework to remove or reduce `synchronized` blocks. If remove it now, some messages get
+    // reordered.
+
+    /** Store a new message. */
+    public void push(PubSubMessage message) {
+        synchronized (lock) {
+            if (firstMessagePromiseRequested.getAndSet(false)) {
+                firstMessagePromise.complete(message);
+                firstMessagePromise = new CompletableFuture<>();
+                return;
+            }
+
+            messageQueue.addLast(message);
+        }
+    }
+
+    /** Get a promise for a next message. */
+    public synchronized CompletableFuture<PubSubMessage> popAsync() {
+        synchronized (lock) {
+            PubSubMessage message = messageQueue.poll();
+            if (message == null) {
+                // this makes first incoming message to be delivered into `firstMessagePromise` instead of
+                // `messageQueue`
+                firstMessagePromiseRequested.set(true);
+                return firstMessagePromise;
+            }
+            var future = new CompletableFuture<PubSubMessage>();
+            future.complete(message);
+            return future;
+        }
+    }
+
+    /** Get a new message or null if nothing stored so far. */
+    public PubSubMessage popSync() {
+        return messageQueue.poll();
+    }
+}