From e68a8a6ca356e91786619a8fd5775cd77f729aee Mon Sep 17 00:00:00 2001
From: Tom Winterhalder <tom@winterhalder.ch>
Date: Mon, 18 Sep 2023 07:17:28 +0200
Subject: [PATCH] Release PR (#140)

* updating poms for branch'release-3.0.18' with non-snapshot versions

* updating poms for 3.0.19-SNAPSHOT development

* updating develop poms to master versions to avoid merge conflicts

* Updating develop poms back to pre merge state

* [SDCISA-13221] K8s: vertx-rest-storage (#139)

- made redis client aware of redis cluster

---------

Co-authored-by: runner <runner@fv-az444-271.mvkj0dyzoqtu3fmr4yatzo1xtd.cx.internal.cloudapp.net>
---
 pom.xml                                       |  2 +-
 .../reststorage/DefaultRedisProvider.java     | 21 ++++++++++++-------
 .../reststorage/util/ModuleConfiguration.java | 10 +++++++++
 3 files changed, 25 insertions(+), 8 deletions(-)

diff --git a/pom.xml b/pom.xml
index c21f22d..ba29020 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2,7 +2,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>org.swisspush</groupId>
     <artifactId>rest-storage</artifactId>
-    <version>3.0.18-SNAPSHOT</version>
+    <version>3.0.19-SNAPSHOT</version>
     <name>rest-storage</name>
     <description>
         Persistence for REST resources in the filesystem or a redis database
diff --git a/src/main/java/org/swisspush/reststorage/DefaultRedisProvider.java b/src/main/java/org/swisspush/reststorage/DefaultRedisProvider.java
index bf458cd..af8ef23 100644
--- a/src/main/java/org/swisspush/reststorage/DefaultRedisProvider.java
+++ b/src/main/java/org/swisspush/reststorage/DefaultRedisProvider.java
@@ -6,6 +6,7 @@
 import io.vertx.redis.client.Redis;
 import io.vertx.redis.client.RedisAPI;
 import io.vertx.redis.client.RedisConnection;
+import io.vertx.redis.client.RedisClientType;
 import io.vertx.redis.client.RedisOptions;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
@@ -42,7 +43,7 @@ public DefaultRedisProvider(Vertx vertx, ModuleConfiguration configuration) {
 
     @Override
     public Future<RedisAPI> redis() {
-        if(redisAPI != null) {
+        if (redisAPI != null) {
             return Future.succeededFuture(redisAPI);
         } else {
             return setupRedisClient();
@@ -53,15 +54,15 @@ private boolean reconnectEnabled() {
         return configuration.getRedisReconnectAttempts() != 0;
     }
 
-    private Future<RedisAPI> setupRedisClient(){
+    private Future<RedisAPI> setupRedisClient() {
         Promise<RedisAPI> currentPromise = Promise.promise();
         Promise<RedisAPI> masterPromise = connectPromiseRef.accumulateAndGet(
                 currentPromise, (oldVal, newVal) -> (oldVal != null) ? oldVal : newVal);
-        if( currentPromise == masterPromise ){
+        if (currentPromise == masterPromise) {
             // Our promise is THE promise. So WE have to resolve it.
             connectToRedis().onComplete(event -> {
                 connectPromiseRef.getAndSet(null);
-                if(event.failed()) {
+                if (event.failed()) {
                     currentPromise.fail(event.cause());
                 } else {
                     redisAPI = event.result();
@@ -89,13 +90,19 @@ private Future<RedisAPI> connectToRedis() {
         }
 
         if (connecting.compareAndSet(false, true)) {
-            redis = Redis.createClient(vertx, new RedisOptions()
-                    .setConnectionString(createConnectString())
+            RedisOptions redisOptions = new RedisOptions()
                     .setPassword((redisAuth == null ? "" : redisAuth))
                     .setMaxPoolSize(redisMaxPoolSize)
                     .setMaxPoolWaiting(redisMaxPoolWaitingSize)
                     .setPoolRecycleTimeout(redisPoolRecycleTimeoutMs)
-                    .setMaxWaitingHandlers(redisMaxPipelineWaitingSize));
+                    .setMaxWaitingHandlers(redisMaxPipelineWaitingSize);
+            if (configuration.isRedisClustered()) {
+                redisOptions.setType(RedisClientType.CLUSTER);
+                redisOptions.addConnectionString(createConnectString());
+            } else {
+                redisOptions.setConnectionString(createConnectString());
+            }
+            redis = Redis.createClient(vertx, redisOptions);
 
             redis.connect().onSuccess(conn -> {
                 log.info("Successfully connected to redis");
diff --git a/src/main/java/org/swisspush/reststorage/util/ModuleConfiguration.java b/src/main/java/org/swisspush/reststorage/util/ModuleConfiguration.java
index 7e0c634..f4b96d9 100644
--- a/src/main/java/org/swisspush/reststorage/util/ModuleConfiguration.java
+++ b/src/main/java/org/swisspush/reststorage/util/ModuleConfiguration.java
@@ -32,6 +32,7 @@ public enum StorageType {
     private String redisHost = "localhost";
     private int redisPort = 6379;
     private boolean redisEnableTls;
+    private boolean redisClustered;
     /**
      * @deprecated Instance authentication is considered as legacy. With Redis from 6.x on the ACL authentication method should be used.
      */
@@ -129,6 +130,11 @@ public ModuleConfiguration redisEnableTls(boolean redisEnableTls) {
         return this;
     }
 
+    public ModuleConfiguration redisClustered(boolean redisClustered) {
+        this.redisClustered = redisClustered;
+        return this;
+    }
+
     public ModuleConfiguration redisReconnectAttempts(int redisReconnectAttempts) {
         this.redisReconnectAttempts = redisReconnectAttempts;
         return this;
@@ -308,6 +314,10 @@ public boolean isRedisEnableTls() {
         return redisEnableTls;
     }
 
+    public boolean isRedisClustered() {
+        return redisClustered;
+    }
+
     public String getRedisAuth() {
         return redisAuth;
     }