package com.ibeetl.admin.core.conf;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cache.Cache;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
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.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;

/**
 * 支持一二级缓存，使得性能逆天快.默认不开启
 * 
 * @author xiandafu
 *
 */

//@Configuration
//@ConditionalOnProperty(name="springext.cache.enabled", havingValue="true" ,matchIfMissing=false)
public class CacheConfig {

	// 定义一个redis 的频道，默认叫cache，用于pub/sub
	@Value("${springext.cache.redis.topic:cache}")
	String topicName;

	@Bean
	public TowLevelCacheManager cacheManager(RedisTemplate redisTemplate) {

		RedisCacheWriter writer = RedisCacheWriter.lockingRedisCacheWriter(redisTemplate.getConnectionFactory());
		SerializationPair pair = SerializationPair
				.fromSerializer(new JdkSerializationRedisSerializer(this.getClass().getClassLoader()));
		RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig().serializeValuesWith(pair);

		TowLevelCacheManager cacheManager = new TowLevelCacheManager(redisTemplate, writer, config);

		return cacheManager;
	}

	@Bean
	RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
			MessageListenerAdapter listenerAdapter) {

		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
		container.setConnectionFactory(connectionFactory);
		container.addMessageListener(listenerAdapter, new PatternTopic(topicName));
		return container;
	}

	@Bean
	MessageListenerAdapter listenerAdapter(final TowLevelCacheManager cacheManager) {
		return new MessageListenerAdapter(new MessageListener() {

			public void onMessage(Message message, byte[] pattern) {
				byte[] bs = message.getChannel();
				try {
					// Sub 一个消息，通知缓存管理器
					String type = new String(bs, "UTF-8");
					cacheManager.receiver(type);
				} catch (UnsupportedEncodingException e) {
					e.printStackTrace();
					// 不可能出错，忽略
				}

			}

		});
	}

	class TowLevelCacheManager extends RedisCacheManager {
		RedisTemplate redisTemplate;

		public TowLevelCacheManager(RedisTemplate redisTemplate, RedisCacheWriter cacheWriter,
				RedisCacheConfiguration defaultCacheConfiguration) {
			super(cacheWriter, defaultCacheConfiguration);
			this.redisTemplate = redisTemplate;
		}

		// 使用RedisAndLocalCache代替Spring Boot自带的RedisCache
		@Override
		protected Cache decorateCache(Cache cache) {
			return new RedisAndLocalCache(this, (RedisCache) cache);
		}

		// 通过其他分布式节点，缓存改变
		public void publishMessage(String cacheName) {
			this.redisTemplate.convertAndSend(topicName, cacheName);
		}

		// 接受一个消息清空本地缓存
		public void receiver(String name) {
			RedisAndLocalCache cache = ((RedisAndLocalCache) this.getCache(name));
			if (cache != null) {
				cache.clearLocal();
			}
		}

	}

	class RedisAndLocalCache implements Cache {
		// 本地缓存提供
		ConcurrentHashMap<Object, Object> local = new ConcurrentHashMap<Object, Object>();
		RedisCache redisCache;
		TowLevelCacheManager cacheManager;

		public RedisAndLocalCache(TowLevelCacheManager cacheManager, RedisCache redisCache) {
			this.redisCache = redisCache;
			this.cacheManager = cacheManager;
		}

		@Override
		public String getName() {
			return redisCache.getName();
		}

		@Override
		public Object getNativeCache() {
			return redisCache.getNativeCache();
		}

		@Override
		public ValueWrapper get(Object key) {
			ValueWrapper wrapper = (ValueWrapper) local.get(key);
			if (wrapper != null) {
				return wrapper;
			} else {
				// 二级缓存取
				wrapper = redisCache.get(key);
				if (wrapper != null) {
					local.put(key, wrapper);
				}

				return wrapper;
			}

		}

		@Override
		public <T> T get(Object key, Class<T> type) {

			return redisCache.get(key, type);
		}

		@Override
		public <T> T get(Object key, Callable<T> valueLoader) {
			return redisCache.get(key, valueLoader);
		}

		@Override
		public void put(Object key, Object value) {
			System.out.println(value.getClass().getClassLoader());
			redisCache.put(key, value);
			clearOtherJVM();

		}

		@Override
		public ValueWrapper putIfAbsent(Object key, Object value) {
			ValueWrapper v = redisCache.putIfAbsent(key, value);
			clearOtherJVM();
			return v;

		}

		@Override
		public void evict(Object key) {
			redisCache.evict(key);
			clearOtherJVM();

		}

		@Override
		public void clear() {
			redisCache.clear();

		}

		public void clearLocal() {
			this.local.clear();
		}

		protected void clearOtherJVM() {
			cacheManager.publishMessage(redisCache.getName());
		}

	}

}