private static <T> void openCircuitBreakerAndHandleError( Consumer<Throwable> errorHandler, ThrowableErrorConsumer<Throwable, T> onFailureRespond, Consumer<Throwable> errorMethodHandler, Consumer<ExecutionResult<T>> resultConsumer, AsyncResult<T> event, Lock lock, Counter counter) { counter.addAndGet( LOCK_VALUE, val -> { lock.release(); errorHandling( errorHandler, onFailureRespond, errorMethodHandler, resultConsumer, Future.failedFuture(event.cause())); }); }
private static <T> void openCircuitBreakerAndHandleError( Future<ExecutionResult<T>> _blockingHandler, Consumer<Throwable> _errorHandler, ThrowableFunction<Throwable, T> _onFailureRespond, Consumer<Throwable> _errorMethodHandler, VxmsShared vxmsShared, Throwable e, Lock lck, Counter counter) { counter.addAndGet( LOCK_VALUE, val -> { lck.release(); final Vertx vertx = vxmsShared.getVertx(); vertx.executeBlocking( bhandler -> { T result = handleError(_errorHandler, _onFailureRespond, _errorMethodHandler, e); if (!_blockingHandler.isComplete()) { _blockingHandler.complete(new ExecutionResult<>(result, true, true, null)); } }, false, res -> {}); }); }
private static <T> void openCircuitBreakerAndHandleError( Future<ExecutionResult<T>> _resultHandler, Consumer<Throwable> _errorHandler, ThrowableFunction<Throwable, T> _onFailureRespond, Consumer<Throwable> _errorMethodHandler, VxmsShared vxmsShared, Throwable e, Lock lck, Counter counter) { counter.addAndGet( LOCK_VALUE, val -> { lck.release(); final Vertx vertx = vxmsShared.getVertx(); vertx.executeBlocking( bhandler -> { T result = handleError(_errorHandler, _onFailureRespond, _errorMethodHandler, e); if (!_resultHandler.isComplete()) { _resultHandler.complete(new ExecutionResult<>(result, true, true, null)); } }, false, res -> {}); }); }
@Test public void testAtomicSyncCounter3() { long max = 3; vertx .sharedData() .getCounter( "counter1", resultHandler -> { final Counter counter = resultHandler.result(); counter.get( handler -> { long val = handler.result(); if (val == 0) { counter.addAndGet( max, hhh -> { System.out.println("::::" + hhh.result()); }); } else { } }); }); }
@Override public void getCounter(final String name, final Handler<AsyncResult<Counter>> resultHandler) { AtomicLong counter = new AtomicLong(); final AtomicLong prev = counters.putIfAbsent(name, counter); if (prev != null) { counter = prev; } final AtomicLong theCounter = counter; final Context context = this.vertx.getOrCreateContext(); context.runOnContext(v -> resultHandler.handle(Future.succeededFuture(new AsynchronousCounter(this.vertx, theCounter)))); }
@Override public void getCounter(String name, Handler<AsyncResult<Counter>> resultHandler) { vertx.executeBlocking(future -> { try { Objects.requireNonNull(name); future.complete(new ZKCounter(name, retryPolicy)); } catch (Exception e) { future.fail(new VertxException(e)); } }, resultHandler); }
private static <T> void executeInitialState( int _retry, long _timeout, ThrowableFutureConsumer<T> _userOperation, VxmsShared vxmsShared, Future<T> operationResult, Lock lock, Counter counter) { final long initialRetryCounterValue = (long) (_retry + 1); counter.addAndGet( initialRetryCounterValue, rHandler -> executeDefaultState(_timeout, _userOperation, vxmsShared, operationResult, lock)); }
private static <T> void handleStatefulError( String _methodId, int _retry, long _timeout, long _circuitBreakerTimeout, ThrowableFutureConsumer<T> _userOperation, Consumer<Throwable> errorHandler, ThrowableErrorConsumer<Throwable, T> onFailureRespond, Consumer<Throwable> errorMethodHandler, VxmsShared vxmsShared, Consumer<ExecutionResult<T>> resultConsumer, AsyncResult<T> event, Lock lock, Counter counter, AsyncResult<Long> valHandler) { long count = valHandler.result(); if (count <= DEFAULT_LONG_VALUE) { setCircuitBreakerReleaseTimer(_retry, _circuitBreakerTimeout, vxmsShared, counter); openCircuitBreakerAndHandleError( errorHandler, onFailureRespond, errorMethodHandler, resultConsumer, event, lock, counter); } else { lock.release(); retry( _methodId, _retry, _timeout, _circuitBreakerTimeout, _userOperation, errorHandler, onFailureRespond, errorMethodHandler, vxmsShared, resultConsumer, event); } }
private static <T, V> void executeInitialState( int _retry, long _timeout, ThrowableFutureBiConsumer<T, V> step, T inputValue, VxmsShared vxmsShared, Future<V> operationResult, Lock lock, Counter counter) { final long initialRetryCounterValue = (long) (_retry + 1); counter.addAndGet( initialRetryCounterValue, rHandler -> executeDefaultState(_timeout, step, inputValue, vxmsShared, operationResult, lock)); }
private static <T, V> void handleStatefulError( String _methodId, int _retry, long _timeout, long _circuitBreakerTimeout, ThrowableFutureBiConsumer<T, V> step, T inputValue, Consumer<Throwable> errorHandler, ThrowableErrorConsumer<Throwable, V> onFailureRespond, Consumer<Throwable> errorMethodHandler, VxmsShared vxmsShared, Consumer<ExecutionResult<V>> resultConsumer, AsyncResult<V> event, Lock lock, Counter counter, AsyncResult<Long> valHandler) { long count = valHandler.result(); if (count <= DEFAULT_LONG_VALUE) { setCircuitBreakerReleaseTimer(_retry, _circuitBreakerTimeout, vxmsShared, counter); openCircuitBreakerAndHandleError( errorHandler, onFailureRespond, errorMethodHandler, resultConsumer, event, lock, counter); } else { lock.release(); retry( _methodId, _retry, _timeout, _circuitBreakerTimeout, step, inputValue, errorHandler, onFailureRespond, errorMethodHandler, vxmsShared, resultConsumer, event); } }
private static <T> void executeInitialState( String _methodId, ThrowableSupplier<T> _supplier, Future<ExecutionResult<T>> _blockingHandler, Consumer<Throwable> _errorHandler, ThrowableFunction<Throwable, T> _onFailureRespond, Consumer<Throwable> _errorMethodHandler, VxmsShared vxmsShared, Throwable _t, int _retry, long _timeout, long _circuitBreakerTimeout, long _delay, Lock lock, Counter counter) { final long initialRetryCounterValue = (long) (_retry + 1); counter.addAndGet( initialRetryCounterValue, rHandler -> executeDefault( _methodId, _supplier, _blockingHandler, _errorHandler, _onFailureRespond, _errorMethodHandler, vxmsShared, _t, _retry, _timeout, _circuitBreakerTimeout, _delay, lock)); }
private static void setCircuitBreakerReleaseTimer( VxmsShared vxmsShared, int _retry, long _circuitBreakerTimeout, Counter counter) { final long initialRetryCounterValue = (long) (_retry + 1); final Vertx vertx = vxmsShared.getVertx(); vertx.setTimer( _circuitBreakerTimeout, timer -> counter.addAndGet(initialRetryCounterValue, val -> {})); }
private static <T, V> void executeInitialState( String _methodId, ThrowableFunction<T, V> step, T value, Future<ExecutionResult<V>> _resultHandler, Consumer<Throwable> _errorHandler, ThrowableFunction<Throwable, V> _onFailureRespond, Consumer<Throwable> _errorMethodHandler, VxmsShared vxmsShared, Throwable _t, int _retry, long _timeout, long _circuitBreakerTimeout, long _delay, Lock lock, Counter counter) { final long initialRetryCounterValue = (long) (_retry + 1); counter.addAndGet( initialRetryCounterValue, rHandler -> executeDefault( _methodId, step, value, _resultHandler, _errorHandler, _onFailureRespond, _errorMethodHandler, vxmsShared, _t, _retry, _timeout, _circuitBreakerTimeout, _delay, lock)); }
private static <T> void openCircuitAndHandleError( String methodId, VxmsShared vxmsShared, Consumer<Throwable> errorMethodHandler, Message<Object> requestMessage, Encoder encoder, Consumer<Throwable> errorHandler, ThrowableErrorConsumer<Throwable, T> onFailureRespond, DeliveryOptions responseDeliveryOptions, int retryCount, long timeout, long circuitBreakerTimeout, RecursiveExecutor<T> executor, AsyncResult<Message<Object>> event, Lock lock, Counter counter) { resetLockTimer(vxmsShared, retryCount, circuitBreakerTimeout, counter); lockAndHandle( counter, val -> { final Throwable cause = event.cause(); handleError( methodId, vxmsShared, errorMethodHandler, requestMessage, encoder, errorHandler, onFailureRespond, responseDeliveryOptions, retryCount, timeout, circuitBreakerTimeout, executor, lock, cause); }); }
private static void resetLockTimer( VxmsShared vxmsShared, int retryCount, long circuitBreakerTimeout, Counter counter) { final Vertx vertx = vxmsShared.getVertx(); vertx.setTimer( circuitBreakerTimeout, timer -> counter.addAndGet(Integer.valueOf(retryCount + 1).longValue(), val -> {})); }
private static <T> void openCircuitAndHandleError( String methodId, VxmsShared vxmsShared, Consumer<Throwable> errorMethodHandler, Message<Object> requestMessage, Encoder encoder, Consumer<Throwable> errorHandler, ThrowableFunction<Throwable, T> onFailureRespond, DeliveryOptions responseDeliveryOptions, int retryCount, long timeout, long delay, long circuitBreakerTimeout, RecursiveExecutor<T> executor, AsyncResult<Message<Object>> event, Lock lock, Counter counter) { resetLockTimer(vxmsShared, retryCount, circuitBreakerTimeout, counter); lockAndHandle( counter, val -> { final Throwable cause = event.cause(); handleError( methodId, vxmsShared, errorMethodHandler, requestMessage, encoder, errorHandler, onFailureRespond, responseDeliveryOptions, retryCount, timeout, delay, circuitBreakerTimeout, executor, lock, cause); }); }
/** * Get a local counter. The counter will be passed to the handler. * * @param name the name of the counter. * @param resultHandler the handler */ public void getCounter(String name, Handler<AsyncResult<Counter>> resultHandler) { Objects.requireNonNull(name, "name"); Objects.requireNonNull(resultHandler, "resultHandler"); Counter counter = this.localCounters .computeIfAbsent(name, (n) -> new AsynchronousCounter((VertxInternal) this.vertx)); Context context = this.vertx.getOrCreateContext(); context.runOnContext((v) -> resultHandler.handle(Future.succeededFuture(counter))); }
private static <T> void executeInitialState( int retry, long timeout, ThrowableFutureConsumer<T> _userOperation, VxmsShared vxmsShared, Future<T> operationResult, Lock lock, Counter counter) { final long initialRetryCounterValue = (long) (retry + 1); counter.addAndGet( initialRetryCounterValue, rHandler -> executeDefaultState(timeout, _userOperation, vxmsShared, operationResult, lock)); }
private static <T> void handleStatefulError( String methodId, int retry, long timeout, long circuitBreakerTimeout, ThrowableFutureConsumer<T> _userOperation, Consumer<Throwable> errorHandler, ThrowableErrorConsumer<Throwable, T> onFailureRespond, Consumer<Throwable> errorMethodHandler, VxmsShared vxmsShared, Consumer<ExecutionResult<T>> resultConsumer, AsyncResult<T> event, Lock lock, Counter counter, AsyncResult<Long> valHandler) { long count = valHandler.result(); if (count <= DEFAULT_LONG_VALUE) { setCircuitBreakerReleaseTimer(retry, circuitBreakerTimeout, vxmsShared.getVertx(), counter); openCircuitBreakerAndHandleError( errorHandler, onFailureRespond, errorMethodHandler, resultConsumer, event, lock, counter); } else { lock.release(); retry( methodId, retry, timeout, circuitBreakerTimeout, _userOperation, errorHandler, onFailureRespond, errorMethodHandler, vxmsShared, resultConsumer, event); } }
private static <T, V> void executeInitialState( int retry, long timeout, ThrowableFutureBiConsumer<T, V> _step, T _inputValue, VxmsShared vxmsShared, Future<V> operationResult, Lock lock, Counter counter) { final long initialRetryCounterValue = (long) (retry + 1); counter.addAndGet( initialRetryCounterValue, rHandler -> executeDefaultState(timeout, _step, _inputValue, vxmsShared, operationResult, lock)); }
private static <T, V> void handleStatefulError( String methodId, int retry, long timeout, long circuitBreakerTimeout, ThrowableFutureBiConsumer<T, V> _step, T _inputValue, Consumer<Throwable> errorHandler, ThrowableErrorConsumer<Throwable, V> onFailureRespond, Consumer<Throwable> errorMethodHandler, VxmsShared vxmsShared, Consumer<ExecutionResult<V>> resultConsumer, AsyncResult<V> event, Lock lock, Counter counter, AsyncResult<Long> valHandler) { long count = valHandler.result(); if (count <= DEFAULT_LONG_VALUE) { setCircuitBreakerReleaseTimer(retry, circuitBreakerTimeout, vxmsShared.getVertx(), counter); openCircuitBreakerAndHandleError( errorHandler, onFailureRespond, errorMethodHandler, resultConsumer, event, lock, counter); } else { lock.release(); retry( methodId, retry, timeout, circuitBreakerTimeout, _step, _inputValue, errorHandler, onFailureRespond, errorMethodHandler, vxmsShared, resultConsumer, event); } }
private static void setCircuitBreakerReleaseTimer( VxmsShared vxmsShared, int _retry, long _circuitBreakerTimeout, Counter counter) { final Vertx vertx = vxmsShared.getVertx(); vertx.setTimer( _circuitBreakerTimeout, timer -> { final long initialRetryCounterValue = (long) (_retry + 1); counter.addAndGet(initialRetryCounterValue, val -> {}); }); }
private static <T, V> void executeInitialState( String _methodId, ThrowableFunction<T, V> step, T value, Future<ExecutionResult<V>> _blockingHandler, Consumer<Throwable> _errorHandler, ThrowableFunction<Throwable, V> _onFailureRespond, Consumer<Throwable> _errorMethodHandler, VxmsShared vxmsShared, Throwable _t, int _retry, long _timeout, long _circuitBreakerTimeout, long _delay, Lock lock, Counter counter) { final long initialRetryCounterValue = (long) (_retry + 1); counter.addAndGet( initialRetryCounterValue, rHandler -> executeDefault( _methodId, step, value, _blockingHandler, _errorHandler, _onFailureRespond, _errorMethodHandler, vxmsShared, _t, _retry, _timeout, _circuitBreakerTimeout, _delay, lock)); }
@Override public void getCounter(String name, Handler<AsyncResult<Counter>> handler) { logTrace(() -> String.format("Create new counter [%s] on address [%s]", name, address)); checkCluster(); vertx.executeBlocking( future -> future.complete(new ClusteredCounterImpl(vertx, counterService.getOrCreateCounter(name, 0L))), handler ); }
@Override public void getCounter(String name, Handler<AsyncResult<Counter>> resultHandler) { vertx.executeBlocking(fut -> fut.complete( USE_HZ_ASYNC_API ? new HazelcastInternalAsyncCounter(vertx, hazelcast.getAtomicLong(name)) : new HazelcastCounter(hazelcast.getAtomicLong(name)) ) , resultHandler); }
@Override public void getCounter(String name, Handler<AsyncResult<Counter>> handler) { vertx.executeBlocking(fut -> fut.complete(new CounterImpl(ignite.atomicLong(name, 0, true))), handler); }
@Override public void getCounter(String name, Handler<AsyncResult<Counter>> resultHandler) { vertx.executeBlocking(future -> { future.complete(new JGroupsCounter(vertx, counterService.getOrCreateCounter(name, 0))); }, false, resultHandler); }
public JGroupsCounter(Vertx vertx, org.jgroups.blocks.atomic.Counter jgroupsCounter) { this.vertx = vertx; this.jgroupsCounter = jgroupsCounter; }
private static void setCircuitBreakerReleaseTimer( int _retry, long _release, VxmsShared vxmsShared, Counter counter) { final long initialRetryCounterValue = (long) (_retry + 1); final Vertx vertx = vxmsShared.getVertx(); vertx.setTimer(_release, timer -> counter.addAndGet(initialRetryCounterValue, val -> {})); }
