RedisQueueCircuitBreakerStorage implements QueueCircuitBreakerStorage { @Override public Future<Void> closeCircuit(PatternAndCircuitHash patternAndCircuitHash) { return closeCircuit(patternAndCircuitHash.getCircuitHash(), false); } 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 testCloseCircuit(TestContext context){ Async async = context.async(); String circuitHash = "anotherCircuitHash"; context.assertFalse(jedis.exists(key(circuitHash, QueueResponseType.SUCCESS))); context.assertFalse(jedis.exists(key(circuitHash, QueueResponseType.FAILURE))); context.assertFalse(jedis.exists(queuesKey(circuitHash))); context.assertFalse(jedis.exists(STORAGE_ALL_CIRCUITS)); context.assertFalse(jedis.exists(STORAGE_HALFOPEN_CIRCUITS)); context.assertFalse(jedis.exists(STORAGE_OPEN_CIRCUITS)); context.assertFalse(jedis.exists(STORAGE_QUEUES_TO_UNLOCK)); writeQueueCircuitStateToDatabase(circuitHash, HALF_OPEN); writeQueueCircuitFailPercentageToDatabase(circuitHash, 50); jedis.zadd(key(circuitHash, QueueResponseType.SUCCESS), 1, "req-1"); jedis.zadd(key(circuitHash, QueueResponseType.SUCCESS), 2, "req-2"); jedis.zadd(key(circuitHash, QueueResponseType.SUCCESS), 3, "req-3"); jedis.zadd(key(circuitHash, QueueResponseType.FAILURE), 4, "req-4"); jedis.zadd(key(circuitHash, QueueResponseType.FAILURE), 5, "req-5"); jedis.zadd(key(circuitHash, QueueResponseType.FAILURE), 6, "req-6"); jedis.zadd(queuesKey(circuitHash), 1, "queue_1"); jedis.zadd(queuesKey(circuitHash), 2, "queue_2"); jedis.zadd(queuesKey(circuitHash), 3, "queue_3"); addToCircuitsSets(HALF_OPEN, "a"); addToCircuitsSets(HALF_OPEN, circuitHash); addToCircuitsSets(HALF_OPEN, "b"); addToCircuitsSets(HALF_OPEN, "c"); jedis.sadd(STORAGE_OPEN_CIRCUITS, "x"); context.assertEquals(3L, jedis.zcard(key(circuitHash, QueueResponseType.SUCCESS))); context.assertEquals(3L, jedis.zcard(key(circuitHash, QueueResponseType.FAILURE))); context.assertEquals(3L, jedis.zcard(queuesKey(circuitHash))); context.assertEquals(4L, jedis.scard(STORAGE_HALFOPEN_CIRCUITS)); context.assertEquals(4L, jedis.scard(STORAGE_ALL_CIRCUITS)); context.assertEquals(1L, jedis.scard(STORAGE_OPEN_CIRCUITS)); context.assertEquals(0L, jedis.llen(STORAGE_QUEUES_TO_UNLOCK)); PatternAndCircuitHash patternAndCircuitHash = buildPatternAndCircuitHash("/anotherCircuit", circuitHash); storage.closeCircuit(patternAndCircuitHash).setHandler(event -> { context.assertTrue(event.succeeded()); context.assertFalse(jedis.exists(key(circuitHash, QueueResponseType.SUCCESS))); context.assertFalse(jedis.exists(key(circuitHash, QueueResponseType.FAILURE))); context.assertFalse(jedis.exists(queuesKey(circuitHash))); context.assertTrue(jedis.exists(STORAGE_ALL_CIRCUITS)); context.assertTrue(jedis.exists(STORAGE_HALFOPEN_CIRCUITS)); context.assertTrue(jedis.exists(STORAGE_QUEUES_TO_UNLOCK)); context.assertEquals(3L, jedis.llen(STORAGE_QUEUES_TO_UNLOCK)); context.assertEquals("queue_1", jedis.lpop(STORAGE_QUEUES_TO_UNLOCK)); context.assertEquals("queue_2", jedis.lpop(STORAGE_QUEUES_TO_UNLOCK)); context.assertEquals("queue_3", jedis.lpop(STORAGE_QUEUES_TO_UNLOCK)); context.assertTrue(jedis.exists(infosKey(circuitHash))); assertStateAndErroPercentage(context, circuitHash, CLOSED, 0); context.assertEquals(3L, jedis.scard(STORAGE_HALFOPEN_CIRCUITS)); context.assertEquals(4L, jedis.scard(STORAGE_ALL_CIRCUITS)); Set<String> halfOpenCircuits = jedis.smembers(STORAGE_HALFOPEN_CIRCUITS); context.assertTrue(halfOpenCircuits.contains("a")); context.assertTrue(halfOpenCircuits.contains("b")); context.assertTrue(halfOpenCircuits.contains("c")); context.assertFalse(halfOpenCircuits.contains(circuitHash)); Set<String> allCircuits = jedis.smembers(STORAGE_ALL_CIRCUITS); context.assertTrue(allCircuits.contains("a")); context.assertTrue(allCircuits.contains("b")); context.assertTrue(allCircuits.contains("c")); context.assertTrue(allCircuits.contains(circuitHash)); context.assertEquals(1L, jedis.scard(STORAGE_OPEN_CIRCUITS)); Set<String> openCircuits = jedis.smembers(STORAGE_OPEN_CIRCUITS); context.assertTrue(openCircuits.contains("x")); async.complete(); }); }