diff --git a/redisson/pom.xml b/redisson/pom.xml index bdf10482f..bbcba5669 100644 --- a/redisson/pom.xml +++ b/redisson/pom.xml @@ -355,7 +355,7 @@ verify - checkstyle + check @@ -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 true false diff --git a/redisson/src/main/java/org/redisson/client/codec/BaseCodec.java b/redisson/src/main/java/org/redisson/client/codec/BaseCodec.java index 18c261357..e89f1929e 100644 --- a/redisson/src/main/java/org/redisson/client/codec/BaseCodec.java +++ b/redisson/src/main/java/org/redisson/client/codec/BaseCodec.java @@ -30,7 +30,7 @@ import org.redisson.jcache.JCacheEventCodec; */ public abstract class BaseCodec implements Codec { - private static final List> SKIPPED_CODECS = Arrays.asList(StringCodec.class, + public static final List> SKIPPED_CODECS = Arrays.asList(StringCodec.class, ByteArrayCodec.class, LocalCachedMessageCodec.class, BitSetCodec.class, JCacheEventCodec.class); public static Codec copy(ClassLoader classLoader, Codec codec) { diff --git a/redisson/src/main/java/org/redisson/command/AsyncDetails.java b/redisson/src/main/java/org/redisson/command/AsyncDetails.java index f414fd978..e8a4b1a44 100644 --- a/redisson/src/main/java/org/redisson/command/AsyncDetails.java +++ b/redisson/src/main/java/org/redisson/command/AsyncDetails.java @@ -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 { - static final ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue(); - RFuture connectionFuture; ConnectionManager connectionManager; RPromise attemptPromise; diff --git a/redisson/src/main/java/org/redisson/command/CommandAsyncExecutor.java b/redisson/src/main/java/org/redisson/command/CommandAsyncExecutor.java index fe25ded99..b0604e8b1 100644 --- a/redisson/src/main/java/org/redisson/command/CommandAsyncExecutor.java +++ b/redisson/src/main/java/org/redisson/command/CommandAsyncExecutor.java @@ -51,13 +51,13 @@ public interface CommandAsyncExecutor { boolean isRedissonReferenceSupportEnabled(); - RedisException convertException(RFuture RFuture); + RedisException convertException(RFuture 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 get(RFuture RFuture); + V get(RFuture future); RFuture writeAsync(MasterSlaveEntry entry, Codec codec, RedisCommand command, Object... params); diff --git a/redisson/src/main/java/org/redisson/command/CommandAsyncService.java b/redisson/src/main/java/org/redisson/command/CommandAsyncService.java index bee00f51c..6090625fa 100644 --- a/redisson/src/main/java/org/redisson/command/CommandAsyncService.java +++ b/redisson/src/main/java/org/redisson/command/CommandAsyncService.java @@ -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() { @@ -540,16 +537,16 @@ public class CommandAsyncService implements CommandAsyncExecutor { return getConnectionManager().getCfg().isUseScriptCache(); } - private static final Map shaCache = new LRUCacheMap(500, 0, 0); + private static final Map SHA_CACHE = new LRUCacheMap(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 result = new ArrayList(); 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 void async(final boolean readOnlyMode, final NodeSource source, final Codec codec, final RedisCommand command, final Object[] params, final RPromise mainPromise, final int attempt, final boolean ignoreRedirect) { @@ -802,28 +800,29 @@ public class CommandAsyncService implements CommandAsyncExecutor { }); } - private static final Map> codecs = ReferenceCacheMap.weak(0, 0); + private static final Map> 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 map = codecs.get(threadClassLoader); + Map 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(); - 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; diff --git a/redisson/src/main/java/org/redisson/command/CommandBatchService.java b/redisson/src/main/java/org/redisson/command/CommandBatchService.java index 08cb3c94c..b375180ac 100644 --- a/redisson/src/main/java/org/redisson/command/CommandBatchService.java +++ b/redisson/src/main/java/org/redisson/command/CommandBatchService.java @@ -98,7 +98,7 @@ public class CommandBatchService extends CommandAsyncService { public static class Entry { - Deque> commands = new LinkedBlockingDeque>(); + Deque> commands = new LinkedBlockingDeque<>(); volatile boolean readOnlyMode = true; public Deque> getCommands() { @@ -122,7 +122,7 @@ public class CommandBatchService extends CommandAsyncService { } - private final AtomicInteger index = new AtomicInteger(); + private AtomicInteger index = new AtomicInteger(); private ConcurrentMap commands = PlatformDependent.newConcurrentHashMap(); private ConcurrentMap connections = PlatformDependent.newConcurrentHashMap(); @@ -206,7 +206,7 @@ public class CommandBatchService extends CommandAsyncService { } @Override - protected void handleSuccess(final AsyncDetails details, RPromise promise, RedisCommand command, R res) { + protected void handleSuccess(AsyncDetails details, RPromise 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 void handleError(final AsyncDetails details, RPromise promise, Throwable cause) { + protected void handleError(AsyncDetails details, RPromise promise, Throwable cause) { if (isRedisBasedQueue() && promise instanceof BatchPromise) { BatchPromise batchPromise = (BatchPromise) promise; RPromise sentPromise = (RPromise) batchPromise.getSentPromise(); @@ -368,7 +368,7 @@ public class CommandBatchService extends CommandAsyncService { } public RFuture executeAsyncVoid() { - final RedissonPromise promise = new RedissonPromise(); + RedissonPromise promise = new RedissonPromise(); RFuture> 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 RFuture 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 resultPromise = new RedissonPromise(); semaphore.acquire(new Runnable() { @@ -491,7 +492,7 @@ public class CommandBatchService extends CommandAsyncService { } } BatchResult r = new BatchResult(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 resultPromise; - final RPromise voidPromise = new RedissonPromise(); + RPromise voidPromise = new RedissonPromise(); if (this.options.isSkipResult()) { voidPromise.onComplete((res, e) -> { // commands = null; @@ -539,7 +540,7 @@ public class CommandBatchService extends CommandAsyncService { }); resultPromise = (RPromise) voidPromise; } else { - final RPromise promise = new RedissonPromise(); + RPromise promise = new RedissonPromise(); voidPromise.onComplete((res, ex) -> { executed.set(true); if (ex != null) { @@ -576,7 +577,7 @@ public class CommandBatchService extends CommandAsyncService { resultPromise = (RPromise) promise; } - final AtomicInteger slots = new AtomicInteger(commands.size()); + AtomicInteger slots = new AtomicInteger(commands.size()); for (java.util.Map.Entry, List> 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 mainPromise, final AtomicInteger slots, - final int attempt, final BatchOptions options) { + @SuppressWarnings({"MethodLength", "NestedIfDepth"}) + private void execute(Entry entry, NodeSource source, RPromise 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 attemptPromise = new RedissonPromise(); + RPromise attemptPromise = new RedissonPromise(); - 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 attemptPromise, final AsyncDetails details, - final RedisConnection connection, ChannelFuture future, long responseTimeout, int attempts, final AtomicInteger slots, final RPromise mainPromise) { + private void checkWriteFuture(Entry entry, RPromise attemptPromise, AsyncDetails details, + RedisConnection connection, ChannelFuture future, long responseTimeout, int attempts, AtomicInteger slots, RPromise 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 mainPromise, final RPromise attemptPromise, final AsyncDetails details, - RFuture 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 mainPromise, RPromise attemptPromise, AsyncDetails details, + RFuture 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 mainPromise, final AtomicInteger slots, RFuture future) { + protected void handle(RPromise mainPromise, AtomicInteger slots, RFuture future) { if (future.isSuccess()) { if (slots.decrementAndGet() == 0) { mainPromise.trySuccess(null);