RedisQueueCircuitBreakerStorage implements QueueCircuitBreakerStorage { @Override public Future<Void> closeAllCircuits() { Future<Void> future = Future.future(); Future<Void> closeOpenCircuitsFuture = closeCircuitsByKey(STORAGE_OPEN_CIRCUITS); Future<Void> closeHalfOpenCircuitsFuture = closeCircuitsByKey(STORAGE_HALFOPEN_CIRCUITS); CompositeFuture.all(closeOpenCircuitsFuture, closeHalfOpenCircuitsFuture).setHandler(event -> { if(event.succeeded()){ future.complete(); } else { future.fail(event.cause().getMessage()); } }); 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 testCloseAllCircuits(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(STORAGE_HALFOPEN_CIRCUITS)); context.assertFalse(jedis.exists(STORAGE_ALL_CIRCUITS)); context.assertFalse(jedis.exists(STORAGE_OPEN_CIRCUITS)); context.assertFalse(jedis.exists(STORAGE_QUEUES_TO_UNLOCK)); buildCircuitEntry("hash_1", HALF_OPEN, Arrays.asList("q1_1", "q1_2", "q1_3")); buildCircuitEntry("hash_2", OPEN, Arrays.asList("q2_1", "q2_2", "q2_3")); buildCircuitEntry("hash_3", OPEN, Arrays.asList("q3_1", "q3_2", "q3_3")); buildCircuitEntry("hash_4", OPEN, Arrays.asList("q4_1", "q4_2", "q4_3")); buildCircuitEntry("hash_5", HALF_OPEN, Arrays.asList("q5_1", "q5_2", "q5_3")); context.assertEquals(2L, jedis.scard(STORAGE_HALFOPEN_CIRCUITS)); context.assertEquals(3L, jedis.scard(STORAGE_OPEN_CIRCUITS)); context.assertEquals(5L, jedis.scard(STORAGE_ALL_CIRCUITS)); for (int index = 1; index <= 5; index++) { String hash = "hash_" + index; context.assertTrue(jedis.exists(key(hash, QueueResponseType.SUCCESS))); context.assertTrue(jedis.exists(key(hash, QueueResponseType.FAILURE))); context.assertTrue(jedis.exists(queuesKey(hash))); } storage.closeAllCircuits().setHandler(event -> { context.assertTrue(event.succeeded()); context.assertEquals(0L, jedis.scard(STORAGE_HALFOPEN_CIRCUITS)); context.assertEquals(0L, jedis.scard(STORAGE_OPEN_CIRCUITS)); context.assertEquals(5L, jedis.scard(STORAGE_ALL_CIRCUITS)); context.assertEquals(15L, jedis.llen(STORAGE_QUEUES_TO_UNLOCK)); for (int index = 1; index <= 5; index++) { String hash = "hash_" + index; context.assertFalse(jedis.exists(queuesKey(hash))); assertStateAndErroPercentage(context, hash, CLOSED, 0); context.assertFalse(jedis.exists(key(hash, QueueResponseType.SUCCESS))); context.assertFalse(jedis.exists(key(hash, QueueResponseType.FAILURE))); context.assertFalse(jedis.exists(queuesKey(hash))); } assertQueuesToUnlockItems(context, Arrays.asList("q1_1", "q1_2", "q1_3", "q2_1", "q2_2", "q2_3", "q3_1", "q3_2", "q3_3", "q4_1", "q4_2", "q4_3", "q5_1", "q5_2", "q5_3")); async.complete(); }); }
@Test public void testCloseAllCircuitsEmptySets(TestContext context){ Async async = context.async(); context.assertFalse(jedis.exists(STORAGE_HALFOPEN_CIRCUITS)); context.assertFalse(jedis.exists(STORAGE_ALL_CIRCUITS)); context.assertFalse(jedis.exists(STORAGE_OPEN_CIRCUITS)); context.assertFalse(jedis.exists(STORAGE_QUEUES_TO_UNLOCK)); context.assertEquals(0L, jedis.scard(STORAGE_ALL_CIRCUITS)); context.assertEquals(0L, jedis.scard(STORAGE_HALFOPEN_CIRCUITS)); context.assertEquals(0L, jedis.scard(STORAGE_OPEN_CIRCUITS)); storage.closeAllCircuits().setHandler(event -> { context.assertTrue(event.succeeded()); context.assertEquals(0L, jedis.scard(STORAGE_ALL_CIRCUITS)); context.assertEquals(0L, jedis.scard(STORAGE_HALFOPEN_CIRCUITS)); context.assertEquals(0L, jedis.scard(STORAGE_OPEN_CIRCUITS)); context.assertEquals(0L, jedis.llen(STORAGE_QUEUES_TO_UNLOCK)); async.complete(); }); }