RedisQueueCircuitBreakerStorage implements QueueCircuitBreakerStorage { @Override public Future<UpdateStatisticsResult> updateStatistics(PatternAndCircuitHash patternAndCircuitHash, String uniqueRequestID, long timestamp, int errorThresholdPercentage, long entriesMaxAgeMS, long minQueueSampleCount, long maxQueueSampleCount, QueueResponseType queueResponseType) { Future<UpdateStatisticsResult> future = Future.future(); String circuitHash = patternAndCircuitHash.getCircuitHash(); List<String> keys = Arrays.asList( buildInfosKey(circuitHash), buildStatsKey(circuitHash, QueueResponseType.SUCCESS), buildStatsKey(circuitHash, QueueResponseType.FAILURE), buildStatsKey(circuitHash, queueResponseType), STORAGE_OPEN_CIRCUITS, STORAGE_ALL_CIRCUITS ); List<String> arguments = Arrays.asList( uniqueRequestID, patternAndCircuitHash.getPattern().pattern(), patternAndCircuitHash.getCircuitHash(), String.valueOf(timestamp), String.valueOf(errorThresholdPercentage), String.valueOf(entriesMaxAgeMS), String.valueOf(minQueueSampleCount), String.valueOf(maxQueueSampleCount) ); UpdateStatsRedisCommand cmd = new UpdateStatsRedisCommand(openCircuitLuaScriptState, 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 testUpdateStatistics(TestContext context){ Async async = context.async(); String circuitHash = "anotherCircuitHash"; PatternAndCircuitHash patternAndCircuitHash = buildPatternAndCircuitHash("/anotherCircuit", circuitHash); context.assertFalse(jedis.exists(key(circuitHash, QueueResponseType.SUCCESS))); context.assertFalse(jedis.exists(key(circuitHash, QueueResponseType.FAILURE))); context.assertFalse(jedis.exists(STORAGE_OPEN_CIRCUITS)); int errorThreshold = 50; long entriesMaxAgeMS = 10; long minQueueSampleCount = 1; long maxQueueSampleCount = 3; storage.updateStatistics(patternAndCircuitHash, "req_1", 1, errorThreshold, entriesMaxAgeMS, minQueueSampleCount, maxQueueSampleCount, QueueResponseType.SUCCESS).setHandler(event -> { context.assertTrue(event.succeeded()); context.assertTrue(jedis.exists(key(circuitHash, QueueResponseType.SUCCESS))); context.assertFalse(jedis.exists(key(circuitHash, QueueResponseType.FAILURE))); context.assertEquals(1L, jedis.zcard(key(circuitHash, QueueResponseType.SUCCESS))); context.assertEquals(UpdateStatisticsResult.OK, event.result()); storage.updateStatistics(patternAndCircuitHash, "req_2", 2, errorThreshold, entriesMaxAgeMS, minQueueSampleCount, maxQueueSampleCount, QueueResponseType.SUCCESS).setHandler(event1 -> { context.assertFalse(jedis.exists(key(circuitHash, QueueResponseType.FAILURE))); context.assertEquals(2L, jedis.zcard(key(circuitHash, QueueResponseType.SUCCESS))); context.assertEquals(UpdateStatisticsResult.OK, event1.result()); storage.updateStatistics(patternAndCircuitHash, "req_3", 3, errorThreshold, entriesMaxAgeMS, minQueueSampleCount, maxQueueSampleCount, QueueResponseType.SUCCESS).setHandler(event2 -> { context.assertFalse(jedis.exists(key(circuitHash, QueueResponseType.FAILURE))); context.assertEquals(3L, jedis.zcard(key(circuitHash, QueueResponseType.SUCCESS))); context.assertEquals(UpdateStatisticsResult.OK, event2.result()); storage.updateStatistics(patternAndCircuitHash, "req_4", 4, errorThreshold, entriesMaxAgeMS, minQueueSampleCount, maxQueueSampleCount, QueueResponseType.SUCCESS).setHandler(event3 -> { context.assertFalse(jedis.exists(key(circuitHash, QueueResponseType.FAILURE))); context.assertEquals(3L, jedis.zcard(key(circuitHash, QueueResponseType.SUCCESS))); context.assertEquals(UpdateStatisticsResult.OK, event3.result()); async.complete(); }); }); }); }); }