/** * Copyright 2018 Nikita Koksharov * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package com.fr.third.org.redisson; import java.io.IOException; import java.io.Serializable; import java.math.BigDecimal; import java.util.AbstractMap; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; import com.fr.third.org.redisson.api.LocalCachedMapOptions; import com.fr.third.org.redisson.api.RFuture; import com.fr.third.org.redisson.api.RLocalCachedMap; import com.fr.third.org.redisson.api.RScoredSortedSet; import com.fr.third.org.redisson.api.RTopic; import com.fr.third.org.redisson.api.RedissonClient; import com.fr.third.org.redisson.api.listener.BaseStatusListener; import com.fr.third.org.redisson.api.listener.MessageListener; import com.fr.third.org.redisson.cache.Cache; import com.fr.third.org.redisson.cache.LFUCacheMap; import com.fr.third.org.redisson.cache.LRUCacheMap; import com.fr.third.org.redisson.cache.LocalCachedMapClear; import com.fr.third.org.redisson.cache.LocalCachedMapInvalidate; import com.fr.third.org.redisson.cache.LocalCachedMapUpdate; import com.fr.third.org.redisson.cache.LocalCachedMessageCodec; import com.fr.third.org.redisson.cache.NoneCacheMap; import com.fr.third.org.redisson.cache.ReferenceCacheMap; import com.fr.third.org.redisson.client.codec.ByteArrayCodec; import com.fr.third.org.redisson.client.codec.Codec; import com.fr.third.org.redisson.client.codec.LongCodec; import com.fr.third.org.redisson.client.codec.StringCodec; import com.fr.third.org.redisson.client.protocol.RedisCommand; import com.fr.third.org.redisson.client.protocol.RedisCommands; import com.fr.third.org.redisson.client.protocol.convertor.NumberConvertor; import com.fr.third.org.redisson.client.protocol.decoder.ObjectMapEntryReplayDecoder; import com.fr.third.org.redisson.client.protocol.decoder.ObjectMapReplayDecoder; import com.fr.third.org.redisson.client.protocol.decoder.ObjectSetReplayDecoder; import com.fr.third.org.redisson.misc.Hash; import com.fr.third.org.redisson.misc.RPromise; import com.fr.third.org.redisson.misc.RedissonPromise; import com.fr.third.org.redisson.api.LocalCachedMapOptions; import com.fr.third.org.redisson.api.LocalCachedMapOptions.EvictionPolicy; import com.fr.third.org.redisson.api.LocalCachedMapOptions.ReconnectionStrategy; import com.fr.third.org.redisson.api.LocalCachedMapOptions.SyncStrategy; import com.fr.third.org.redisson.api.RFuture; import com.fr.third.org.redisson.api.RLocalCachedMap; import com.fr.third.org.redisson.api.RScoredSortedSet; import com.fr.third.org.redisson.api.RTopic; import com.fr.third.org.redisson.api.RedissonClient; import com.fr.third.org.redisson.api.listener.BaseStatusListener; import com.fr.third.org.redisson.api.listener.MessageListener; import com.fr.third.org.redisson.cache.Cache; import com.fr.third.org.redisson.cache.LFUCacheMap; import com.fr.third.org.redisson.cache.LRUCacheMap; import com.fr.third.org.redisson.cache.LocalCachedMapClear; import com.fr.third.org.redisson.cache.LocalCachedMapInvalidate; import com.fr.third.org.redisson.cache.LocalCachedMapUpdate; import com.fr.third.org.redisson.cache.LocalCachedMessageCodec; import com.fr.third.org.redisson.cache.NoneCacheMap; import com.fr.third.org.redisson.cache.ReferenceCacheMap; import com.fr.third.org.redisson.client.codec.ByteArrayCodec; import com.fr.third.org.redisson.client.codec.Codec; import com.fr.third.org.redisson.client.codec.LongCodec; import com.fr.third.org.redisson.client.codec.StringCodec; import com.fr.third.org.redisson.client.protocol.RedisCommand; import com.fr.third.org.redisson.client.protocol.RedisCommand.ValueType; import com.fr.third.org.redisson.client.protocol.RedisCommands; import com.fr.third.org.redisson.client.protocol.convertor.NumberConvertor; import com.fr.third.org.redisson.client.protocol.decoder.ObjectMapEntryReplayDecoder; import com.fr.third.org.redisson.client.protocol.decoder.ObjectMapReplayDecoder; import com.fr.third.org.redisson.client.protocol.decoder.ObjectSetReplayDecoder; import com.fr.third.org.redisson.command.CommandAsyncExecutor; import com.fr.third.org.redisson.eviction.EvictionScheduler; import com.fr.third.org.redisson.misc.Hash; import com.fr.third.org.redisson.misc.RPromise; import com.fr.third.org.redisson.misc.RedissonPromise; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.FutureListener; import io.netty.util.internal.PlatformDependent; /** * * @author Nikita Koksharov * */ @SuppressWarnings("serial") public class RedissonLocalCachedMap extends RedissonMap implements RLocalCachedMap { private static final Logger log = LoggerFactory.getLogger(RedissonLocalCachedMap.class); public static class CacheKey implements Serializable { private final byte[] keyHash; public CacheKey(byte[] keyHash) { super(); this.keyHash = keyHash; } public byte[] getKeyHash() { return keyHash; } @Override public int hashCode() { final int prime = 31; int result = 1; result = prime * result + Arrays.hashCode(keyHash); return result; } @Override public boolean equals(Object obj) { if (this == obj) return true; if (obj == null) return false; if (getClass() != obj.getClass()) return false; CacheKey other = (CacheKey) obj; if (!Arrays.equals(keyHash, other.keyHash)) return false; return true; } @Override public String toString() { return "CacheKey [keyHash=" + Arrays.toString(keyHash) + "]"; } } public static class CacheValue implements Serializable { private final Object key; private final Object value; public CacheValue(Object key, Object value) { super(); this.key = key; this.value = value; } public Object getKey() { return key; } public Object getValue() { return value; } @Override public boolean equals(Object obj) { if (this == obj) return true; if (obj == null) return false; if (getClass() != obj.getClass()) return false; CacheValue other = (CacheValue) obj; if (value == null) { if (other.value != null) return false; } else if (!value.equals(other.value)) return false; return true; } @Override public String toString() { return "CacheValue [key=" + key + ", value=" + value + "]"; } } private static final RedisCommand> ALL_KEYS = new RedisCommand>("EVAL", new ObjectSetReplayDecoder(), RedisCommand.ValueType.MAP_KEY); private static final RedisCommand>> ALL_ENTRIES = new RedisCommand>>("EVAL", new ObjectMapEntryReplayDecoder(), RedisCommand.ValueType.MAP); private static final RedisCommand> ALL_MAP = new RedisCommand>("EVAL", new ObjectMapReplayDecoder(), RedisCommand.ValueType.MAP); private long cacheUpdateLogTime = TimeUnit.MINUTES.toMillis(10); private byte[] instanceId; private RTopic invalidationTopic; private Cache cache; private int invalidateEntryOnChange; private int syncListenerId; private int reconnectionListenerId; private volatile long lastInvalidate; private LocalCachedMapOptions.SyncStrategy syncStrategy; private final Codec topicCodec = new LocalCachedMessageCodec(); protected RedissonLocalCachedMap(CommandAsyncExecutor commandExecutor, String name, LocalCachedMapOptions options, EvictionScheduler evictionScheduler, RedissonClient redisson) { super(commandExecutor, name, redisson, options); init(name, options, redisson, evictionScheduler); } protected RedissonLocalCachedMap(Codec codec, CommandAsyncExecutor connectionManager, String name, LocalCachedMapOptions options, EvictionScheduler evictionScheduler, RedissonClient redisson) { super(codec, connectionManager, name, redisson, options); init(name, options, redisson, evictionScheduler); } private void init(String name, LocalCachedMapOptions options, RedissonClient redisson, EvictionScheduler evictionScheduler) { instanceId = generateId(); syncStrategy = options.getSyncStrategy(); if (options.getSyncStrategy() != LocalCachedMapOptions.SyncStrategy.NONE) { invalidateEntryOnChange = 1; } if (options.getReconnectionStrategy() == LocalCachedMapOptions.ReconnectionStrategy.LOAD) { invalidateEntryOnChange = 2; evictionScheduler.schedule(getUpdatesLogName(), cacheUpdateLogTime + TimeUnit.MINUTES.toMillis(1)); } cache = createCache(options); addListeners(name, options, redisson); } private void addListeners(String name, final LocalCachedMapOptions options, final RedissonClient redisson) { invalidationTopic = new RedissonTopic(topicCodec, commandExecutor, suffixName(name, "topic")); if (options.getReconnectionStrategy() != LocalCachedMapOptions.ReconnectionStrategy.NONE) { reconnectionListenerId = invalidationTopic.addListener(new BaseStatusListener() { @Override public void onSubscribe(String channel) { if (options.getReconnectionStrategy() == LocalCachedMapOptions.ReconnectionStrategy.CLEAR) { cache.clear(); } if (options.getReconnectionStrategy() == LocalCachedMapOptions.ReconnectionStrategy.LOAD // check if instance has already been used && lastInvalidate > 0) { if (System.currentTimeMillis() - lastInvalidate > cacheUpdateLogTime) { cache.clear(); return; } isExistsAsync().addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { if (!future.isSuccess()) { log.error("Can't check existance", future.cause()); return; } if (!future.getNow()) { cache.clear(); return; } RScoredSortedSet logs = redisson.getScoredSortedSet(getUpdatesLogName(), ByteArrayCodec.INSTANCE); logs.valueRangeAsync(lastInvalidate, true, Double.POSITIVE_INFINITY, true) .addListener(new FutureListener>() { @Override public void operationComplete(Future> future) throws Exception { if (!future.isSuccess()) { log.error("Can't load update log", future.cause()); return; } for (byte[] entry : future.getNow()) { byte[] keyHash = Arrays.copyOf(entry, 16); CacheKey key = new CacheKey(keyHash); cache.remove(key); } } }); } }); } } }); } if (options.getSyncStrategy() != LocalCachedMapOptions.SyncStrategy.NONE) { syncListenerId = invalidationTopic.addListener(new MessageListener() { @Override public void onMessage(String channel, Object msg) { if (msg instanceof LocalCachedMapClear) { cache.clear(); } if (msg instanceof LocalCachedMapInvalidate) { LocalCachedMapInvalidate invalidateMsg = (LocalCachedMapInvalidate)msg; if (!Arrays.equals(invalidateMsg.getExcludedId(), instanceId)) { for (byte[] keyHash : invalidateMsg.getKeyHashes()) { CacheKey key = new CacheKey(keyHash); cache.remove(key); } } } if (msg instanceof LocalCachedMapUpdate) { LocalCachedMapUpdate updateMsg = (LocalCachedMapUpdate) msg; for (LocalCachedMapUpdate.Entry entry : updateMsg.getEntries()) { ByteBuf keyBuf = Unpooled.wrappedBuffer(entry.getKey()); ByteBuf valueBuf = Unpooled.wrappedBuffer(entry.getValue()); try { CacheKey cacheKey = toCacheKey(keyBuf); Object key = codec.getMapKeyDecoder().decode(keyBuf, null); Object value = codec.getMapValueDecoder().decode(valueBuf, null); cache.put(cacheKey, new CacheValue(key, value)); } catch (IOException e) { log.error("Can't decode map entry", e); } finally { keyBuf.release(); valueBuf.release(); } } } if (options.getReconnectionStrategy() == LocalCachedMapOptions.ReconnectionStrategy.LOAD) { lastInvalidate = System.currentTimeMillis(); } } }); } } protected Cache createCache(LocalCachedMapOptions options) { if (options.getEvictionPolicy() == LocalCachedMapOptions.EvictionPolicy.NONE) { return new NoneCacheMap(options.getTimeToLiveInMillis(), options.getMaxIdleInMillis()); } if (options.getEvictionPolicy() == LocalCachedMapOptions.EvictionPolicy.LRU) { return new LRUCacheMap(options.getCacheSize(), options.getTimeToLiveInMillis(), options.getMaxIdleInMillis()); } if (options.getEvictionPolicy() == LocalCachedMapOptions.EvictionPolicy.LFU) { return new LFUCacheMap(options.getCacheSize(), options.getTimeToLiveInMillis(), options.getMaxIdleInMillis()); } if (options.getEvictionPolicy() == LocalCachedMapOptions.EvictionPolicy.SOFT) { return ReferenceCacheMap.soft(options.getTimeToLiveInMillis(), options.getMaxIdleInMillis()); } if (options.getEvictionPolicy() == LocalCachedMapOptions.EvictionPolicy.WEAK) { return ReferenceCacheMap.weak(options.getTimeToLiveInMillis(), options.getMaxIdleInMillis()); } throw new IllegalArgumentException("Invalid eviction policy: " + options.getEvictionPolicy()); } private CacheKey toCacheKey(Object key) { ByteBuf encoded = encodeMapKey(key); try { return toCacheKey(encoded); } finally { encoded.release(); } } private CacheKey toCacheKey(ByteBuf encodedKey) { return new CacheKey(Hash.hash128toArray(encodedKey)); } @Override public RFuture containsKeyAsync(Object key) { checkKey(key); CacheKey cacheKey = toCacheKey(key); if (!cache.containsKey(cacheKey)) { return super.containsKeyAsync(key); } return RedissonPromise.newSucceededFuture(true); } @Override public RFuture containsValueAsync(Object value) { checkValue(value); CacheValue cacheValue = new CacheValue(null, value); if (!cache.containsValue(cacheValue)) { return super.containsValueAsync(value); } return RedissonPromise.newSucceededFuture(true); } @Override public RFuture getAsync(final Object key) { checkKey(key); final CacheKey cacheKey = toCacheKey(key); CacheValue cacheValue = cache.get(cacheKey); if (cacheValue != null && cacheValue.getValue() != null) { return RedissonPromise.newSucceededFuture((V)cacheValue.getValue()); } RFuture future = super.getAsync((K)key); future.addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { if (!future.isSuccess()) { return; } V value = future.getNow(); if (value != null) { cache.put(cacheKey, new CacheValue(key, value)); } } }); return future; } String getUpdatesLogName() { return prefixName("redisson__cache_updates_log", getName()); } protected static byte[] generateId() { byte[] id = new byte[16]; // TODO JDK UPGRADE replace to native ThreadLocalRandom PlatformDependent.threadLocalRandom().nextBytes(id); return id; } protected static byte[] generateLogEntryId(byte[] keyHash) { byte[] result = new byte[keyHash.length + 1 + 8]; result[16] = ':'; byte[] id = new byte[8]; // TODO JDK UPGRADE replace to native ThreadLocalRandom PlatformDependent.threadLocalRandom().nextBytes(id); System.arraycopy(keyHash, 0, result, 0, keyHash.length); System.arraycopy(id, 0, result, 17, id.length); return result; } @Override protected RFuture putOperationAsync(K key, V value) { ByteBuf mapKey = encodeMapKey(key); ByteBuf mapValue = encodeMapValue(value); CacheKey cacheKey = toCacheKey(mapKey); byte[] entryId = generateLogEntryId(cacheKey.getKeyHash()); ByteBuf msg = createSyncMessage(mapKey, mapValue, cacheKey); CacheValue cacheValue = new CacheValue(key, value); cache.put(cacheKey, cacheValue); return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_MAP_VALUE, "local v = redis.call('hget', KEYS[1], ARGV[1]); " + "redis.call('hset', KEYS[1], ARGV[1], ARGV[2]); " + "if ARGV[4] == '1' then " + "redis.call('publish', KEYS[2], ARGV[3]); " + "end;" + "if ARGV[4] == '2' then " + "redis.call('zadd', KEYS[3], ARGV[5], ARGV[6]);" + "redis.call('publish', KEYS[2], ARGV[3]); " + "end;" + "return v; ", Arrays.asList(getName(), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), mapKey, mapValue, msg, invalidateEntryOnChange, System.currentTimeMillis(), entryId); } protected ByteBuf createSyncMessage(ByteBuf mapKey, ByteBuf mapValue, CacheKey cacheKey) { if (syncStrategy == LocalCachedMapOptions.SyncStrategy.UPDATE) { return encode(new LocalCachedMapUpdate(mapKey, mapValue)); } return encode(new LocalCachedMapInvalidate(instanceId, cacheKey.getKeyHash())); } @Override protected RFuture fastPutOperationAsync(K key, V value) { ByteBuf encodedKey = encodeMapKey(key); ByteBuf encodedValue = encodeMapValue(value); CacheKey cacheKey = toCacheKey(encodedKey); byte[] entryId = generateLogEntryId(cacheKey.getKeyHash()); ByteBuf msg = createSyncMessage(encodedKey, encodedValue, cacheKey); CacheValue cacheValue = new CacheValue(key, value); cache.put(cacheKey, cacheValue); return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_BOOLEAN, "if ARGV[4] == '1' then " + "redis.call('publish', KEYS[2], ARGV[3]); " + "end;" + "if ARGV[4] == '2' then " + "redis.call('zadd', KEYS[3], ARGV[5], ARGV[6]);" + "redis.call('publish', KEYS[2], ARGV[3]); " + "end;" + "if redis.call('hset', KEYS[1], ARGV[1], ARGV[2]) == 0 then " + "return 0; " + "end; " + "return 1; ", Arrays.asList(getName(), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), encodedKey, encodedValue, msg, invalidateEntryOnChange, System.currentTimeMillis(), entryId); } @Override public void destroy() { if (syncListenerId != 0) { invalidationTopic.removeListener(syncListenerId); } if (reconnectionListenerId != 0) { invalidationTopic.removeListener(reconnectionListenerId); } } @Override public RFuture removeOperationAsync(K key) { ByteBuf keyEncoded = encodeMapKey(key); CacheKey cacheKey = toCacheKey(keyEncoded); byte[] entryId = generateLogEntryId(cacheKey.getKeyHash()); ByteBuf msgEncoded = encode(new LocalCachedMapInvalidate(instanceId, cacheKey.getKeyHash())); cache.remove(cacheKey); return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_MAP_VALUE, "local v = redis.call('hget', KEYS[1], ARGV[1]); " + "if redis.call('hdel', KEYS[1], ARGV[1]) == 1 then " + "if ARGV[3] == '1' then " + "redis.call('publish', KEYS[2], ARGV[2]); " + "end; " + "if ARGV[3] == '2' then " + "redis.call('zadd', KEYS[3], ARGV[4], ARGV[5]);" + "redis.call('publish', KEYS[2], ARGV[2]); " + "end;" + "end; " + "return v", Arrays.asList(getName(), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), keyEncoded, msgEncoded, invalidateEntryOnChange, System.currentTimeMillis(), entryId); } @Override protected RFuture> fastRemoveOperationBatchAsync(@SuppressWarnings("unchecked") K... keys) { if (invalidateEntryOnChange == 1) { List params = new ArrayList(keys.length*2); for (K k : keys) { ByteBuf keyEncoded = encodeMapKey(k); params.add(keyEncoded); CacheKey cacheKey = toCacheKey(keyEncoded); cache.remove(cacheKey); ByteBuf msgEncoded = encode(new LocalCachedMapInvalidate(instanceId, cacheKey.getKeyHash())); params.add(msgEncoded); } return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_LIST, "local result = {}; " + "for j = 1, #ARGV, 2 do " + "local val = redis.call('hdel', KEYS[1], ARGV[j]);" + "if val == 1 then " + "redis.call('publish', KEYS[2], ARGV[j+1]); " + "end;" + "table.insert(result, val);" + "end;" + "return result;", Arrays.asList(getName(), invalidationTopic.getChannelNames().get(0)), params.toArray()); } if (invalidateEntryOnChange == 2) { List params = new ArrayList(keys.length*3); params.add(System.currentTimeMillis()); for (K k : keys) { ByteBuf keyEncoded = encodeMapKey(k); params.add(keyEncoded); CacheKey cacheKey = toCacheKey(keyEncoded); cache.remove(cacheKey); ByteBuf msgEncoded = encode(new LocalCachedMapInvalidate(instanceId, cacheKey.getKeyHash())); params.add(msgEncoded); byte[] entryId = generateLogEntryId(cacheKey.getKeyHash()); params.add(entryId); } return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_LIST, "local result = {}; " + "for j = 2, #ARGV, 3 do " + "local val = redis.call('hdel', KEYS[1], ARGV[j]);" + "if val == 1 then " + "redis.call('zadd', KEYS[3], ARGV[1], ARGV[j+2]);" + "redis.call('publish', KEYS[2], ARGV[j+1]); " + "end;" + "table.insert(result, val);" + "end;" + "return result;", Arrays.asList(getName(), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), params.toArray()); } List params = new ArrayList(keys.length); for (K k : keys) { ByteBuf keyEncoded = encodeMapKey(k); params.add(keyEncoded); CacheKey cacheKey = toCacheKey(keyEncoded); cache.remove(cacheKey); } RFuture> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_LIST, "local result = {}; " + "for i = 1, #ARGV, 1 do " + "local val = redis.call('hdel', KEYS[1], ARGV[i]); " + "table.insert(result, val); " + "end;" + "return result;", Arrays.asList(getName()), params.toArray()); return future; } @Override protected RFuture fastRemoveOperationAsync(K ... keys) { if (invalidateEntryOnChange == 1) { List params = new ArrayList(keys.length*2); for (K k : keys) { ByteBuf keyEncoded = encodeMapKey(k); params.add(keyEncoded); CacheKey cacheKey = toCacheKey(keyEncoded); cache.remove(cacheKey); ByteBuf msgEncoded = encode(new LocalCachedMapInvalidate(instanceId, cacheKey.getKeyHash())); params.add(msgEncoded); } return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_LONG, "local counter = 0; " + "for j = 1, #ARGV, 2 do " + "if redis.call('hdel', KEYS[1], ARGV[j]) == 1 then " + "redis.call('publish', KEYS[2], ARGV[j+1]); " + "counter = counter + 1;" + "end;" + "end;" + "return counter;", Arrays.asList(getName(), invalidationTopic.getChannelNames().get(0)), params.toArray()); } if (invalidateEntryOnChange == 2) { List params = new ArrayList(keys.length*3); params.add(System.currentTimeMillis()); for (K k : keys) { ByteBuf keyEncoded = encodeMapKey(k); params.add(keyEncoded); CacheKey cacheKey = toCacheKey(keyEncoded); cache.remove(cacheKey); ByteBuf msgEncoded = encode(new LocalCachedMapInvalidate(instanceId, cacheKey.getKeyHash())); params.add(msgEncoded); byte[] entryId = generateLogEntryId(cacheKey.getKeyHash()); params.add(entryId); } return commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_LONG, "local counter = 0; " + "for j = 2, #ARGV, 3 do " + "if redis.call('hdel', KEYS[1], ARGV[j]) == 1 then " + "redis.call('zadd', KEYS[3], ARGV[1], ARGV[j+2]);" + "redis.call('publish', KEYS[2], ARGV[j+1]); " + "counter = counter + 1;" + "end;" + "end;" + "return counter;", Arrays.asList(getName(), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), params.toArray()); } List params = new ArrayList(keys.length + 1); params.add(getName()); for (K k : keys) { ByteBuf keyEncoded = encodeMapKey(k); params.add(keyEncoded); CacheKey cacheKey = toCacheKey(keyEncoded); cache.remove(cacheKey); } return commandExecutor.writeAsync(getName(), codec, RedisCommands.HDEL, params.toArray()); } @Override public RFuture deleteAsync() { cache.clear(); ByteBuf msgEncoded = encode(new LocalCachedMapClear()); return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if redis.call('del', KEYS[1], KEYS[3]) > 0 and ARGV[2] ~= '0' then " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1;" + "end; " + "return 0;", Arrays.asList(getName(), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), msgEncoded, invalidateEntryOnChange); } @Override public RFuture> getAllAsync(Set keys) { if (keys.isEmpty()) { return RedissonPromise.newSucceededFuture(Collections.emptyMap()); } final Map result = new HashMap(); Set mapKeys = new HashSet(keys); for (Iterator iterator = mapKeys.iterator(); iterator.hasNext();) { K key = iterator.next(); final CacheKey cacheKey = toCacheKey(key); CacheValue value = cache.get(cacheKey); if (value != null) { result.put(key, (V)value.getValue()); iterator.remove(); } } final RPromise> promise = new RedissonPromise>(); RFuture> future = super.getAllAsync(mapKeys); future.addListener(new FutureListener>() { @Override public void operationComplete(Future> future) throws Exception { if (!future.isSuccess()) { promise.tryFailure(future.cause()); return; } Map map = future.getNow(); result.putAll(map); cacheMap(map); promise.trySuccess(result); } }); return promise; } private void cacheMap(Map map) { for (java.util.Map.Entry entry : map.entrySet()) { CacheKey cacheKey = toCacheKey(entry.getKey()); CacheValue cacheValue = new CacheValue(entry.getKey(), entry.getValue()); cache.put(cacheKey, cacheValue); } } @Override protected RFuture putAllOperationAsync(final Map map) { List params = new ArrayList(map.size()*3); params.add(invalidateEntryOnChange); params.add(map.size()*2); byte[][] hashes = new byte[map.size()][]; int i = 0; for (java.util.Map.Entry t : map.entrySet()) { ByteBuf mapKey = encodeMapKey(t.getKey()); ByteBuf mapValue = encodeMapValue(t.getValue()); params.add(mapKey); params.add(mapValue); CacheKey cacheKey = toCacheKey(mapKey); hashes[i] = cacheKey.getKeyHash(); i++; } ByteBuf msgEncoded = null; if (syncStrategy == LocalCachedMapOptions.SyncStrategy.UPDATE) { List entries = new ArrayList(); for (int j = 2; j < params.size(); j += 2) { ByteBuf key = (ByteBuf) params.get(j); ByteBuf value = (ByteBuf) params.get(j+1); entries.add(new LocalCachedMapUpdate.Entry(key, value)); } msgEncoded = encode(new LocalCachedMapUpdate(entries)); } else { msgEncoded = encode(new LocalCachedMapInvalidate(instanceId, hashes)); } if (invalidateEntryOnChange == 2) { long time = System.currentTimeMillis(); for (byte[] hash : hashes) { byte[] entryId = generateLogEntryId(hash); params.add(time); params.add(entryId); } } if (msgEncoded != null) { params.add(msgEncoded); } final RPromise result = new RedissonPromise(); RFuture future = commandExecutor.evalWriteAsync(getName(), codec, RedisCommands.EVAL_VOID, "for i=3, tonumber(ARGV[2]) + 2, 5000 do " + "redis.call('hmset', KEYS[1], unpack(ARGV, i, math.min(i+4999, tonumber(ARGV[2]) + 2))); " + "end; " + "if ARGV[1] == '1' then " + "redis.call('publish', KEYS[2], ARGV[#ARGV]); " + "end;" + "if ARGV[1] == '2' then " + "for i=tonumber(ARGV[2]) + 2 + 1, #ARGV - 1, 5000 do " + "redis.call('hmset', KEYS[3], unpack(ARGV, i, math.min(i+4999, #ARGV - 1))); " + "end; " + "redis.call('publish', KEYS[2], ARGV[#ARGV]); " + "end;", Arrays.asList(getName(), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), params.toArray()); future.addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { if (!future.isSuccess()) { result.tryFailure(future.cause()); return; } cacheMap(map); result.trySuccess(null); } }); return result; } @Override public RFuture addAndGetOperationAsync(final K key, Number value) { final ByteBuf keyState = encodeMapKey(key); CacheKey cacheKey = toCacheKey(keyState); ByteBuf msg = encode(new LocalCachedMapInvalidate(instanceId, cacheKey.getKeyHash())); byte[] entryId = generateLogEntryId(cacheKey.getKeyHash()); RFuture future = commandExecutor.evalWriteAsync(getName(), StringCodec.INSTANCE, new RedisCommand("EVAL", new NumberConvertor(value.getClass())), "local result = redis.call('HINCRBYFLOAT', KEYS[1], ARGV[1], ARGV[2]); " + "if ARGV[3] == '1' then " + "redis.call('publish', KEYS[2], ARGV[4]); " + "end;" + "if ARGV[3] == '2' then " + "redis.call('zadd', KEYS[3], ARGV[5], ARGV[6]);" + "redis.call('publish', KEYS[2], ARGV[4]); " + "end;" + "return result; ", Arrays.asList(getName(), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), keyState, new BigDecimal(value.toString()).toPlainString(), invalidateEntryOnChange, msg, System.currentTimeMillis(), entryId); future.addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { if (!future.isSuccess()) { return; } V value = future.getNow(); if (value != null) { CacheKey cacheKey = toCacheKey(key); cache.put(cacheKey, new CacheValue(key, value)); } } }); return future; } @Override public RFuture fastPutIfAbsentAsync(final K key, final V value) { RFuture future = super.fastPutIfAbsentAsync(key, value); future.addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { if (!future.isSuccess()) { return; } if (future.getNow()) { CacheKey cacheKey = toCacheKey(key); cache.put(cacheKey, new CacheValue(key, value)); } } }); return future; } @Override public RFuture> readAllValuesAsync() { final List result = new ArrayList(); final List mapKeys = new ArrayList(); for (CacheValue value : cache.values()) { mapKeys.add(encodeMapKey(value.getKey())); result.add((V) value.getValue()); } final RPromise> promise = new RedissonPromise>(); RFuture> future = commandExecutor.evalReadAsync(getName(), codec, ALL_KEYS, "local entries = redis.call('hgetall', KEYS[1]); " + "local result = {};" + "for j, v in ipairs(entries) do " + "if j % 2 ~= 0 then " + "local founded = false;" + "for i = 1, #ARGV, 1 do " + "if ARGV[i] == entries[j] then " + "founded = true;" + "end;" + "end; " + "if founded == false then " + "table.insert(result, entries[j+1]);" + "end;" + "end; " + "end; " + "return result; ", Arrays.asList(getName()), mapKeys.toArray()); future.addListener(new FutureListener>() { @Override public void operationComplete(Future> future) throws Exception { if (!future.isSuccess()) { promise.tryFailure(future.cause()); return; } result.addAll(future.get()); promise.trySuccess(result); } }); return promise; } @Override public RFuture> readAllMapAsync() { final Map result = new HashMap(); List mapKeys = new ArrayList(); for (CacheValue value : cache.values()) { mapKeys.add(encodeMapKey(value.getKey())); result.put((K)value.getKey(), (V)value.getValue()); } final RPromise> promise = new RedissonPromise>(); RFuture> future = readAll(ALL_MAP, mapKeys, result); future.addListener(new FutureListener>() { @Override public void operationComplete(Future> future) throws Exception { if (!future.isSuccess()) { promise.tryFailure(future.cause()); return; } for (java.util.Map.Entry entry : future.getNow().entrySet()) { CacheKey cacheKey = toCacheKey(entry.getKey()); cache.put(cacheKey, new CacheValue(entry.getKey(), entry.getValue())); } result.putAll(future.getNow()); promise.trySuccess(result); } }); return promise; } @Override public void preloadCache() { // Best-attempt warmup - just enumerate as an uncached map (super) and // add anything found into the cache. This does not guarantee to find // entries added during the warmUp, but statistically the cache will have // few misses after this process for(Entry entry : super.entrySet()) { CacheKey cacheKey = toCacheKey(entry.getKey()); cache.put(cacheKey, new CacheValue(entry.getKey(), entry.getValue())); } } @Override public void clearLocalCache() { get(clearLocalCacheAsync()); } @Override public RFuture clearLocalCacheAsync() { final RPromise result = new RedissonPromise(); RFuture future = invalidationTopic.publishAsync(new LocalCachedMapClear()); future.addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { if (!future.isSuccess()) { result.tryFailure(future.cause()); return; } result.trySuccess(null); } }); return result; } @Override public RFuture>> readAllEntrySetAsync() { final Set> result = new HashSet>(); List mapKeys = new ArrayList(); for (CacheValue value : cache.values()) { mapKeys.add(encodeMapKey(value.getKey())); result.add(new AbstractMap.SimpleEntry((K)value.getKey(), (V)value.getValue())); } final RPromise>> promise = new RedissonPromise>>(); RFuture>> future = readAll(ALL_ENTRIES, mapKeys, result); future.addListener(new FutureListener>>() { @Override public void operationComplete(Future>> future) throws Exception { if (!future.isSuccess()) { promise.tryFailure(future.cause()); return; } for (java.util.Map.Entry entry : future.getNow()) { CacheKey cacheKey = toCacheKey(entry.getKey()); cache.put(cacheKey, new CacheValue(entry.getKey(), entry.getValue())); } result.addAll(future.getNow()); promise.trySuccess(result); } }); return promise; } private RFuture readAll(RedisCommand evalCommandType, List mapKeys, R result) { return commandExecutor.evalReadAsync(getName(), codec, evalCommandType, "local entries = redis.call('hgetall', KEYS[1]); " + "local result = {};" + "for j, v in ipairs(entries) do " + "if j % 2 ~= 0 then " + "local founded = false;" + "for i = 1, #ARGV, 1 do " + "if ARGV[i] == entries[j] then " + "founded = true;" + "end;" + "end; " + "if founded == false then " + "table.insert(result, entries[j]);" + "table.insert(result, entries[j+1]);" + "end;" + "end; " + "end; " + "return result; ", Arrays.asList(getName()), mapKeys.toArray()); } @Override protected RFuture replaceOperationAsync(K key, V value) { final ByteBuf keyState = encodeMapKey(key); ByteBuf valueState = encodeMapValue(value); final CacheKey cacheKey = toCacheKey(keyState); byte[] entryId = generateLogEntryId(cacheKey.getKeyHash()); ByteBuf msg = createSyncMessage(keyState, valueState, cacheKey); return commandExecutor.evalWriteAsync(getName(key), codec, RedisCommands.EVAL_MAP_VALUE, "if redis.call('hexists', KEYS[1], ARGV[1]) == 1 then " + "local v = redis.call('hget', KEYS[1], ARGV[1]); " + "redis.call('hset', KEYS[1], ARGV[1], ARGV[2]); " + "if ARGV[3] == '1' then " + "redis.call('publish', KEYS[2], ARGV[4]); " + "end;" + "if ARGV[3] == '2' then " + "redis.call('zadd', KEYS[3], ARGV[5], ARGV[6]);" + "redis.call('publish', KEYS[2], ARGV[4]); " + "end;" + "return v; " + "else " + "return nil; " + "end", Arrays.asList(getName(key), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), keyState, valueState, invalidateEntryOnChange, msg, System.currentTimeMillis(), entryId); } @Override public RFuture replaceAsync(final K key, final V value) { RFuture future = super.replaceAsync(key, value); future.addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { if (!future.isSuccess()) { return; } if (future.getNow() != null) { CacheKey cacheKey = toCacheKey(key); cache.put(cacheKey, new CacheValue(key, value)); } } }); return future; } @Override protected RFuture replaceOperationAsync(K key, V oldValue, V newValue) { final ByteBuf keyState = encodeMapKey(key); ByteBuf oldValueState = encodeMapValue(oldValue); ByteBuf newValueState = encodeMapValue(newValue); final CacheKey cacheKey = toCacheKey(keyState); byte[] entryId = generateLogEntryId(cacheKey.getKeyHash()); ByteBuf msg = createSyncMessage(keyState, newValueState, cacheKey); return commandExecutor.evalWriteAsync(getName(key), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if redis.call('hget', KEYS[1], ARGV[1]) == ARGV[2] then " + "redis.call('hset', KEYS[1], ARGV[1], ARGV[3]); " + "if ARGV[4] == '1' then " + "redis.call('publish', KEYS[2], ARGV[5]); " + "end;" + "if ARGV[4] == '2' then " + "redis.call('zadd', KEYS[3], ARGV[6], ARGV[7]);" + "redis.call('publish', KEYS[2], ARGV[5]); " + "end;" + "return 1; " + "else " + "return 0; " + "end", Arrays.asList(getName(key), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), keyState, oldValueState, newValueState, invalidateEntryOnChange, msg, System.currentTimeMillis(), entryId); } @Override public RFuture replaceAsync(final K key, V oldValue, final V newValue) { final CacheKey cacheKey = toCacheKey(key); RFuture future = super.replaceAsync(key, oldValue, newValue); future.addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { if (!future.isSuccess()) { return; } if (future.getNow()) { cache.put(cacheKey, new CacheValue(key, newValue)); } } }); return future; } @Override protected RFuture removeOperationAsync(Object key, Object value) { final ByteBuf keyState = encodeMapKey(key); ByteBuf valueState = encodeMapValue(value); final CacheKey cacheKey = toCacheKey(keyState); byte[] entryId = generateLogEntryId(cacheKey.getKeyHash()); ByteBuf msg = encode(new LocalCachedMapInvalidate(instanceId, cacheKey.getKeyHash())); return commandExecutor.evalWriteAsync(getName(key), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if redis.call('hget', KEYS[1], ARGV[1]) == ARGV[2] then " + "if ARGV[3] == '1' then " + "redis.call('publish', KEYS[2], ARGV[4]); " + "end;" + "if ARGV[3] == '2' then " + "redis.call('zadd', KEYS[3], ARGV[5], ARGV[6]);" + "redis.call('publish', KEYS[2], ARGV[4]); " + "end;" + "return redis.call('hdel', KEYS[1], ARGV[1]) " + "else " + "return 0 " + "end", Arrays.asList(getName(key), invalidationTopic.getChannelNames().get(0), getUpdatesLogName()), keyState, valueState, invalidateEntryOnChange, msg, System.currentTimeMillis(), entryId); } @Override public RFuture removeAsync(Object key, Object value) { final CacheKey cacheKey = toCacheKey(key); RFuture future = super.removeAsync(key, value); future.addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { if (!future.isSuccess()) { return; } if (future.getNow()) { cache.remove(cacheKey); } } }); return future; } @Override public RFuture putIfAbsentAsync(final K key, final V value) { RFuture future = super.putIfAbsentAsync(key, value); future.addListener(new FutureListener() { @Override public void operationComplete(Future future) throws Exception { if (!future.isSuccess()) { return; } if (future.getNow() == null) { CacheKey cacheKey = toCacheKey(key); cache.put(cacheKey, new CacheValue(key, value)); } } }); return future; } @Override public ByteBuf encode(Object value) { try { return topicCodec.getValueEncoder().encode(value); } catch (IOException e) { throw new IllegalArgumentException(e); } } }