RedisQueueCircuitBreakerStorage implements QueueCircuitBreakerStorage { @Override public Future<String> popQueueToUnlock() { Future<String> future = Future.future(); redisClient.lpop(STORAGE_QUEUES_TO_UNLOCK, event -> { if(event.failed()){ future.fail(event.cause().getMessage()); return; } future.complete(event.result()); }); return future; } RedisQueueCircuitBreakerStorage(RedisClient redisClient); @Override Future<QueueCircuitState> getQueueCircuitState(PatternAndCircuitHash patternAndCircuitHash); @Override Future<QueueCircuitState> getQueueCircuitState(String circuitHash); @Override Future<JsonObject> getQueueCircuitInformation(String circuitHash); @Override Future<JsonObject> getAllCircuits(); @Override Future<UpdateStatisticsResult> updateStatistics(PatternAndCircuitHash patternAndCircuitHash, String uniqueRequestID, long timestamp,
                                                           int errorThresholdPercentage, long entriesMaxAgeMS, long minQueueSampleCount,
                                                           long maxQueueSampleCount, QueueResponseType queueResponseType); @Override Future<Void> lockQueue(String queueName, PatternAndCircuitHash patternAndCircuitHash); @Override Future<String> popQueueToUnlock(); @Override Future<Void> closeCircuit(PatternAndCircuitHash patternAndCircuitHash); @Override Future<Void> closeAndRemoveCircuit(PatternAndCircuitHash patternAndCircuitHash); @Override Future<Void> closeAllCircuits(); @Override Future<Void> reOpenCircuit(PatternAndCircuitHash patternAndCircuitHash); @Override Future<Long> setOpenCircuitsToHalfOpen(); @Override Future<List<String>> unlockSampleQueues(); static final String STORAGE_PREFIX; static final String STORAGE_INFOS_SUFFIX; static final String STORAGE_QUEUES_SUFFIX; static final String STORAGE_ALL_CIRCUITS; static final String STORAGE_HALFOPEN_CIRCUITS; static final String STORAGE_OPEN_CIRCUITS; static final String STORAGE_QUEUES_TO_UNLOCK; static final String FIELD_STATE; static final String FIELD_FAILRATIO; static final String FIELD_CIRCUIT; }
@Test public void testPopQueueToUnlock(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(STORAGE_QUEUES_TO_UNLOCK)); storage.popQueueToUnlock().setHandler(event -> { context.assertTrue(event.succeeded()); context.assertNull(event.result()); addToQueuesToUnlock("q3"); addToQueuesToUnlock("q1"); addToQueuesToUnlock("q2"); assertQueuesToUnlockItems(context, Arrays.asList("q1", "q2", "q3")); storage.popQueueToUnlock().setHandler(event1 -> { context.assertTrue(event1.succeeded()); context.assertEquals("q3", event1.result()); storage.popQueueToUnlock().setHandler(event2 -> { context.assertTrue(event2.succeeded()); context.assertEquals("q1", event2.result()); storage.popQueueToUnlock().setHandler(event3 -> { context.assertTrue(event3.succeeded()); context.assertEquals("q2", event3.result()); storage.popQueueToUnlock().setHandler(event4 -> { context.assertTrue(event4.succeeded()); context.assertNull(event4.result()); async.complete(); }); }); }); }); }); }