github-forks
/
redisson
mirror of https://github.com/redisson/redisson.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge branch 'master' into 3.0.0

Browse Source
pull/1303/head
Nikita 7 years ago
parent f8c076d7b0 9882f72e69
commit 8cdac0de5a
5 changed files with 190 additions and 62 deletions
Show Stats Download Patch File Download Diff File

157
redisson/src/main/java/org/redisson/RedissonBaseAdder.java
Unescape Escape View File

@ -15,6 +15,9 @@
*/
package org.redisson;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.redisson.api.RFuture;
import org.redisson.api.RSemaphore;
import org.redisson.api.RTopic;
@ -37,6 +40,83 @@ import io.netty.util.concurrent.FutureListener;
*/
public abstract class RedissonBaseAdder<T extends Number> extends RedissonExpirable {
private class ResetListener implements FutureListener<Long> {
private final RPromise<Void> result;
private ResetListener(RPromise<Void> result) {
this.result = result;
}
@Override
public void operationComplete(Future<Long> future) throws Exception {
if (!future.isSuccess()) {
result.tryFailure(future.cause());
return;
}
acquireAsync(future.getNow().intValue()).addListener(new FutureListener<Void>() {
@Override
public void operationComplete(Future<Void> future) throws Exception {
if (!future.isSuccess()) {
result.tryFailure(future.cause());
return;
}
result.trySuccess(null);
}
});
}
protected RFuture<Void> acquireAsync(int value) {
return semaphore.acquireAsync(value);
}
}
private class SumListener implements FutureListener<Long> {
private final RPromise<T> result;
private SumListener(RPromise<T> result) {
this.result = result;
}
@Override
public void operationComplete(Future<Long> future) throws Exception {
if (!future.isSuccess()) {
result.tryFailure(future.cause());
return;
}
acquireAsync(future.getNow().intValue()).addListener(new FutureListener<Void>() {
@Override
public void operationComplete(Future<Void> future) throws Exception {
if (!future.isSuccess()) {
result.tryFailure(future.cause());
return;
}
RFuture<T> valueFuture = getAndDeleteAsync();
valueFuture.addListener(new FutureListener<T>() {
@Override
public void operationComplete(Future<T> future) throws Exception {
if (!future.isSuccess()) {
result.tryFailure(future.cause());
return;
}
result.trySuccess(future.getNow());
}
});
}
});
}
protected RFuture<Void> acquireAsync(int value) {
return semaphore.acquireAsync(value);
}
}
private static final Logger log = LoggerFactory.getLogger(RedissonBaseAdder.class);
private static final long CLEAR_MSG = 0;
@ -102,76 +182,71 @@ public abstract class RedissonBaseAdder<T extends Number> extends RedissonExpira
get(resetAsync());
}
public void reset(long timeout, TimeUnit timeUnit) {
get(resetAsync(timeout, timeUnit));
}
public RFuture<T> sumAsync() {
final RPromise<T> result = new RedissonPromise<T>();
RFuture<Long> future = topic.publishAsync(SUM_MSG);
future.addListener(new FutureListener<Long>() {
future.addListener(new SumListener(result));
@Override
public void operationComplete(Future<Long> future) throws Exception {
if (!future.isSuccess()) {
result.tryFailure(future.cause());
return;
return result;
}
semaphore.acquireAsync(future.getNow().intValue()).addListener(new FutureListener<Void>() {
public RFuture<T> sumAsync(final long timeout, final TimeUnit timeUnit) {
RPromise<T> result = new RedissonPromise<T>();
RFuture<Long> future = topic.publishAsync(SUM_MSG);
future.addListener(new SumListener(result) {
@Override
public void operationComplete(Future<Void> future) throws Exception {
if (!future.isSuccess()) {
result.tryFailure(future.cause());
return;
protected RFuture<Void> acquireAsync(int value) {
return tryAcquire(timeout, timeUnit, value);
}
RFuture<T> valueFuture = getAndDeleteAsync();
valueFuture.addListener(new FutureListener<T>() {
});
return result;
}
protected RFuture<Void> tryAcquire(long timeout, TimeUnit timeUnit, int value) {
final RPromise<Void> acquireResult = new RedissonPromise<Void>();
semaphore.tryAcquireAsync(value, timeout, timeUnit).addListener(new FutureListener<Boolean>() {
@Override
public void operationComplete(Future<T> future) throws Exception {
public void operationComplete(Future<Boolean> future) throws Exception {
if (!future.isSuccess()) {
result.tryFailure(future.cause());
acquireResult.tryFailure(future.cause());
return;
}
result.trySuccess(future.getNow());
if (future.getNow()) {
acquireResult.trySuccess(null);
} else {
acquireResult.tryFailure(new TimeoutException());
}
});
}
});
}
});
return result;
return acquireResult;
}
public RFuture<Void> resetAsync() {
final RPromise<Void> result = new RedissonPromise<Void>();
RFuture<Long> future = topic.publishAsync(CLEAR_MSG);
future.addListener(new FutureListener<Long>() {
future.addListener(new ResetListener(result));
@Override
public void operationComplete(Future<Long> future) throws Exception {
if (!future.isSuccess()) {
result.tryFailure(future.cause());
return;
return result;
}
int value = 0;
if (future.getNow() != null) {
value = future.getNow().intValue();
}
public RFuture<Void> resetAsync(final long timeout, final TimeUnit timeUnit) {
final RPromise<Void> result = new RedissonPromise<Void>();
semaphore.acquireAsync(value).addListener(new FutureListener<Void>() {
RFuture<Long> future = topic.publishAsync(CLEAR_MSG);
future.addListener(new ResetListener(result) {
@Override
public void operationComplete(Future<Void> future) throws Exception {
if (!future.isSuccess()) {
result.tryFailure(future.cause());
return;
}
result.trySuccess(null);
}
});
protected RFuture<Void> acquireAsync(int value) {
return tryAcquire(timeout, timeUnit, value);
}
});

36
redisson/src/main/java/org/redisson/api/RDoubleAdder.java
Unescape Escape View File

@ -15,8 +15,10 @@
*/
package org.redisson.api;
import java.util.concurrent.TimeUnit;
/**
* Distributed implementation of {@link java.util.concurrent.atomic.LongAdder}
* Distributed implementation of {@link java.util.concurrent.atomic.DoubleAdder}
* <p>
* Internal state maintained on client side.
*
@ -43,28 +45,52 @@ public interface RDoubleAdder extends RExpirable, RDestroyable {
void decrement();
/**
* Accumulates sum across all RLongAdder instances
* Accumulates sum across all RDoubleAdder instances
*
* @return accumulated sum
*/
double sum();
/**
* Resets value across all RLongAdder instances
* Resets value across all RDoubleAdder instances
*/
void reset();
/**
* Accumulates sum across all RLongAdder instances
* Accumulates sum across all RDoubleAdder instances
*
* @return accumulated sum
*/
RFuture<Double> sumAsync();
/**
* Accumulates sum across all RDoubleAdder instances
* within defined <code>timeout</code>.
*
* @param timeout for accumulation
* @param timeUnit for timeout
*
* @return accumulated sum
*/
RFuture<Double> sumAsync(long timeout, TimeUnit timeUnit);
/**
* Resets value across all RLongAdder instances
* Resets value across all RDoubleAdder instances
*
* @return void
*/
RFuture<Void> resetAsync();
/**
* Resets value across all RDoubleAdder instances
* within defined <code>timeout</code>.
*
* @param timeout for reset
* @param timeUnit for timeout
*
* @return void
*/
RFuture<Void> resetAsync(long timeout, TimeUnit timeUnit);
}

25
redisson/src/main/java/org/redisson/api/RLongAdder.java
Unescape Escape View File

@ -15,6 +15,8 @@
*/
package org.redisson.api;
import java.util.concurrent.TimeUnit;
/**
* Distributed implementation of {@link java.util.concurrent.atomic.LongAdder}
* <p>
@ -61,10 +63,33 @@ public interface RLongAdder extends RExpirable, RDestroyable {
*/
RFuture<Long> sumAsync();
/**
* Accumulates sum across all RLongAdder instances
* within defined <code>timeout</code>.
*
* @param timeout for accumulation
* @param timeUnit for timeout
*
* @return accumulated sum
*/
RFuture<Long> sumAsync(long timeout, TimeUnit timeUnit);
/**
* Resets value across all RLongAdder instances
*
* @return void
*/
RFuture<Void> resetAsync();
/**
* Resets value across all RLongAdder instances
* within defined <code>timeout</code>.
*
* @param timeout for reset
* @param timeUnit for timeout
*
* @return void
*/
RFuture<Void> resetAsync(long timeout, TimeUnit timeUnit);
}

4
redisson/src/main/java/org/redisson/command/CommandAsyncService.java
Unescape Escape View File

@ -294,7 +294,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
return allAsync(true, command, callback, params);
}
private <T, R> RFuture<R> allAsync(boolean readOnlyMode, RedisCommand<T> command, final SlotCallback<T, R> callback, Object... params) {
private <T, R> RFuture<R> allAsync(boolean readOnlyMode, final RedisCommand<T> command, final SlotCallback<T, R> callback, Object... params) {
final RPromise<R> mainPromise = new RedissonPromise<R>();
final Collection<MasterSlaveEntry> nodes = connectionManager.getEntrySet();
final AtomicInteger counter = new AtomicInteger(nodes.size());
@ -788,7 +788,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
}
private <R, V> void checkAttemptFuture(final NodeSource source, final AsyncDetails<V, R> details,
Future<R> future, boolean ignoreRedirect) {
Future<R> future, final boolean ignoreRedirect) {
details.getTimeout().cancel();
if (future.isCancelled()) {
return;

8
redisson/src/main/java/org/redisson/spring/cache/RedissonCache.java vendored
Unescape Escape View File

@ -16,6 +16,7 @@
package org.redisson.spring.cache;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@ -164,15 +165,16 @@ public class RedissonCache implements Cache {
if (value == null) {
try {
value = toStoreValue(valueLoader.call());
} catch (Exception ex) {
} catch (Throwable ex) {
RuntimeException exception;
try {
Class<?> c = Class.forName("org.springframework.cache.Cache$ValueRetrievalException");
Constructor<?> constructor = c.getConstructor(Object.class, Callable.class, Throwable.class);
RuntimeException exception = (RuntimeException) constructor.newInstance(key, valueLoader, ex.getCause());
throw exception;
exception = (RuntimeException) constructor.newInstance(key, valueLoader, ex);
} catch (Exception e) {
throw new IllegalStateException(e);
}
throw exception;
}
put(key, value);
}

Write Preview
Loading…
Cancel
Save