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

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

@@ -19,6 +19,7 @@
 import java.util.List;
 import java.util.Objects;
 
+import static io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE;
 import static org.swisspush.redisques.util.RedisquesAPI.*;
 
 public abstract class AbstractQueueAction implements QueueAction {
@@ -60,9 +61,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 message, Throwable cause) {
+        event.reply(exceptionFactory.newReplyException(RECIPIENT_FAILURE, 0, null, cause));
     }
 
     protected long getMaxAgeTimestamp() {

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

@@ -12,6 +12,7 @@
 
 import java.util.List;
 
+import static io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE;
 import static org.swisspush.redisques.util.RedisquesAPI.*;
 
 public class GetQueueItemsAction extends AbstractQueueAction {
@@ -37,9 +38,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(RECIPIENT_FAILURE, 0,
+                        "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/handler/AddQueueItemHandler.java

@@ -8,6 +8,7 @@
 import org.slf4j.Logger;
 import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 
+import static io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE;
 import static org.slf4j.LoggerFactory.getLogger;
 import static org.swisspush.redisques.util.RedisquesAPI.OK;
 import static org.swisspush.redisques.util.RedisquesAPI.STATUS;
@@ -33,8 +34,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(RECIPIENT_FAILURE, 0, null, reply.cause()));
         }
     }
 }

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

@@ -8,6 +8,7 @@
 import org.slf4j.Logger;
 import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 
+import static io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE;
 import static org.slf4j.LoggerFactory.getLogger;
 import static org.swisspush.redisques.util.RedisquesAPI.OK;
 import static org.swisspush.redisques.util.RedisquesAPI.STATUS;
@@ -33,8 +34,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(RECIPIENT_FAILURE, 0, null, reply.cause()));
         }
     }
 }

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

@@ -8,6 +8,7 @@
 import org.slf4j.Logger;
 import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 
+import static io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE;
 import static org.slf4j.LoggerFactory.getLogger;
 import static org.swisspush.redisques.util.RedisquesAPI.NO_SUCH_LOCK;
 import static org.swisspush.redisques.util.RedisquesAPI.OK;
@@ -39,8 +40,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(RECIPIENT_FAILURE, 0, 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(io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE, 0, 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/PutLockHandler.java

@@ -8,6 +8,7 @@
 import org.slf4j.Logger;
 import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 
+import static io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE;
 import static org.slf4j.LoggerFactory.getLogger;
 import static org.swisspush.redisques.util.RedisquesAPI.OK;
 import static org.swisspush.redisques.util.RedisquesAPI.STATUS;
@@ -33,8 +34,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(RECIPIENT_FAILURE, 0, reply.cause().getMessage(), reply.cause()));
         }
     }
 

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

@@ -8,6 +8,7 @@
 import org.slf4j.Logger;
 import org.swisspush.redisques.exception.RedisQuesExceptionFactory;
 
+import static io.vertx.core.eventbus.ReplyFailure.RECIPIENT_FAILURE;
 import static org.slf4j.LoggerFactory.getLogger;
 import static org.swisspush.redisques.util.RedisquesAPI.ERROR;
 import static org.swisspush.redisques.util.RedisquesAPI.OK;
@@ -36,8 +37,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(RECIPIENT_FAILURE, 0, reply.cause().getMessage(), 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));
     }
 }

src/test/java/org/swisspush/redisques/action/GetQueueItemActionTest.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 testGetQueueItemWhenRedisIsNotReady(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,6 @@ public void testGetQueueItemLINDEXFail(TestContext context){
         action.execute(message);
 
         verify(redisAPI, times(1)).lindex(anyString(), anyString(), any());
-        verify(message, times(1)).fail(eq(0), eq("booom"));
+        verify(message, times(1)).reply(isA(ReplyException.class));
     }
 }