Subversion Repositories SmartDukaan

package com.spice.profitmandi.dao.config;

import org.springframework.data.redis.cache.RedisCache;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.cache.RedisCacheWriter;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.ReturnType;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Map;
import java.util.Set;

// An offer write that commits after a reader's transaction began leaves that reader with the old row;
// without this gate the reader caches it once the writer has evicted and it stays stale until the TTL
public class OfferCacheGate {

        private static final byte[] LAST_CHANGE_KEY = "offer.cache.lastChange".getBytes(StandardCharsets.UTF_8);
        // fofo, web and cron share one host clock; the margin absorbs small drift
        private static final long CLOCK_MARGIN_MS = 2000;
        // outside a transaction every statement reads fresh data, so only the loader's own run time matters
        private static final long NO_TX_LOOKBACK_MS = 10000;

        private static final byte[] MARK_SCRIPT = (
                        "local cur = tonumber(redis.call('GET', KEYS[1]) or '0') "
                                        + "if tonumber(ARGV[1]) > cur then redis.call('SET', KEYS[1], ARGV[1]) end "
                                        + "return 1").getBytes(StandardCharsets.UTF_8);

        private static final byte[] PUT_IF_FRESH_SCRIPT = (
                        "local last = tonumber(redis.call('GET', KEYS[1]) or '0') "
                                        + "if tonumber(ARGV[1]) <= last + tonumber(ARGV[2]) then return 0 end "
                                        + "if tonumber(ARGV[4]) > 0 then redis.call('SET', KEYS[2], ARGV[3], 'PX', ARGV[4]) "
                                        + "else redis.call('SET', KEYS[2], ARGV[3]) end "
                                        + "return 1").getBytes(StandardCharsets.UTF_8);

        private final RedisConnectionFactory connectionFactory;

        public OfferCacheGate(RedisConnectionFactory connectionFactory) {
                this.connectionFactory = connectionFactory;
        }

        // Called after an offer write's transaction completes, before its evictions
        public void markChanged() {
                eval(MARK_SCRIPT, 1, LAST_CHANGE_KEY, bytes(System.currentTimeMillis()));
        }

        private void putIfFresh(byte[] key, byte[] value, Duration ttl) {
                Long txStart = TxStartTrackingTransactionManager.currentTxStartMillis();
                long readerStart = txStart != null ? txStart : System.currentTimeMillis() - NO_TX_LOOKBACK_MS;
                eval(PUT_IF_FRESH_SCRIPT, 2, LAST_CHANGE_KEY, key, bytes(readerStart), bytes(CLOCK_MARGIN_MS), value,
                                bytes(ttl.toMillis()));
        }

        public RedisCacheManager manager(RedisCacheConfiguration defaults, Map<String, RedisCacheConfiguration> initial,
                        Set<String> gatedCaches) {
                return new Manager(RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory), defaults, initial,
                                gatedCaches);
        }

        private void eval(byte[] script, int numKeys, byte[]... keysAndArgs) {
                RedisConnection connection = connectionFactory.getConnection();
                try {
                        connection.scriptingCommands().eval(script, ReturnType.INTEGER, numKeys, keysAndArgs);
                } finally {
                        connection.close();
                }
        }

        private static byte[] bytes(long value) {
                return Long.toString(value).getBytes(StandardCharsets.UTF_8);
        }

        private class Manager extends RedisCacheManager {
                private final Set<String> gatedCaches;

                Manager(RedisCacheWriter writer, RedisCacheConfiguration defaults, Map<String, RedisCacheConfiguration> initial,
                                Set<String> gatedCaches) {
                        super(writer, defaults, initial);
                        this.gatedCaches = gatedCaches;
                }

                @Override
                protected RedisCache createRedisCache(String name, RedisCacheConfiguration config) {
                        RedisCache cache = super.createRedisCache(name, config);
                        return gatedCaches.contains(name) ? new GatedCache(cache) : cache;
                }
        }

        private class GatedCache extends RedisCache {
                GatedCache(RedisCache cache) {
                        super(cache.getName(), cache.getNativeCache(), cache.getCacheConfiguration());
                }

                @Override
                public void put(Object key, Object value) {
                        Object cacheValue = preProcessCacheValue(value);
                        if (cacheValue == null) {
                                super.put(key, value);
                                return;
                        }
                        putIfFresh(serializeCacheKey(createCacheKey(key)), serializeCacheValue(cacheValue),
                                        getCacheConfiguration().getTtl());
                }
        }
}