RedisQueueCircuitBreakerStorage implements QueueCircuitBreakerStorage { @Override public Future<List<String>> unlockSampleQueues() { Future<List<String>> future = Future.future(); List<String> keys = Collections.singletonList(STORAGE_HALFOPEN_CIRCUITS); List<String> arguments = Arrays.asList( STORAGE_PREFIX, STORAGE_QUEUES_SUFFIX, String.valueOf(System.currentTimeMillis())); UnlockSampleQueuesRedisCommand cmd = new UnlockSampleQueuesRedisCommand(unlockSampleQueuesLuaScriptState, keys, arguments, redisClient, log, future); cmd.exec(0); 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 testUnlockSampleQueues(TestContext context){ Async async = context.async(); String c1 = "circuit_1"; String c2 = "circuit_2"; String c3 = "circuit_3"; context.assertFalse(jedis.exists(STORAGE_ALL_CIRCUITS)); context.assertFalse(jedis.exists(STORAGE_HALFOPEN_CIRCUITS)); context.assertFalse(jedis.exists(queuesKey(c1))); context.assertFalse(jedis.exists(queuesKey(c2))); context.assertFalse(jedis.exists(queuesKey(c3))); addToCircuitsSets(HALF_OPEN, c1); addToCircuitsSets(HALF_OPEN, c2); addToCircuitsSets(HALF_OPEN, c3); jedis.zadd(queuesKey(c1), 1, "c1_1"); jedis.zadd(queuesKey(c1), 2, "c1_2"); jedis.zadd(queuesKey(c1), 3, "c1_3"); jedis.zadd(queuesKey(c2), 1, "c2_1"); jedis.zadd(queuesKey(c2), 2, "c2_2"); jedis.zadd(queuesKey(c3), 1, "c3_1"); jedis.zadd(queuesKey(c3), 2, "c3_2"); jedis.zadd(queuesKey(c3), 3, "c3_3"); jedis.zadd(queuesKey(c3), 4, "c3_4"); context.assertEquals(3L, jedis.scard(STORAGE_ALL_CIRCUITS)); context.assertEquals(3L, jedis.scard(STORAGE_HALFOPEN_CIRCUITS)); context.assertEquals(3L, jedis.zcard(queuesKey(c1))); context.assertEquals(2L, jedis.zcard(queuesKey(c2))); context.assertEquals(4L, jedis.zcard(queuesKey(c3))); storage.unlockSampleQueues().setHandler(event -> { context.assertTrue(event.succeeded()); context.assertTrue(event.result().contains("c1_1")); context.assertTrue(event.result().contains("c2_1")); context.assertTrue(event.result().contains("c3_1")); context.assertEquals(3L, jedis.scard(STORAGE_ALL_CIRCUITS)); context.assertEquals(3L, jedis.scard(STORAGE_HALFOPEN_CIRCUITS)); context.assertEquals(3L, jedis.zcard(queuesKey(c1))); context.assertEquals(2L, jedis.zcard(queuesKey(c2))); context.assertEquals(4L, jedis.zcard(queuesKey(c3))); storage.unlockSampleQueues().setHandler(event1 -> { context.assertTrue(event1.result().contains("c1_2")); context.assertTrue(event1.result().contains("c2_2")); context.assertTrue(event1.result().contains("c3_2")); context.assertEquals(3L, jedis.scard(STORAGE_ALL_CIRCUITS)); context.assertEquals(3L, jedis.scard(STORAGE_HALFOPEN_CIRCUITS)); context.assertEquals(3L, jedis.zcard(queuesKey(c1))); context.assertEquals(2L, jedis.zcard(queuesKey(c2))); context.assertEquals(4L, jedis.zcard(queuesKey(c3))); storage.unlockSampleQueues().setHandler(event2 -> { context.assertTrue(event2.result().contains("c1_3")); context.assertTrue(event2.result().contains("c2_1")); context.assertTrue(event2.result().contains("c3_3")); context.assertEquals(3L, jedis.scard(STORAGE_ALL_CIRCUITS)); context.assertEquals(3L, jedis.scard(STORAGE_HALFOPEN_CIRCUITS)); context.assertEquals(3L, jedis.zcard(queuesKey(c1))); context.assertEquals(2L, jedis.zcard(queuesKey(c2))); context.assertEquals(4L, jedis.zcard(queuesKey(c3))); async.complete(); }); }); }); }