From 180c3536c0059f9fe3e99de3e7a9c66052305f06 Mon Sep 17 00:00:00 2001
From: agrgr <agrgr@users.noreply.github.com>
Date: Tue, 18 Jun 2024 12:45:04 +0300
Subject: [PATCH] add support for 'sync=true' property in Cacheable annotation

---
 .../data/aerospike/cache/AerospikeCache.java  | 42 +++++++++++++++----
 ...AerospikeCacheManagerIntegrationTests.java | 35 ++++++++++++++++
 2 files changed, 69 insertions(+), 8 deletions(-)

diff --git a/src/main/java/org/springframework/data/aerospike/cache/AerospikeCache.java b/src/main/java/org/springframework/data/aerospike/cache/AerospikeCache.java
index a0631b210..cdd369b43 100644
--- a/src/main/java/org/springframework/data/aerospike/cache/AerospikeCache.java
+++ b/src/main/java/org/springframework/data/aerospike/cache/AerospikeCache.java
@@ -112,20 +112,46 @@ public Object getNativeCache() {
      * @return The value (bins) to which this cache maps the specified key.
      */
     @Override
-    @SuppressWarnings({"unchecked", "NullableProblems"})
+    @SuppressWarnings({"NullableProblems"})
     public <T> T get(Object key, Callable<T> valueLoader) {
-        T value = (T) client.get(null, getKey(key)).getValue(VALUE);
-        if (Objects.isNull(value)) {
-            try {
-                value = valueLoader.call();
+        Key dbKey = getKey(key);
+        Record record;
+        synchronized (this) {
+            record = client.get(null, dbKey);
+            if (record == null) {
+                T value = callValueLoader(valueLoader, key);
                 if (Objects.nonNull(value)) {
                     put(key, value);
                 }
-            } catch (Exception e) {
-                throw new Cache.ValueRetrievalException(key, valueLoader, e);
+                return value;
             }
         }
-        return value;
+        if (record.getValue(VALUE) != null) {
+            AerospikeReadData data = AerospikeReadData.forRead(dbKey, record);
+            Class<T> type = getValueType(valueLoader); // determine the class of T
+            return aerospikeConverter.read(type, data);
+        } else {
+            return null;
+        }
+    }
+
+    private <T> T callValueLoader(Callable<T> valueLoader, Object key) {
+        try {
+            return valueLoader.call();
+        } catch (Exception e) {
+            throw new Cache.ValueRetrievalException(key, valueLoader, e);
+        }
+    }
+
+    // Helper method to determine the class of T
+    @SuppressWarnings("unchecked")
+    private static <T> Class<T> getValueType(Callable<T> valueLoader) {
+        try {
+            // Use reflection to get the return type of the Callable
+            return (Class<T>) valueLoader.getClass().getMethod("call").getReturnType();
+        } catch (NoSuchMethodException e) {
+            throw new IllegalArgumentException("Cannot determine value type", e);
+        }
     }
 
     /**
diff --git a/src/test/java/org/springframework/data/aerospike/cache/AerospikeCacheManagerIntegrationTests.java b/src/test/java/org/springframework/data/aerospike/cache/AerospikeCacheManagerIntegrationTests.java
index e99c72f30..156af2ed7 100644
--- a/src/test/java/org/springframework/data/aerospike/cache/AerospikeCacheManagerIntegrationTests.java
+++ b/src/test/java/org/springframework/data/aerospike/cache/AerospikeCacheManagerIntegrationTests.java
@@ -66,6 +66,35 @@ public void shouldCache() {
         assertThat(cachingComponent.getNoOfCalls()).isEqualTo(1);
     }
 
+    @Test
+    public void testCacheableMethodSync() throws InterruptedException {
+        assertThat(cachingComponent.getNoOfCalls() == 0).isTrue();
+
+        // Creating two threads that will call cacheableMethod concurrently
+        Thread thread1 = new Thread(() -> {
+            CachedObject response = cachingComponent.cacheableMethodSynchronized(KEY);
+            assertThat(response).isNotNull();
+            assertThat(response.getValue()).isEqualTo(VALUE);
+        });
+
+        Thread thread2 = new Thread(() -> {
+            CachedObject response = cachingComponent.cacheableMethodSynchronized(KEY);
+            assertThat(response).isNotNull();
+            assertThat(response.getValue()).isEqualTo(VALUE);
+        });
+
+        // Starting both threads
+        thread1.start();
+        thread2.start();
+
+        // Waiting for both threads to complete
+        thread1.join();
+        thread2.join();
+
+        // Expecting method to be called only once due to synchronization
+        assertThat(cachingComponent.getNoOfCalls() == 1).isTrue();
+    }
+
     @Test
     public void shouldEvictCache() {
         CachedObject response1 = cachingComponent.cacheableMethod(KEY);
@@ -185,6 +214,12 @@ public CachedObject cacheableMethod(String param) {
             return new CachedObject("id", VALUE);
         }
 
+        @Cacheable(value = "TEST", sync = true)
+        public CachedObject cacheableMethodSynchronized(String param) {
+            noOfCalls++;
+            return new CachedObject("id", VALUE);
+        }
+
         @Cacheable(value = "CACHE-WITH-TTL")
         public CachedObject cacheableMethodWithTTL(String param) {
             noOfCalls++;