Immediately notify processor when a queue is unlocked. This fixes #23

Author: mcweba

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

@@ -231,7 +231,13 @@ public void handle(final Message<JsonObject> event) {
                         redisClient.hget(redisques_locks, event.body().getJsonObject(PAYLOAD).getString(QUEUENAME),new GetLockHandler(event));
                         break;
                     case deleteLock:
-                        redisClient.hdel(redisques_locks, event.body().getJsonObject(PAYLOAD).getString(QUEUENAME), new DeleteLockHandler(event));
+                        String queueName = event.body().getJsonObject(PAYLOAD).getString(QUEUENAME);
+                        redisClient.exists(queuesPrefix + queueName, event1 -> {
+                            if(event1.succeeded() && event1.result() == 1){
+                                notifyConsumer(queueName);
+                            }
+                            redisClient.hdel(redisques_locks, queueName, new DeleteLockHandler(event));
+                        });
                         break;
                     case getQueueItemsCount:
                         redisClient.llen(queuesPrefix + event.body().getJsonObject(PAYLOAD).getString(QUEUENAME), new GetQueueItemsCountHandler(event));

src/test/java/org/swisspush/redisques/RedisQuesProcessorTest.java

@@ -347,17 +347,45 @@ public void queueProcessorShouldNotBeNotNotifiedWithLockedQueue(TestContext cont
 
         lockQueue(queue);
 
-
         queueProcessor.handler(event -> processorCalled.set(true));
 
         eventBusSend(buildEnqueueOperation(queue, "hello"), reply -> {
             context.assertEquals(OK, reply.result().body().getString(STATUS));
+            sleep(5000);
+            context.assertFalse(processorCalled.get(), "QueueProcessor should not have been called after enqueue into a locked queue");
+            async.complete();
         });
+    }
 
-        // after at most 5 seconds, the processor-address consumer should not have been called
-        Awaitility.await().atMost(Duration.FIVE_SECONDS).until(processorCalled::get, equalTo(false));
+    @Test
+    public void queueProcessorShouldHaveBeenNotifiedImmediatelyAfterQueueUnlock(TestContext context) throws Exception {
+        Async async = context.async();
+        flushAll();
 
-        async.complete();
+        String queue = "queue1";
+        final AtomicBoolean processorCalled = new AtomicBoolean(false);
+
+        lockQueue(queue);
+
+        queueProcessor.handler(event -> {
+            processorCalled.set(true);
+        });
+
+        eventBusSend(buildEnqueueOperation(queue, "hello"), reply -> {
+            context.assertEquals(OK, reply.result().body().getString(STATUS));
+
+            sleep(5000);
+
+            // after at most 5 seconds, the processor-address consumer should not have been called
+            context.assertFalse(processorCalled.get(), "QueueProcessor should not have been called after enqueue into a locked queue");
+
+            eventBusSend(buildDeleteLockOperation(queue), event -> {
+                context.assertEquals(OK, event.result().body().getString(STATUS));
+                sleep(100);
+                context.assertTrue(processorCalled.get(), "QueueProcessor should have been called immediately after queue unlock");
+                async.complete();
+            });
+        });
     }
 
     private void sleep(int millis) {