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

Code formatted according to checkstyle rules

Browse Source
pull/1923/head
Nikita Koksharov 6 years ago
parent 99d67bca19
commit 7126b0bef0
6 changed files with 61 additions and 61 deletions
Show Stats Download Patch File Download Diff File

5
redisson/pom.xml
Unescape Escape View File

@ -355,7 +355,7 @@
<execution>
<phase>verify</phase>
<goals>
<goal>checkstyle</goal>
<goal>check</goal>
</goals>
</execution>
</executions>
@ -365,7 +365,8 @@
**/org/redisson/cache/**/*.java,
**/org/redisson/client/**/*.java,
**/org/redisson/cluster/**/*.java,
**/org/redisson/codec/**/*.java
**/org/redisson/codec/**/*.java,
**/org/redisson/command/**/*.java
</includes>
<consoleOutput>true</consoleOutput>
<enableRSS>false</enableRSS>

2
redisson/src/main/java/org/redisson/client/codec/BaseCodec.java
Unescape Escape View File

@ -30,7 +30,7 @@ import org.redisson.jcache.JCacheEventCodec;
*/
public abstract class BaseCodec implements Codec {
private static final List<Class<?>> SKIPPED_CODECS = Arrays.asList(StringCodec.class,
public static final List<Class<?>> SKIPPED_CODECS = Arrays.asList(StringCodec.class,
ByteArrayCodec.class, LocalCachedMessageCodec.class, BitSetCodec.class, JCacheEventCodec.class);
public static Codec copy(ClassLoader classLoader, Codec codec) {

3
redisson/src/main/java/org/redisson/command/AsyncDetails.java
Unescape Escape View File

@ -15,7 +15,6 @@
*/
package org.redisson.command;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.function.BiConsumer;
import org.redisson.api.RFuture;
@ -32,8 +31,6 @@ import io.netty.util.Timeout;
public class AsyncDetails<V, R> {
static final ConcurrentLinkedQueue<AsyncDetails> queue = new ConcurrentLinkedQueue<AsyncDetails>();
RFuture<RedisConnection> connectionFuture;
ConnectionManager connectionManager;
RPromise<R> attemptPromise;

6
redisson/src/main/java/org/redisson/command/CommandAsyncExecutor.java
Unescape Escape View File

@ -51,13 +51,13 @@ public interface CommandAsyncExecutor {
boolean isRedissonReferenceSupportEnabled();
<V> RedisException convertException(RFuture<V> RFuture);
<V> RedisException convertException(RFuture<V> future);
boolean await(RFuture<?> RFuture, long timeout, TimeUnit timeoutUnit) throws InterruptedException;
boolean await(RFuture<?> future, long timeout, TimeUnit timeoutUnit) throws InterruptedException;
void syncSubscription(RFuture<?> future);
<V> V get(RFuture<V> RFuture);
<V> V get(RFuture<V> future);
<T, R> RFuture<R> writeAsync(MasterSlaveEntry entry, Codec codec, RedisCommand<T> command, Object... params);

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

@ -44,7 +44,6 @@ import org.redisson.api.RedissonClient;
import org.redisson.api.RedissonReactiveClient;
import org.redisson.api.RedissonRxClient;
import org.redisson.cache.LRUCacheMap;
import org.redisson.cache.LocalCachedMessageCodec;
import org.redisson.cache.ReferenceCacheMap;
import org.redisson.client.RedisAskException;
import org.redisson.client.RedisClient;
@ -57,8 +56,7 @@ import org.redisson.client.RedisResponseTimeoutException;
import org.redisson.client.RedisTimeoutException;
import org.redisson.client.RedisTryAgainException;
import org.redisson.client.WriteRedisConnectionException;
import org.redisson.client.codec.BitSetCodec;
import org.redisson.client.codec.ByteArrayCodec;
import org.redisson.client.codec.BaseCodec;
import org.redisson.client.codec.Codec;
import org.redisson.client.codec.StringCodec;
import org.redisson.client.protocol.CommandData;
@ -75,7 +73,6 @@ import org.redisson.connection.ConnectionManager;
import org.redisson.connection.MasterSlaveEntry;
import org.redisson.connection.NodeSource;
import org.redisson.connection.NodeSource.Redirect;
import org.redisson.jcache.JCacheEventCodec;
import org.redisson.liveobject.core.RedissonObjectBuilder;
import org.redisson.misc.LogHelper;
import org.redisson.misc.RPromise;
@ -167,7 +164,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
try {
int timeout = config.getTimeout() + config.getRetryInterval() * config.getRetryAttempts();
if (!future.await(timeout)) {
((RPromise<?>)future).tryFailure(new RedisTimeoutException("Subscribe timeout: (" + timeout + "ms). Increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters."));
((RPromise<?>) future).tryFailure(new RedisTimeoutException("Subscribe timeout: (" + timeout + "ms). Increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters."));
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@ -517,9 +514,9 @@ public class CommandAsyncService implements CommandAsyncExecutor {
if (!keys.isEmpty()) {
Object key = keys.get(0);
if (key instanceof byte[]) {
return writeAsync((byte[])key, StringCodec.INSTANCE, RedisCommands.SCRIPT_LOAD, script);
return writeAsync((byte[]) key, StringCodec.INSTANCE, RedisCommands.SCRIPT_LOAD, script);
}
return writeAsync((String)key, StringCodec.INSTANCE, RedisCommands.SCRIPT_LOAD, script);
return writeAsync((String) key, StringCodec.INSTANCE, RedisCommands.SCRIPT_LOAD, script);
}
return writeAllAsync(RedisCommands.SCRIPT_LOAD, new SlotCallback<String, String>() {
@ -540,16 +537,16 @@ public class CommandAsyncService implements CommandAsyncExecutor {
return getConnectionManager().getCfg().isUseScriptCache();
}
private static final Map<String, String> shaCache = new LRUCacheMap<String, String>(500, 0, 0);
private static final Map<String, String> SHA_CACHE = new LRUCacheMap<String, String>(500, 0, 0);
private String calcSHA(String script) {
String digest = shaCache.get(script);
String digest = SHA_CACHE.get(script);
if (digest == null) {
try {
MessageDigest mdigest = MessageDigest.getInstance("SHA-1");
byte[] s = mdigest.digest(script.getBytes());
digest = ByteBufUtil.hexDump(s);
shaCache.put(script, digest);
SHA_CACHE.put(script, digest);
} catch (Exception e) {
throw new IllegalStateException(e);
}
@ -561,7 +558,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
List<Object> result = new ArrayList<Object>();
for (Object object : params) {
if (object instanceof ByteBuf) {
ByteBuf b = ((ByteBuf) object);
ByteBuf b = (ByteBuf) object;
ByteBuf nb = ByteBufAllocator.DEFAULT.buffer(b.readableBytes());
int ri = b.readerIndex();
nb.writeBytes(b);
@ -651,6 +648,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
return mainPromise;
}
@SuppressWarnings({"NestedIfDepth"})
public <V, R> void async(final boolean readOnlyMode, final NodeSource source, final Codec codec,
final RedisCommand<V> command, final Object[] params, final RPromise<R> mainPromise, final int attempt,
final boolean ignoreRedirect) {
@ -802,28 +800,29 @@ public class CommandAsyncService implements CommandAsyncExecutor {
});
}
private static final Map<ClassLoader, Map<Codec, Codec>> codecs = ReferenceCacheMap.weak(0, 0);
private static final Map<ClassLoader, Map<Codec, Codec>> CODECS = ReferenceCacheMap.weak(0, 0);
protected Codec getCodec(Codec codec) {
if (codec instanceof StringCodec
|| codec instanceof ByteArrayCodec
|| codec instanceof LocalCachedMessageCodec
|| codec instanceof BitSetCodec
|| codec instanceof JCacheEventCodec
|| codec == null) {
if (codec == null) {
return codec;
}
for (Class<?> clazz : BaseCodec.SKIPPED_CODECS) {
if (clazz.isAssignableFrom(codec.getClass())) {
return codec;
}
}
Codec codecToUse = codec;
ClassLoader threadClassLoader = Thread.currentThread().getContextClassLoader();
if (threadClassLoader != null) {
Map<Codec, Codec> map = codecs.get(threadClassLoader);
Map<Codec, Codec> map = CODECS.get(threadClassLoader);
if (map == null) {
synchronized (codecs) {
map = codecs.get(threadClassLoader);
synchronized (CODECS) {
map = CODECS.get(threadClassLoader);
if (map == null) {
map = new ConcurrentHashMap<Codec, Codec>();
codecs.put(threadClassLoader, map);
CODECS.put(threadClassLoader, map);
}
}
}
@ -1079,9 +1078,9 @@ public class CommandAsyncService implements CommandAsyncExecutor {
}
AsyncDetails.release(details);
} catch (RuntimeException e) {
} catch (Exception e) {
handleError(details, details.getMainPromise(), e);
throw e;
throw e;
}
}
@ -1197,7 +1196,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
if (o instanceof RedissonReference) {
return fromReference(o);
} else if (o instanceof ScoredEntry && ((ScoredEntry) o).getValue() instanceof RedissonReference) {
ScoredEntry<?> se = ((ScoredEntry<?>) o);
ScoredEntry<?> se = (ScoredEntry<?>) o;
return new ScoredEntry(se.getScore(), fromReference(se.getValue()));
} else if (o instanceof Map.Entry) {
Map.Entry old = (Map.Entry) o;

55
redisson/src/main/java/org/redisson/command/CommandBatchService.java
Unescape Escape View File

@ -98,7 +98,7 @@ public class CommandBatchService extends CommandAsyncService {
public static class Entry {
Deque<BatchCommandData<?, ?>> commands = new LinkedBlockingDeque<BatchCommandData<?,?>>();
Deque<BatchCommandData<?, ?>> commands = new LinkedBlockingDeque<>();
volatile boolean readOnlyMode = true;
public Deque<BatchCommandData<?, ?>> getCommands() {
@ -122,7 +122,7 @@ public class CommandBatchService extends CommandAsyncService {
}
private final AtomicInteger index = new AtomicInteger();
private AtomicInteger index = new AtomicInteger();
private ConcurrentMap<MasterSlaveEntry, Entry> commands = PlatformDependent.newConcurrentHashMap();
private ConcurrentMap<MasterSlaveEntry, ConnectionEntry> connections = PlatformDependent.newConcurrentHashMap();
@ -206,7 +206,7 @@ public class CommandBatchService extends CommandAsyncService {
}
@Override
protected <V, R> void handleSuccess(final AsyncDetails<V, R> details, RPromise<R> promise, RedisCommand<?> command, R res) {
protected <V, R> void handleSuccess(AsyncDetails<V, R> details, RPromise<R> promise, RedisCommand<?> command, R res) {
if (RedisCommands.EXEC.getName().equals(command.getName())) {
super.handleSuccess(details, promise, command, res);
return;
@ -230,7 +230,7 @@ public class CommandBatchService extends CommandAsyncService {
}
@Override
protected <V, R> void handleError(final AsyncDetails<V, R> details, RPromise<R> promise, Throwable cause) {
protected <V, R> void handleError(AsyncDetails<V, R> details, RPromise<R> promise, Throwable cause) {
if (isRedisBasedQueue() && promise instanceof BatchPromise) {
BatchPromise<R> batchPromise = (BatchPromise<R>) promise;
RPromise<R> sentPromise = (RPromise<R>) batchPromise.getSentPromise();
@ -368,7 +368,7 @@ public class CommandBatchService extends CommandAsyncService {
}
public RFuture<Void> executeAsyncVoid() {
final RedissonPromise<Void> promise = new RedissonPromise<Void>();
RedissonPromise<Void> promise = new RedissonPromise<Void>();
RFuture<BatchResult<?>> resFuture = executeAsync(BatchOptions.defaults());
resFuture.onComplete((res, e) -> {
if (e == null) {
@ -384,6 +384,7 @@ public class CommandBatchService extends CommandAsyncService {
return executeAsync(BatchOptions.defaults());
}
@SuppressWarnings("MethodLength")
public <R> RFuture<R> executeAsync(BatchOptions options) {
if (executed.get()) {
throw new IllegalStateException("Batch already executed!");
@ -403,7 +404,7 @@ public class CommandBatchService extends CommandAsyncService {
int permits = 0;
for (Entry entry : commands.values()) {
permits += entry.getCommands().size();
};
}
RPromise<R> resultPromise = new RedissonPromise<R>();
semaphore.acquire(new Runnable() {
@ -491,7 +492,7 @@ public class CommandBatchService extends CommandAsyncService {
}
}
BatchResult<Object> r = new BatchResult<Object>(responses, syncedSlaves);
resultPromise.trySuccess((R)r);
resultPromise.trySuccess((R) r);
} catch (Exception e) {
resultPromise.tryFailure(e);
}
@ -530,7 +531,7 @@ public class CommandBatchService extends CommandAsyncService {
}
RPromise<R> resultPromise;
final RPromise<Void> voidPromise = new RedissonPromise<Void>();
RPromise<Void> voidPromise = new RedissonPromise<Void>();
if (this.options.isSkipResult()) {
voidPromise.onComplete((res, e) -> {
// commands = null;
@ -539,7 +540,7 @@ public class CommandBatchService extends CommandAsyncService {
});
resultPromise = (RPromise<R>) voidPromise;
} else {
final RPromise<Object> promise = new RedissonPromise<Object>();
RPromise<Object> promise = new RedissonPromise<Object>();
voidPromise.onComplete((res, ex) -> {
executed.set(true);
if (ex != null) {
@ -576,7 +577,7 @@ public class CommandBatchService extends CommandAsyncService {
resultPromise = (RPromise<R>) promise;
}
final AtomicInteger slots = new AtomicInteger(commands.size());
AtomicInteger slots = new AtomicInteger(commands.size());
for (java.util.Map.Entry<RFuture<?>, List<CommandBatchService>> entry : nestedServices.entrySet()) {
slots.incrementAndGet();
@ -599,8 +600,9 @@ public class CommandBatchService extends CommandAsyncService {
return options != null && (this.options.getExecutionMode() == ExecutionMode.REDIS_READ_ATOMIC || this.options.getExecutionMode() == ExecutionMode.REDIS_WRITE_ATOMIC);
}
private void execute(final Entry entry, final NodeSource source, final RPromise<Void> mainPromise, final AtomicInteger slots,
final int attempt, final BatchOptions options) {
@SuppressWarnings({"MethodLength", "NestedIfDepth"})
private void execute(Entry entry, NodeSource source, RPromise<Void> mainPromise, AtomicInteger slots,
int attempt, BatchOptions options) {
if (mainPromise.isCancelled()) {
free(entry);
return;
@ -612,9 +614,9 @@ public class CommandBatchService extends CommandAsyncService {
return;
}
final RPromise<Void> attemptPromise = new RedissonPromise<Void>();
RPromise<Void> attemptPromise = new RedissonPromise<Void>();
final AsyncDetails details = new AsyncDetails();
AsyncDetails details = new AsyncDetails();
details.init(null, attemptPromise,
entry.isReadOnlyMode(), source, null, null, null, mainPromise, attempt);
@ -654,7 +656,7 @@ public class CommandBatchService extends CommandAsyncService {
}
};
final TimerTask retryTimerTask = new TimerTask() {
TimerTask retryTimerTask = new TimerTask() {
@Override
public void run(Timeout t) throws Exception {
if (attemptPromise.isDone()) {
@ -732,14 +734,14 @@ public class CommandBatchService extends CommandAsyncService {
skip.set(true);
if (e instanceof RedisMovedException) {
RedisMovedException ex = (RedisMovedException)e;
RedisMovedException ex = (RedisMovedException) e;
entry.clearErrors();
NodeSource nodeSource = new NodeSource(ex.getSlot(), ex.getUrl(), Redirect.MOVED);
execute(entry, nodeSource, mainPromise, slots, attempt, options);
return;
}
if (e instanceof RedisAskException) {
RedisAskException ex = (RedisAskException)e;
RedisAskException ex = (RedisAskException) e;
entry.clearErrors();
NodeSource nodeSource = new NodeSource(ex.getSlot(), ex.getUrl(), Redirect.ASK);
execute(entry, nodeSource, mainPromise, slots, attempt, options);
@ -765,14 +767,14 @@ public class CommandBatchService extends CommandAsyncService {
});
}
protected void free(final Entry entry) {
protected void free(Entry entry) {
for (BatchCommandData<?, ?> command : entry.getCommands()) {
free(command.getParams());
}
}
private void checkWriteFuture(final Entry entry, final RPromise<Void> attemptPromise, final AsyncDetails details,
final RedisConnection connection, ChannelFuture future, long responseTimeout, int attempts, final AtomicInteger slots, final RPromise<Void> mainPromise) {
private void checkWriteFuture(Entry entry, RPromise<Void> attemptPromise, AsyncDetails details,
RedisConnection connection, ChannelFuture future, long responseTimeout, int attempts, AtomicInteger slots, RPromise<Void> mainPromise) {
if (future.isCancelled() || attemptPromise.isDone()) {
return;
}
@ -820,10 +822,11 @@ public class CommandBatchService extends CommandAsyncService {
details.setTimeout(timeoutTask);
}
private void checkConnectionFuture(final Entry entry, final NodeSource source,
final RPromise<Void> mainPromise, final RPromise<Void> attemptPromise, final AsyncDetails details,
RFuture<RedisConnection> connFuture, final boolean noResult, final long responseTimeout, final int attempts,
ExecutionMode executionMode, final AtomicInteger slots) {
@SuppressWarnings("ParameterNumber")
private void checkConnectionFuture(Entry entry, NodeSource source,
RPromise<Void> mainPromise, RPromise<Void> attemptPromise, AsyncDetails details,
RFuture<RedisConnection> connFuture, boolean noResult, long responseTimeout, int attempts,
ExecutionMode executionMode, AtomicInteger slots) {
if (connFuture.isCancelled()) {
connectionManager.getShutdownLatch().release();
return;
@ -840,7 +843,7 @@ public class CommandBatchService extends CommandAsyncService {
return;
}
final RedisConnection connection = connFuture.getNow();
RedisConnection connection = connFuture.getNow();
boolean isAtomic = executionMode != ExecutionMode.IN_MEMORY;
boolean isQueued = executionMode == ExecutionMode.REDIS_READ_ATOMIC || executionMode == ExecutionMode.REDIS_WRITE_ATOMIC;
@ -874,7 +877,7 @@ public class CommandBatchService extends CommandAsyncService {
return c.getCommand().getName().equals(RedisCommands.WAIT.getName());
}
protected void handle(final RPromise<Void> mainPromise, final AtomicInteger slots, RFuture<?> future) {
protected void handle(RPromise<Void> mainPromise, AtomicInteger slots, RFuture<?> future) {
if (future.isSuccess()) {
if (slots.decrementAndGet() == 0) {
mainPromise.trySuccess(null);

Write Preview
Loading…
Cancel
Save