PR for release

pom.xml

@@ -2,7 +2,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>org.swisspush</groupId>
     <artifactId>redisques</artifactId>
-    <version>3.1.13-SNAPSHOT</version>
+    <version>4.1.1-SNAPSHOT</version>
     <name>redisques</name>
     <description>
         A highly scalable redis-persistent queuing system for vertx

src/main/java/org/swisspush/redisques/RedisQues.java

@@ -675,10 +675,10 @@ int updateQueueFailureCountAndGetRetryInterval(final String queueName, boolean s
 
     private void registerQueueCheck() {
         vertx.setPeriodic(configurationProvider.configuration().getCheckIntervalTimerMs(), periodicEvent -> {
-            redisProvider.redis().<Response>compose((RedisAPI redisAPI) -> {
+            redisProvider.redis().compose((RedisAPI redisAPI) -> {
                 int checkInterval = configurationProvider.configuration().getCheckInterval();
                 return redisAPI.send(Command.SET, queueCheckLastexecKey, String.valueOf(currentTimeMillis()), "NX", "EX", String.valueOf(checkInterval));
-            }).<Void>compose((Response todoExplainWhyThisIsIgnored) -> {
+            }).compose((Response todoExplainWhyThisIsIgnored) -> {
                 log.info("periodic queue check is triggered now");
                 return checkQueues();
             }).onFailure((Throwable ex) -> {
@@ -815,7 +815,7 @@ private Future<Void> consume(final String queueName) {
             log.trace("RedisQues consume get: {}", consumerKey);
             redisProvider.redis().onSuccess(redisAPI -> redisAPI.get(consumerKey, event1 -> {
                         if (event1.failed()) {
-                            log.error("Unable to get consumer for queue " + queueName, event1.cause());
+                            log.error("Unable to get consumer for queue {}", queueName, event1.cause());
                             return;
                         }
                         String consumer = Objects.toString(event1.result(), "");
@@ -857,7 +857,7 @@ private Future<Void> consume(final String queueName) {
                             });
                         }
                     }))
-                    .onFailure(throwable -> log.error("Redis: Unable to get consumer for queue " + queueName, throwable));
+                    .onFailure(throwable -> log.error("Redis: Unable to get consumer for queue {}", queueName, throwable));
         });
         return promise.future();
     }
@@ -1151,17 +1151,17 @@ private Future<Void> checkQueues() {
             AtomicInteger counter;
             Iterator<Response> iter;
         };
-        return Future.<Void>succeededFuture().<RedisAPI>compose((Void v) -> {
+        return Future.<Void>succeededFuture().compose((Void v) -> {
             log.debug("Checking queues timestamps");
             // List all queues that look inactive (i.e. that have not been updated since 3 periods).
             ctx.limit = currentTimeMillis() - 3L * configurationProvider.configuration().getRefreshPeriod() * 1000;
             return redisProvider.redis();
-        }).<Response>compose((RedisAPI redisAPI) -> {
+        }).compose((RedisAPI redisAPI) -> {
             ctx.redisAPI = redisAPI;
             var p = Promise.<Response>promise();
             redisAPI.zrangebyscore(Arrays.asList(queuesKey, "-inf", String.valueOf(ctx.limit)), p);
             return p.future();
-        }).<Void>compose((Response queues) -> {
+        }).compose((Response queues) -> {
             assert ctx.counter == null;
             assert ctx.iter == null;
             ctx.counter = new AtomicInteger(queues.size());

src/main/java/org/swisspush/redisques/RedisQuesRunner.java

@@ -10,7 +10,7 @@
  * Deploys vertx-redisques to vert.x.
  * Used in the standalone scenario.
  *
- * @author https://github.com/mcweba [Marc-Andre Weber]
+ * @author <a href="https://github.com/mcweba">Marc-André Weber</a>
  */
 public class RedisQuesRunner {
 
@@ -20,7 +20,7 @@ public static void main(String[] args) {
                 .httpRequestHandlerEnabled(true)
                 .redisReconnectAttempts(-1)
                 .redisPoolRecycleTimeoutMs(-1)
-                .redisReadyCheckIntervalMs(10000)
+                .redisReadyCheckIntervalMs(5000)
                 .build().asJsonObject();
 
         Vertx.vertx().deployVerticle(new RedisQues(), new DeploymentOptions().setConfig(configuration),

src/main/java/org/swisspush/redisques/action/AbstractQueueAction.java

@@ -9,6 +9,7 @@
 import io.vertx.core.json.JsonObject;
 import io.vertx.redis.client.Response;
 import org.slf4j.Logger;
+import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 import org.swisspush.redisques.util.QueueConfiguration;
 import org.swisspush.redisques.util.QueueStatisticsCollector;
 import org.swisspush.redisques.util.RedisProvider;
@@ -33,11 +34,12 @@ public abstract class AbstractQueueAction implements QueueAction {
     protected final String consumersPrefix;
     protected final String locksKey;
     protected final List<QueueConfiguration> queueConfigurations;
+    protected final RedisQuesExceptionFactory exceptionFactory;
     protected final QueueStatisticsCollector queueStatisticsCollector;
 
     public AbstractQueueAction(Vertx vertx, RedisProvider redisProvider, String address, String queuesKey,
                                String queuesPrefix, String consumersPrefix, String locksKey, List<QueueConfiguration> queueConfigurations,
-                               QueueStatisticsCollector queueStatisticsCollector, Logger log) {
+                               RedisQuesExceptionFactory exceptionFactory, QueueStatisticsCollector queueStatisticsCollector, Logger log) {
         this.vertx = vertx;
         this.redisProvider = redisProvider;
         this.address = address;
@@ -46,6 +48,7 @@ public AbstractQueueAction(Vertx vertx, RedisProvider redisProvider, String addr
         this.consumersPrefix = consumersPrefix;
         this.locksKey = locksKey;
         this.queueConfigurations = queueConfigurations;
+        this.exceptionFactory = exceptionFactory;
         this.queueStatisticsCollector = queueStatisticsCollector;
         this.log = log;
     }
@@ -57,6 +60,11 @@ protected Handler<Throwable> replyErrorMessageHandler(Message<JsonObject> event)
         };
     }
 
+    protected void handleFail(Message<JsonObject> event, String message, Throwable throwable) {
+        log.warn(message, exceptionFactory.newException(throwable));
+        event.fail(0, throwable.getMessage());
+    }
+
     protected long getMaxAgeTimestamp() {
         return System.currentTimeMillis() - MAX_AGE_MILLISECONDS;
     }

src/main/java/org/swisspush/redisques/action/AddQueueItemAction.java

@@ -4,6 +4,7 @@
 import io.vertx.core.eventbus.Message;
 import io.vertx.core.json.JsonObject;
 import org.slf4j.Logger;
+import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 import org.swisspush.redisques.handler.AddQueueItemHandler;
 import org.swisspush.redisques.util.QueueConfiguration;
 import org.swisspush.redisques.util.QueueStatisticsCollector;
@@ -19,19 +20,19 @@ public class AddQueueItemAction extends AbstractQueueAction {
     public AddQueueItemAction(
             Vertx vertx, RedisProvider redisProvider, String address, String queuesKey, String queuesPrefix,
             String consumersPrefix, String locksKey, List<QueueConfiguration> queueConfigurations,
-            QueueStatisticsCollector queueStatisticsCollector, Logger log
+            RedisQuesExceptionFactory exceptionFactory, QueueStatisticsCollector queueStatisticsCollector, Logger log
     ) {
         super(vertx, redisProvider, address, queuesKey, queuesPrefix, consumersPrefix, locksKey,
-                queueConfigurations, queueStatisticsCollector, log);
+                queueConfigurations, exceptionFactory, queueStatisticsCollector, log);
     }
 
     @Override
     public void execute(Message<JsonObject> event) {
         String key1 = queuesPrefix + event.body().getJsonObject(PAYLOAD).getString(QUEUENAME);
         String valueAddItem = event.body().getJsonObject(PAYLOAD).getString(BUFFER);
         var p = redisProvider.redis();
-        p.onSuccess(redisAPI -> redisAPI.rpush(Arrays.asList(key1, valueAddItem), new AddQueueItemHandler(event)));
-        p.onFailure(ex -> replyErrorMessageHandler(event).handle(ex));
+        p.onSuccess(redisAPI -> redisAPI.rpush(Arrays.asList(key1, valueAddItem), new AddQueueItemHandler(event, exceptionFactory)));
+        p.onFailure(ex -> handleFail(event,"Operation AddQueueItemAction failed", ex));
     }
 
 }

src/main/java/org/swisspush/redisques/action/BulkDeleteLocksAction.java

@@ -8,6 +8,7 @@
 import io.vertx.redis.client.impl.types.MultiType;
 import io.vertx.redis.client.impl.types.SimpleStringType;
 import org.slf4j.Logger;
+import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 import org.swisspush.redisques.util.QueueConfiguration;
 import org.swisspush.redisques.util.QueueStatisticsCollector;
 import org.swisspush.redisques.util.RedisProvider;
@@ -20,9 +21,9 @@ public class BulkDeleteLocksAction extends AbstractQueueAction {
 
     public BulkDeleteLocksAction(Vertx vertx, RedisProvider redisProvider, String address, String queuesKey, String queuesPrefix,
                                  String consumersPrefix, String locksKey, List<QueueConfiguration> queueConfigurations,
-                                 QueueStatisticsCollector queueStatisticsCollector, Logger log) {
+                                 RedisQuesExceptionFactory exceptionFactory, QueueStatisticsCollector queueStatisticsCollector, Logger log) {
         super(vertx, redisProvider, address, queuesKey, queuesPrefix, consumersPrefix, locksKey, queueConfigurations,
-                queueStatisticsCollector, log);
+                exceptionFactory, queueStatisticsCollector, log);
     }
 
     @Override

src/main/java/org/swisspush/redisques/action/BulkDeleteQueuesAction.java

@@ -5,6 +5,7 @@
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import org.slf4j.Logger;
+import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 import org.swisspush.redisques.util.QueueConfiguration;
 import org.swisspush.redisques.util.QueueStatisticsCollector;
 import org.swisspush.redisques.util.RedisProvider;
@@ -18,10 +19,10 @@ public class BulkDeleteQueuesAction extends AbstractQueueAction {
     public BulkDeleteQueuesAction(
             Vertx vertx, RedisProvider redisProvider, String address, String queuesKey, String queuesPrefix,
             String consumersPrefix, String locksKey, List<QueueConfiguration> queueConfigurations,
-            QueueStatisticsCollector queueStatisticsCollector, Logger log
+            RedisQuesExceptionFactory exceptionFactory, QueueStatisticsCollector queueStatisticsCollector, Logger log
     ) {
         super(vertx, redisProvider, address, queuesKey, queuesPrefix, consumersPrefix, locksKey,
-                queueConfigurations, queueStatisticsCollector, log);
+                queueConfigurations, exceptionFactory, queueStatisticsCollector, log);
     }
 
     @Override
@@ -49,11 +50,10 @@ public void execute(Message<JsonObject> event) {
             if (delManyReply.succeeded()) {
                 event.reply(createOkReply().put(VALUE, delManyReply.result().toLong()));
             } else {
-                log.error("Failed to bulkDeleteQueues", new Exception(delManyReply.cause()));
-                event.reply(createErrorReply());
+                handleFail(event, "Failed to bulkDeleteQueues", delManyReply.cause());
             }
         }));
-        p.onFailure(ex -> replyErrorMessageHandler(event).handle(ex));
+        p.onFailure(ex -> handleFail(event, "Operation BulkDeleteQueuesAction failed", ex));
     }
 
 }

src/main/java/org/swisspush/redisques/action/BulkPutLocksAction.java

@@ -5,6 +5,7 @@
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import org.slf4j.Logger;
+import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 import org.swisspush.redisques.handler.PutLockHandler;
 import org.swisspush.redisques.util.QueueConfiguration;
 import org.swisspush.redisques.util.QueueStatisticsCollector;
@@ -19,10 +20,10 @@ public class BulkPutLocksAction extends AbstractQueueAction {
     public BulkPutLocksAction(
             Vertx vertx, RedisProvider redisProvider, String address, String queuesKey, String queuesPrefix,
             String consumersPrefix, String locksKey, List<QueueConfiguration> queueConfigurations,
-            QueueStatisticsCollector queueStatisticsCollector, Logger log
+            RedisQuesExceptionFactory exceptionFactory, QueueStatisticsCollector queueStatisticsCollector, Logger log
     ) {
         super(vertx, redisProvider, address, queuesKey, queuesPrefix, consumersPrefix, locksKey,
-                queueConfigurations, queueStatisticsCollector, log);
+                queueConfigurations, exceptionFactory, queueStatisticsCollector, log);
     }
 
     @Override
@@ -45,9 +46,7 @@ public void execute(Message<JsonObject> event) {
         }
 
         var p = redisProvider.redis();
-        p.onSuccess(redisAPI -> {
-            redisAPI.hmset(buildLocksItems(locksKey, locks, lockInfo), new PutLockHandler(event));
-        });
+        p.onSuccess(redisAPI -> redisAPI.hmset(buildLocksItems(locksKey, locks, lockInfo), new PutLockHandler(event, exceptionFactory)));
         p.onFailure(ex -> replyErrorMessageHandler(event).handle(ex));
     }
 

src/main/java/org/swisspush/redisques/action/DeleteAllLocksAction.java

@@ -5,6 +5,7 @@
 import io.vertx.core.json.JsonObject;
 import io.vertx.redis.client.Response;
 import org.slf4j.Logger;
+import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 import org.swisspush.redisques.util.QueueConfiguration;
 import org.swisspush.redisques.util.QueueStatisticsCollector;
 import org.swisspush.redisques.util.RedisProvider;
@@ -18,10 +19,10 @@ public class DeleteAllLocksAction extends AbstractQueueAction {
     public DeleteAllLocksAction(
             Vertx vertx, RedisProvider redisProvider, String address, String queuesKey, String queuesPrefix,
             String consumersPrefix, String locksKey, List<QueueConfiguration> queueConfigurations,
-            QueueStatisticsCollector queueStatisticsCollector, Logger log
+            RedisQuesExceptionFactory exceptionFactory, QueueStatisticsCollector queueStatisticsCollector, Logger log
     ) {
         super(vertx, redisProvider, address, queuesKey, queuesPrefix, consumersPrefix, locksKey,
-                queueConfigurations, queueStatisticsCollector, log);
+                queueConfigurations, exceptionFactory, queueStatisticsCollector, log);
     }
 
     @Override

src/main/java/org/swisspush/redisques/action/DeleteAllQueueItemsAction.java

@@ -6,6 +6,7 @@
 import io.vertx.core.json.JsonObject;
 import io.vertx.redis.client.Response;
 import org.slf4j.Logger;
+import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 import org.swisspush.redisques.util.QueueConfiguration;
 import org.swisspush.redisques.util.QueueStatisticsCollector;
 import org.swisspush.redisques.util.RedisProvider;
@@ -21,10 +22,10 @@ public class DeleteAllQueueItemsAction extends AbstractQueueAction {
     public DeleteAllQueueItemsAction(
             Vertx vertx, RedisProvider redisProvider, String address, String queuesKey, String queuesPrefix,
             String consumersPrefix, String locksKey, List<QueueConfiguration> queueConfigurations,
-            QueueStatisticsCollector queueStatisticsCollector, Logger log
+            RedisQuesExceptionFactory exceptionFactory, QueueStatisticsCollector queueStatisticsCollector, Logger log
     ) {
         super(vertx, redisProvider, address, queuesKey, queuesPrefix, consumersPrefix, locksKey,
-                queueConfigurations, queueStatisticsCollector, log);
+                queueConfigurations, exceptionFactory, queueStatisticsCollector, log);
     }
 
     @Override
@@ -36,40 +37,33 @@ public void execute(Message<JsonObject> event) {
         p.onSuccess(redisAPI -> {
             redisAPI.del(Collections.singletonList(buildQueueKey(queue)), deleteReply -> {
                 if (deleteReply.failed()) {
-                    log.warn("Failed to deleteAllQueueItems. But we'll continue anyway",
-                            new Exception(deleteReply.cause()));
-                    // May we should 'fail()' here. But:
-                    // 1st: We don't, to keep backward compatibility
-                    // 2nd: We don't, to may unlock below.
+                    handleFail(event, "Operation DeleteAllQueueItems failed", deleteReply.cause());
+                    return;
                 }
                 queueStatisticsCollector.resetQueueFailureStatistics(queue, (Throwable ex, Void v) -> {
                     if (ex != null) log.warn("TODO_2958iouhj error handling", ex);
                 });
                 if (unlock) {
                     redisAPI.hdel(Arrays.asList(locksKey, queue), unlockReply -> {
                         if (unlockReply.failed()) {
-                            log.warn("Failed to unlock queue '{}'. Will continue anyway",
-                                    queue, unlockReply.cause());
-                            // IMO we should 'fail()' here. But we don't, to keep backward compatibility.
+                            handleFail(event, "Failed to unlock queue " + queue, unlockReply.cause());
+                        } else {
+                            handleDeleteQueueReply(event, deleteReply);
                         }
-                        handleDeleteQueueReply(event, deleteReply);
                     });
                 } else {
                     handleDeleteQueueReply(event, deleteReply);
                 }
             });
         });
-        p.onFailure(ex -> {
-            log.error("Redis: Failed to delete all queue items", new Exception(ex));
-            event.reply(createErrorReply());
-        });
+        p.onFailure(ex -> handleFail(event, "Operation DeleteAllQueueItems failed", ex));
     }
 
     private void handleDeleteQueueReply(Message<JsonObject> event, AsyncResult<Response> reply) {
         if (reply.succeeded()) {
             event.reply(createOkReply().put(VALUE, reply.result().toLong()));
         } else {
-            log.error("Failed to replyResultGreaterThanZero", new Exception(reply.cause()));
+            log.error("Failed to replyResultGreaterThanZero", exceptionFactory.newException(reply.cause()));
             event.reply(createErrorReply());
         }
     }

src/main/java/org/swisspush/redisques/action/DeleteLockAction.java

@@ -4,6 +4,7 @@
 import io.vertx.core.eventbus.Message;
 import io.vertx.core.json.JsonObject;
 import org.slf4j.Logger;
+import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 import org.swisspush.redisques.handler.DeleteLockHandler;
 import org.swisspush.redisques.util.QueueConfiguration;
 import org.swisspush.redisques.util.QueueStatisticsCollector;
@@ -21,10 +22,10 @@ public class DeleteLockAction extends AbstractQueueAction {
     public DeleteLockAction(
             Vertx vertx, RedisProvider redisProvider, String address, String queuesKey, String queuesPrefix,
             String consumersPrefix, String locksKey, List<QueueConfiguration> queueConfigurations,
-            QueueStatisticsCollector queueStatisticsCollector, Logger log
+            RedisQuesExceptionFactory exceptionFactory, QueueStatisticsCollector queueStatisticsCollector, Logger log
     ) {
         super(vertx, redisProvider, address, queuesKey, queuesPrefix, consumersPrefix, locksKey,
-                queueConfigurations, queueStatisticsCollector, log);
+                queueConfigurations, exceptionFactory, queueStatisticsCollector, log);
     }
 
     @Override
@@ -33,14 +34,18 @@ public void execute(Message<JsonObject> event) {
         var p = redisProvider.redis();
         p.onSuccess(redisAPI -> {
             redisAPI.exists(Collections.singletonList(queuesPrefix + queueName), event1 -> {
-                if( event1.failed() ) log.warn("Concealed error", new Exception(event1.cause()));
+                if (event1.failed()) {
+                    log.warn("Concealed error", exceptionFactory.newException(event1.cause()));
+                }
+
                 if (event1.succeeded() && event1.result() != null && event1.result().toInteger() == 1) {
                     notifyConsumer(queueName);
                 }
-                redisAPI.hdel(Arrays.asList(locksKey, queueName), new DeleteLockHandler(event));
+
+                redisAPI.hdel(Arrays.asList(locksKey, queueName), new DeleteLockHandler(event, exceptionFactory));
             });
         });
-        p.onFailure(ex -> replyErrorMessageHandler(event).handle(ex));
+        p.onFailure(ex -> handleFail(event, "Operation DeleteLockAction failed", ex));
     }
 
 }