Cache programmatic API - support asynchronous value loader

- resolves #30311

extensions/cache/deployment/src/test/java/io/quarkus/cache/test/runtime/ProgrammaticApiTest.java

@@ -29,6 +29,7 @@
 import io.quarkus.cache.CaffeineCache;
 import io.quarkus.cache.runtime.caffeine.CaffeineCacheImpl;
 import io.quarkus.test.QuarkusUnitTest;
+import io.smallrye.mutiny.Uni;
 
 public class ProgrammaticApiTest {
 
@@ -212,10 +213,25 @@ public void testPutShouldPopulateCache() {
         }
     }
 
+    @Test
+    public void testAsyncLoader() throws Exception {
+        // Action: value retrieval from the cache.
+        // Expected effect: asyncvalue loader function used.
+        // Verified by: same object reference between STEPS 1 and 2 results.
+        String key = "alpha";
+        String result = "foo";
+        Uni<String> uni = Uni.createFrom().item(result);
+        String value = cache.getAsync(key, k -> uni).await().indefinitely();
+        assertSame(value, result);
+        assertKeySetContains(key);
+        assertGetIfPresent(key, value);
+
+    }
+
     private void assertKeySetContains(Object... expectedKeys) {
         Set<Object> expectedKeySet = new HashSet<>(Arrays.asList(expectedKeys));
         Set<Object> actualKeySet = cache.as(CaffeineCache.class).keySet();
-        assertEquals(expectedKeySet, actualKeySet);
+        assertTrue(actualKeySet.containsAll(expectedKeySet));
     }
 
     private void assertGetIfPresent(Object key, Object value) throws Exception {

extensions/cache/deployment/src/test/java/io/quarkus/cache/test/runtime/SpecializedCacheTest.java

@@ -65,6 +65,11 @@ public <K, V> Uni<V> get(K key, Function<K, V> valueLoader) {
             throw new UnsupportedOperationException("This method is not tested here");
         }
 
+        @Override
+        public <K, V> Uni<V> getAsync(K key, Function<K, Uni<V>> valueLoader) {
+            throw new UnsupportedOperationException("This method is not tested here");
+        }
+
         @Override
         public Uni<Void> invalidate(Object key) {
             throw new UnsupportedOperationException("This method is not tested here");

extensions/cache/runtime/src/main/java/io/quarkus/cache/Cache.java

@@ -40,6 +40,19 @@ public interface Cache {
      */
     <K, V> Uni<V> get(K key, Function<K, V> valueLoader);
 
+    /**
+     * Returns a lazy asynchronous action that will emit the cache value identified by {@code key}, obtaining that value from
+     * {@code valueLoader} if necessary.
+     *
+     * @param <K>
+     * @param <V>
+     * @param key
+     * @param valueLoader
+     * @return a lazy asynchronous action that will emit a cache value
+     * @throws NullPointerException if the key is {@code null}
+     */
+    <K, V> Uni<V> getAsync(K key, Function<K, Uni<V>> valueLoader);
+
     /**
      * Removes the cache entry identified by {@code key} from the cache. If the key does not identify any cache entry, nothing
      * will happen.
@@ -61,4 +74,5 @@ public interface Cache {
      * @throws IllegalStateException if this cache is not an instance of {@code type}
      */
     <T extends Cache> T as(Class<T> type);
+
 }

extensions/cache/runtime/src/main/java/io/quarkus/cache/runtime/caffeine/CaffeineCacheImpl.java

@@ -93,6 +93,21 @@ public CompletionStage<V> get() {
                 });
     }
 
+    @SuppressWarnings("unchecked")
+    @Override
+    public <K, V> Uni<V> getAsync(K key, Function<K, Uni<V>> valueLoader) {
+        Objects.requireNonNull(key, NULL_KEYS_NOT_SUPPORTED_MSG);
+        return (Uni<V>) Uni.createFrom()
+                .completionStage(cache.asMap().computeIfAbsent(key, new Function<Object, CompletableFuture<Object>>() {
+                    @Override
+                    public CompletableFuture<Object> apply(Object key) {
+                        return (CompletableFuture<Object>) valueLoader.apply((K) key)
+                                .map(i -> NullValueConverter.toCacheValue(i))
+                                .subscribeAsCompletionStage();
+                    }
+                }));
+    }
+
     @Override
     public <V> CompletableFuture<V> getIfPresent(Object key) {
         Objects.requireNonNull(key, NULL_KEYS_NOT_SUPPORTED_MSG);

extensions/cache/runtime/src/main/java/io/quarkus/cache/runtime/noop/NoOpCache.java

@@ -29,6 +29,11 @@ public V get() {
         });
     }
 
+    @Override
+    public <K, V> Uni<V> getAsync(K key, Function<K, Uni<V>> valueLoader) {
+        return valueLoader.apply(key);
+    }
+
     @Override
     public Uni<Void> invalidate(Object key) {
         return Uni.createFrom().voidItem();