fix: support prefix for resource queue (#258)(#256)

epam · Mar 5, 2024 · f3be802 · f3be802
1 parent e8c5dc2
commit f3be802
Showing 1 changed file with 6 additions and 5 deletions.
diff --git a/src/main/java/com/epam/aidial/core/service/ResourceService.java b/src/main/java/com/epam/aidial/core/service/ResourceService.java
@@ -34,7 +34,6 @@
 
 @Slf4j
 public class ResourceService implements AutoCloseable {
-    private static final String REDIS_QUEUE = "resource:queue";
     private static final Set<String> REDIS_FIELDS = Set.of("body", "created_at", "updated_at", "synced", "exists");
     private static final Set<String> REDIS_FIELDS_NO_BODY = Set.of("created_at", "updated_at", "synced", "exists");
 
@@ -52,6 +51,7 @@ public class ResourceService implements AutoCloseable {
     private final Duration cacheExpiration;
     private final int compressionMinSize;
     private final String prefix;
+    private final String resourceQueue;
 
     public ResourceService(Vertx vertx,
                            RedissonClient redis,
@@ -99,6 +99,7 @@ public ResourceService(Vertx vertx,
         this.cacheExpiration = Duration.ofMillis(cacheExpiration);
         this.compressionMinSize = compressionMinSize;
         this.prefix = prefix;
+        this.resourceQueue = "resource:" + BlobStorageUtil.toStoragePath(prefix, "queue");
 
         // vertex timer is called from event loop, so sync is done in worker thread to not block event loop
         this.syncTimer = vertx.setPeriodic(syncPeriod, syncPeriod, ignore -> vertx.executeBlocking(() -> sync()));
@@ -267,7 +268,7 @@ public boolean deleteResource(ResourceDescription descriptor) {
     private Void sync() {
         log.debug("Syncing");
         try {
-            RScoredSortedSet<String> set = redis.getScoredSortedSet(REDIS_QUEUE, StringCodec.INSTANCE);
+            RScoredSortedSet<String> set = redis.getScoredSortedSet(resourceQueue, StringCodec.INSTANCE);
             long now = time();
 
             for (String redisKey : set.valueRange(Double.NEGATIVE_INFINITY, true, now, true, 0, syncBatch)) {
@@ -290,7 +291,7 @@ private void sync(String redisKey) {
             Result result = redisGet(redisKey, false);
             if (result == null || result.synced) {
                 redis.getMap(redisKey, StringCodec.INSTANCE).expireIfNotSet(cacheExpiration);
-                redis.getScoredSortedSet(REDIS_QUEUE, StringCodec.INSTANCE).remove(redisKey);
+                redis.getScoredSortedSet(resourceQueue, StringCodec.INSTANCE).remove(redisKey);
                 return;
             }
 
@@ -396,7 +397,7 @@ private Result redisGet(String key, boolean withBody) {
     }
 
     private void redisPut(String key, Result result) {
-        RScoredSortedSet<String> set = redis.getScoredSortedSet(REDIS_QUEUE, StringCodec.INSTANCE);
+        RScoredSortedSet<String> set = redis.getScoredSortedSet(resourceQueue, StringCodec.INSTANCE);
         set.add(time() + syncDelay, key); // add resource to sync set before changing because calls below can fail
 
         RMap<String, String> map = redis.getMap(key, StringCodec.INSTANCE);
@@ -425,7 +426,7 @@ private void redisSync(String key) {
         map.put("synced", "true");
         map.expire(cacheExpiration);
 
-        RScoredSortedSet<String> set = redis.getScoredSortedSet(REDIS_QUEUE, StringCodec.INSTANCE);
+        RScoredSortedSet<String> set = redis.getScoredSortedSet(resourceQueue, StringCodec.INSTANCE);
         set.remove(key);
     }