Pass error information through to caller in place of out-of-context logging

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()));
         }
     }
 

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

@@ -33,8 +33,7 @@ public GetQueueItemHandler(Message<JsonObject> event, RedisQuesExceptionFactory
     @Override
     public void handle(AsyncResult<Response> reply) {
         if(reply.failed()) {
-            log.warn("Concealed error", exceptionFactory.newException(reply.cause()));
-            event.fail(0, reply.cause().getMessage());
+            event.reply(exceptionFactory.newReplyException(null, reply.cause()));
         } else if (reply.result() != null) {
             event.reply(new JsonObject().put(STATUS, OK).put(VALUE, reply.result().toString()));
         } else {

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

@@ -5,7 +5,6 @@
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.eventbus.Message;
-import io.vertx.core.eventbus.ReplyFailure;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.redis.client.Command;
@@ -87,7 +86,7 @@ public void handle(AsyncResult<Response> handleQueues) {
             return;
         }
         if (redisRequestQuota.availablePermits() <= 0) {
-            event.reply(exceptionFactory.newReplyException(ReplyFailure.RECIPIENT_FAILURE, 429,
+            event.reply(exceptionFactory.newReplyException(429,
                     "Too many simultaneous '" + GetQueuesItemsCountHandler.class.getSimpleName() + "' requests in progress", null));
             return;
         }

src/main/java/org/swisspush/redisques/handler/PutLockHandler.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(reply.cause().getMessage(), reply.cause()));
         }
     }
 

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

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

src/test/java/org/swisspush/redisques/action/AbstractQueueActionTest.java

@@ -19,6 +19,7 @@
 import java.util.Optional;
 
 import static org.mockito.Mockito.*;
+import static org.swisspush.redisques.exception.RedisQuesExceptionFactory.newWastefulExceptionFactory;
 
 public abstract class AbstractQueueActionTest {
 
@@ -40,7 +41,7 @@ public abstract class AbstractQueueActionTest {
     public void setup() {
         redisAPI = Mockito.mock(RedisAPI.class);
         redisProvider = Mockito.mock(RedisProvider.class);
-        exceptionFactory = Mockito.mock(RedisQuesExceptionFactory.class);
+        exceptionFactory = newWastefulExceptionFactory();
         when(redisProvider.redis()).thenReturn(Future.succeededFuture(redisAPI));
 
         memoryUsageProvider = Mockito.mock(MemoryUsageProvider.class);

src/test/java/org/swisspush/redisques/action/AddQueueItemActionTest.java

@@ -1,6 +1,7 @@
 package org.swisspush.redisques.action;
 
 import io.vertx.core.Future;
+import io.vertx.core.eventbus.ReplyException;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
 import org.junit.Before;
@@ -12,6 +13,7 @@
 
 import java.util.ArrayList;
 
+import static io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE;
 import static org.mockito.Mockito.*;
 import static org.swisspush.redisques.util.RedisquesAPI.buildAddQueueItemOperation;
 
@@ -39,7 +41,7 @@ public void testAddQueueItemWhenRedisIsNotReady(TestContext context){
 
         action.execute(message);
 
-        verify(message, times(1)).fail(eq(0), eq("not ready"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
         verifyNoInteractions(redisAPI);
     }
 
@@ -73,6 +75,6 @@ public void testAddQueueItemRPUSHFail(TestContext context){
         action.execute(message);
 
         verify(redisAPI, times(1)).rpush(anyList(), any());
-        verify(message, times(1)).fail(eq(0), eq("booom"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
     }
 }

src/test/java/org/swisspush/redisques/action/BulkDeleteQueuesActionTest.java

@@ -2,6 +2,7 @@
 
 import io.vertx.core.Future;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.eventbus.ReplyException;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.unit.TestContext;
@@ -45,7 +46,7 @@ public void testBulkDeleteQueuesWhenRedisIsNotReady(TestContext context){
 
         action.execute(message);
 
-        verify(message, times(1)).fail(eq(0), eq("not ready"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
         verifyNoInteractions(redisAPI);
     }
 
@@ -78,7 +79,7 @@ public void testBulkDeleteQueuesDELFail(TestContext context){
         action.execute(message);
 
         verify(redisAPI, times(1)).del(anyList(), any());
-        verify(message, times(1)).fail(eq(0), eq("booom"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
     }
 
     @Test

src/test/java/org/swisspush/redisques/action/DeleteAllQueueItemsActionTest.java

@@ -2,6 +2,7 @@
 
 import io.vertx.core.Future;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.eventbus.ReplyException;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
@@ -82,7 +83,7 @@ public void testDeleteAllQueueItemsWhenRedisIsNotReady(TestContext context){
 
         action.execute(message);
 
-        verify(message, times(1)).fail(eq(0), eq("not ready"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
         verifyNoInteractions(redisAPI);
     }
 
@@ -99,7 +100,7 @@ public void testRedisApiDELFail(TestContext context){
         action.execute(message);
 
         verify(redisAPI, times(1)).del(anyList(), any());
-        verify(message, times(1)).fail(eq(0), eq("boooom"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
     }
 
     @Test
@@ -122,6 +123,7 @@ public void testRedisApiUnlockFail(TestContext context){
 
         verify(redisAPI, times(1)).del(anyList(), any());
         verify(redisAPI, times(1)).hdel(anyList(), any());
-        verify(message, times(1)).fail(eq(0), eq("boooom"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
+        //verify(message, times(1)).fail(eq(0), eq("boooom"));
     }
 }

src/test/java/org/swisspush/redisques/action/DeleteLockActionTest.java

@@ -1,6 +1,7 @@
 package org.swisspush.redisques.action;
 
 import io.vertx.core.Future;
+import io.vertx.core.eventbus.ReplyException;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
 import io.vertx.redis.client.impl.types.SimpleStringType;
@@ -40,7 +41,7 @@ public void testDeleteLockWhenRedisIsNotReady(TestContext context){
 
         action.execute(message);
 
-        verify(message, times(1)).fail(eq(0), eq("not ready"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
         verifyNoInteractions(redisAPI);
     }
 
@@ -110,6 +111,6 @@ public void testDeleteLockHDELFail(TestContext context){
 
         verify(redisAPI, times(1)).exists(anyList(), any());
         verify(redisAPI, times(1)).hdel(anyList(), any());
-        verify(message, times(1)).fail(eq(0), eq("booom"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
     }
 }

src/test/java/org/swisspush/redisques/action/DeleteQueueItemActionTest.java

@@ -1,6 +1,7 @@
 package org.swisspush.redisques.action;
 
 import io.vertx.core.Future;
+import io.vertx.core.eventbus.ReplyException;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
 import org.junit.Before;
@@ -39,7 +40,7 @@ public void testDeleteQueueItemWhenRedisIsNotReady(TestContext context){
 
         action.execute(message);
 
-        verify(message, times(1)).fail(eq(0), eq("not ready"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
         verifyNoInteractions(redisAPI);
     }
 
@@ -57,7 +58,7 @@ public void testFailedLSET(TestContext context){
 
         verify(redisAPI, times(1)).lset(anyString(), eq("0"), eq("TO_DELETE"), any());
         verify(redisAPI, never()).lrem(anyString(), anyString(), anyString());
-        verify(message, times(1)).fail(eq(0), eq("boooom"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
     }
 
     @Test
@@ -97,7 +98,7 @@ public void testFailedLREM(TestContext context){
 
         verify(redisAPI, times(1)).lset(anyString(), eq("0"), eq("TO_DELETE"), any());
         verify(redisAPI, times(1)).lrem(anyString(), eq("0"), eq("TO_DELETE"), any());
-        verify(message, times(1)).fail(eq(0), eq("boooom"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
     }
 
     @Test

src/test/java/org/swisspush/redisques/action/GetAllLocksActionTest.java

@@ -2,6 +2,7 @@
 
 import io.vertx.core.Future;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.eventbus.ReplyException;
 import io.vertx.core.impl.future.FailedFuture;
 import io.vertx.core.impl.future.SucceededFuture;
 import io.vertx.core.json.JsonObject;
@@ -45,7 +46,7 @@ public void testGetAllLocksWhenRedisIsNotReady(TestContext context){
 
         action.execute(message);
 
-        verify(message, times(1)).fail(eq(0), eq("not ready"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
         verifyNoInteractions(redisAPI);
     }
 

src/test/java/org/swisspush/redisques/action/GetLockActionTest.java

@@ -2,6 +2,7 @@
 
 import io.vertx.core.Future;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.eventbus.ReplyException;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
@@ -42,7 +43,7 @@ public void testGetLockWhenRedisIsNotReady(TestContext context){
 
         action.execute(message);
 
-        verify(message, times(1)).fail(eq(0), eq("not ready"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
         verifyNoInteractions(redisAPI);
     }
 
@@ -92,6 +93,7 @@ public void testGetLockHGETFail(TestContext context){
         action.execute(message);
 
         verify(redisAPI, times(1)).hget(anyString(), anyString(), any());
-        verify(message, times(1)).fail(eq(0), eq("booom"));
+        //verify(message, times(1)).fail(eq(0), eq("booom"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
     }
 }