Merge pull request #68 from swisspush/develop

PR for new release
swisspost · Nov 2, 2018 · 64bdde6 · 64bdde6
2 parents 9dae698 + a37cb12
commit 64bdde6
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 25 deletions.
diff --git a/pom.xml b/pom.xml
@@ -2,7 +2,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>org.swisspush</groupId>
     <artifactId>redisques</artifactId>
-    <version>2.5.4-SNAPSHOT</version>
+    <version>2.5.5-SNAPSHOT</version>
     <name>redisques</name>
     <description>
         A highly scalable redis-persistent queuing system for vertx

diff --git a/src/main/java/org/swisspush/redisques/RedisQues.java b/src/main/java/org/swisspush/redisques/RedisQues.java
@@ -4,6 +4,7 @@
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
+import io.vertx.core.eventbus.DeliveryOptions;
 import io.vertx.core.eventbus.EventBus;
 import io.vertx.core.eventbus.Message;
 import io.vertx.core.eventbus.MessageConsumer;
@@ -975,12 +976,12 @@ private void readQueue(final String queue) {
                         log.trace("RedisQues read queue lindex result: " + answer.result());
                     }
                     if (answer.result() != null) {
-                        processMessageWithTimeout(queue, answer.result(), sendResult -> {
+                        processMessageWithTimeout(queue, answer.result(), success -> {
 
                             // update the queue failure count and get a retry interval
-                            int retryInterval = updateQueueFailureCountAndGetRetryInterval(queue, sendResult.success);
+                            int retryInterval = updateQueueFailureCountAndGetRetryInterval(queue, success);
 
-                            if (sendResult.success) {
+                            if (success) {
                                 // Remove the processed message from the
                                 // queue
                                 String key1 = getQueuesPrefix() + queue;
@@ -994,7 +995,6 @@ private void readQueue(final String queue) {
                                     }
                                     log.debug("RedisQues Message removed, queue " + queue + " is ready again");
                                     myQueues.put(queue, QueueState.READY);
-                                    vertx.cancelTimer(sendResult.timeoutId);
                                     // Notify that we are stopped in
                                     // case it
                                     // was the last active consumer
@@ -1021,7 +1021,6 @@ private void readQueue(final String queue) {
 
                                 // reschedule
                                 log.debug("RedisQues will re-send the message to queue '" + queue + "' in " + retryInterval + " seconds");
-                                vertx.cancelTimer(sendResult.timeoutId);
                                 rescheduleSendMessageAfterFailure(queue, retryInterval);
                             }
                         });
@@ -1052,7 +1051,7 @@ private void rescheduleSendMessageAfterFailure(final String queue, int retryInte
         });
     }
 
-    private void processMessageWithTimeout(final String queue, final String payload, final Handler<SendResult> handler) {
+    private void processMessageWithTimeout(final String queue, final String payload, final Handler<Boolean> handler) {
         if (processorDelayMax > 0) {
             log.info("About to process message for queue " + queue + " with a maximum delay of " + processorDelayMax + "ms");
         }
@@ -1070,36 +1069,22 @@ private void processMessageWithTimeout(final String queue, final String payload,
                 log.trace("RedisQues process message: " + message + " for queue: " + queue + " send it to processor: " + processorAddress);
             }
 
-            // start a timer, which will cancel the processing, if the consumer didn't respond
-            final long timeoutId = vertx.setTimer(processorTimeout, timeoutId1 -> {
-                log.info("RedisQues QUEUE_ERROR: Consumer timeout " + uid + " queue: " + queue);
-                handler.handle(new SendResult(false, timeoutId1));
-            });
-
             // send the message to the consumer
-            eb.send(processorAddress, message, (Handler<AsyncResult<Message<JsonObject>>>) reply -> {
+            DeliveryOptions options = new DeliveryOptions().setSendTimeout(processorTimeout);
+            eb.send(processorAddress, message, options, (Handler<AsyncResult<Message<JsonObject>>>) reply -> {
                 Boolean success;
                 if (reply.succeeded()) {
                     success = OK.equals(reply.result().body().getString(STATUS));
                 } else {
+                    log.info("RedisQues QUEUE_ERROR: Consumer failed " + uid + " queue: " + queue + " (" + reply.cause().getMessage() + ")");
                     success = Boolean.FALSE;
                 }
-                handler.handle(new SendResult(success, timeoutId));
+                handler.handle(success);
             });
             updateTimestamp(queue, null);
         });
     }
 
-    private class SendResult {
-        public final Boolean success;
-        public final Long timeoutId;
-
-        public SendResult(Boolean success, Long timeoutId) {
-            this.success = success;
-            this.timeoutId = timeoutId;
-        }
-    }
-
     private void notifyConsumer(final String queue) {
         log.debug("RedisQues Notifying consumer of queue " + queue);
         final EventBus eb = vertx.eventBus();

diff --git a/src/test/java/org/swisspush/redisques/RedisQuesProcessorTest.java b/src/test/java/org/swisspush/redisques/RedisQuesProcessorTest.java
@@ -70,6 +70,7 @@ protected void deployRedisques(TestContext context) {
                 .processorAddress("processor-address")
                 .redisEncoding("ISO-8859-1")
                 .refreshPeriod(2)
+                .processorTimeout(10)
                 .build()
                 .asJsonObject();