Merge pull request #208 from swisspost/develop

PR for Release

Author: mcweba

pom.xml

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

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

@@ -19,7 +19,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
 
-import static io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE;
 import static java.lang.Long.compare;
 import static java.lang.System.currentTimeMillis;
 import static java.util.Collections.emptyList;
@@ -72,7 +71,7 @@ public QueueStatsService(
 
     public <CTX> void getQueueStats(CTX mCtx, GetQueueStatsMentor<CTX> mentor) {
         if (!incomingRequestQuota.tryAcquire()) {
-            Throwable ex = exceptionFactory.newReplyException(RECIPIENT_FAILURE, 429,
+            Throwable ex = exceptionFactory.newReplyException(429,
                     "Server too busy to handle yet-another-queue-stats-request now", null);
             vertx.runOnContext(v -> mentor.onError(ex, mCtx));
             return;

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

@@ -60,9 +60,8 @@ 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 void handleFail(Message<JsonObject> event, String msg, Throwable cause) {
+        event.reply(exceptionFactory.newReplyException(msg, cause));
     }
 
     protected long getMaxAgeTimestamp() {

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

@@ -37,9 +37,8 @@ public void execute(Message<JsonObject> event) {
                 redisAPI.lrange(keyListRange, "0", String.valueOf(maxQueueItemCountIndex),
                         new GetQueueItemsHandler(event, queueItemCount));
             } else {
-                String msg = "Operation getQueueItems failed to extract queueItemCount";
-                log.warn(msg);
-                event.fail(0, msg);
+                event.reply(exceptionFactory.newReplyException(
+                    "Operation getQueueItems failed to extract queueItemCount", null));
             }
         }).onFailure(throwable -> handleFail(event, "Operation getQueueItems failed", throwable)))
                 .onFailure(throwable -> handleFail(event, "Operation getQueueItems failed", throwable));

src/main/java/org/swisspush/redisques/exception/RedisQuesExceptionFactory.java

@@ -1,7 +1,6 @@
 package org.swisspush.redisques.exception;
 
 import io.vertx.core.eventbus.ReplyException;
-import io.vertx.core.eventbus.ReplyFailure;
 
 
 /**
@@ -30,7 +29,11 @@ public interface RedisQuesExceptionFactory {
 
     public RuntimeException newRuntimeException(String message, Throwable cause);
 
-    public ReplyException newReplyException(ReplyFailure failureType, int failureCode, String msg, Throwable cause);
+    default ReplyException newReplyException(String msg, Throwable cause) {
+        return newReplyException(0, msg, cause);
+    }
+
+    public ReplyException newReplyException(int failureCode, String msg, Throwable cause);
 
 
     /**

src/main/java/org/swisspush/redisques/exception/ThriftyRedisQuesExceptionFactory.java

@@ -33,13 +33,13 @@ public RuntimeException newRuntimeException(String message, Throwable cause) {
     }
 
     @Override
-    public ReplyException newReplyException(ReplyFailure failureType, int failureCode, String msg, Throwable cause) {
+    public ReplyException newReplyException(int failureCode, String msg, Throwable cause) {
         // There was once a fix in vertx for this (https://github.com/eclipse-vertx/vert.x/issues/4840)
         // but for whatever reason in our case we still see stack-trace recordings. Passing
         // this subclass to {@link io.vertx.core.eventbus.Message#reply(Object)} seems to
         // do the trick.
         if (msg == null && cause != null) msg = cause.getMessage();
-        return new ReplyException(failureType, failureCode, msg) {
+        return new ReplyException(ReplyFailure.RECIPIENT_FAILURE, failureCode, msg) {
             @Override public Throwable fillInStackTrace() { return this; }
         };
     }

src/main/java/org/swisspush/redisques/exception/WastefulRedisQuesExceptionFactory.java

@@ -26,9 +26,9 @@ public RuntimeException newRuntimeException(String message, Throwable cause) {
     }
 
     @Override
-    public ReplyException newReplyException(ReplyFailure failureType, int failureCode, String msg, Throwable cause) {
+    public ReplyException newReplyException(int failureCode, String msg, Throwable cause) {
         if (msg == null && cause != null) msg = cause.getMessage();
-        ReplyException ex = new ReplyException(failureType, failureCode, msg);
+        ReplyException ex = new ReplyException(ReplyFailure.RECIPIENT_FAILURE, failureCode, msg);
         if (cause != null) ex.initCause(cause);
         return ex;
     }

src/main/java/org/swisspush/redisques/handler/AddQueueItemHandler.java

@@ -33,8 +33,7 @@ public void handle(AsyncResult<Response> reply) {
         if(reply.succeeded()){
             event.reply(new JsonObject().put(STATUS, OK));
         } else {
-            log.warn("Concealed error", exceptionFactory.newException(reply.cause()));
-            event.fail(0, reply.cause().getMessage());
+            event.reply(exceptionFactory.newReplyException(null, reply.cause()));
         }
     }
 }

src/main/java/org/swisspush/redisques/handler/DeleteLockHandler.java

@@ -33,8 +33,7 @@ public void handle(AsyncResult<Response> reply) {
         if (reply.succeeded()) {
             event.reply(new JsonObject().put(STATUS, OK));
         } else {
-            log.warn("Concealed error", exceptionFactory.newException(reply.cause()));
-            event.fail(0, reply.cause().getMessage());
+            event.reply(exceptionFactory.newReplyException(null, reply.cause()));
         }
     }
 }

src/main/java/org/swisspush/redisques/handler/GetLockHandler.java

@@ -39,8 +39,7 @@ public void handle(AsyncResult<Response> reply) {
                 event.reply(new JsonObject().put(STATUS, NO_SUCH_LOCK));
             }
         } else {
-            log.warn("Concealed error", exceptionFactory.newException(reply.cause()));
-            event.fail(0, reply.cause().getMessage());
+            event.reply(exceptionFactory.newReplyException(null, reply.cause()));
         }
     }