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());

		}

	}

}