RedisQueueCircuitBreakerStorage implements QueueCircuitBreakerStorage { @Override public Future<QueueCircuitState> getQueueCircuitState(PatternAndCircuitHash patternAndCircuitHash) { Future<QueueCircuitState> future = Future.future(); redisClient.hget(buildInfosKey(patternAndCircuitHash.getCircuitHash()), FIELD_STATE, event -> { if(event.failed()){ future.fail(event.cause()); } else { String stateAsString = event.result(); if(StringUtils.isEmpty(stateAsString)){ log.info("No status information found for circuit " + patternAndCircuitHash.getPattern().pattern() + ". Using default value " + QueueCircuitState.CLOSED); } future.complete(QueueCircuitState.fromString(stateAsString, QueueCircuitState.CLOSED)); } }); 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 testGetQueueCircuitState(TestContext context){ Async async = context.async(); PatternAndCircuitHash patternAndCircuitHash = buildPatternAndCircuitHash("/someCircuit", "someCircuitHash"); storage.getQueueCircuitState(patternAndCircuitHash).setHandler(event -> { context.assertEquals(CLOSED, event.result()); writeQueueCircuitStateToDatabase("someCircuitHash", HALF_OPEN); storage.getQueueCircuitState(patternAndCircuitHash).setHandler(event1 -> { context.assertEquals(HALF_OPEN, event1.result()); async.complete(); }); }); }
@Test public void testGetQueueCircuitStateByHash(TestContext context){ Async async = context.async(); storage.getQueueCircuitState("someCircuitHash").setHandler(event -> { context.assertEquals(CLOSED, event.result()); writeQueueCircuitStateToDatabase("someCircuitHash", HALF_OPEN); storage.getQueueCircuitState("someCircuitHash").setHandler(event1 -> { context.assertEquals(HALF_OPEN, event1.result()); async.complete(); }); }); }