package com.zhaoonline.redis.transaction;

import java.util.concurrent.TimeUnit;

import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.BoundValueOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.TimeoutUtils;

import com.zhaoonline.redis.template.DeSerializer;
import com.zhaoonline.redis.template.RedisTemplateFactory;

import redis.clients.util.SafeEncoder;

/**
 * class name:TransactionValueOperation <BR>
 * class description: 所有的操作都是基于Transaction的<BR>
 * Remark: <BR>
 * @version 1.00 2016年8月22日
 * @author zhaoonline)yangyoupeng
 */
public class TransactionValueOperation<V> {

	private RedisTemplate<String,V>  redisTemplate;
	private DeSerializer deSerializer=new DeSerializer();
	
	public TransactionValueOperation(RedisTemplateFactory factory){
	   redisTemplate = factory.createTemplate();
	   deSerializer.setValueSerializer(redisTemplate.getValueSerializer());
	   deSerializer.setKeySerializer(redisTemplate.getKeySerializer());
	   
	}
	
	public void setIfAbsent(String key,V value){
		final byte[] rawKey = deSerializer.rawKey(key);
		final byte[] rawValue = deSerializer.rawValue(value);
		redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				connection.watch(rawKey);
				connection.multi();
				connection.setNX(rawKey, rawValue);
				connection.exec();
				connection.unwatch();
				return null;
			}
		});
	}
	
	public void  set(String key,V value){
		final byte[] rawValue = deSerializer.rawValue(value);
		final byte[] rawKey = deSerializer.rawKey(key);
		redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) throws DataAccessException {
				connection.watch(rawKey);
				connection.multi();
				connection.set(rawKey, rawValue);
				connection.exec();
				connection.unwatch();
				return null;
			}
		});
	}
	public void  set(String key,V value,long expireTimeount,TimeUnit unit){
		final byte[] rawKey = deSerializer.rawKey(key);
		final byte[] rawValue = deSerializer.rawValue(value);
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) throws DataAccessException {

				potentiallyUsePsetEx(connection);
				return null;
			}
			public void potentiallyUsePsetEx(RedisConnection connection) {

				if (!TimeUnit.MILLISECONDS.equals(unit) || !failsafeInvokePsetEx(connection)) {
					connection.watch(rawKey);
					connection.multi();
					connection.setEx(rawKey, TimeoutUtils.toSeconds(expireTimeount, unit), rawValue);
					connection.exec();
					connection.unwatch();
					
				}
			}
			private boolean failsafeInvokePsetEx(RedisConnection connection) {
				boolean failed = false;
				try {
					connection.pSetEx(rawKey, expireTimeount, rawValue);
				} catch (UnsupportedOperationException e) {
					// in case the connection does not support pSetEx return false to allow fallback to other operation.
					failed = true;
				}
				return !failed;
			}

		}, true);
	}
	
	public  V  get(String key){
		final byte[] rawKey = deSerializer.rawKey(key);
		return redisTemplate.execute(new RedisCallback<V>() {
			@Override
			public V doInRedis(RedisConnection connection) throws DataAccessException {
				byte[] value=connection.get(rawKey);
				V valueObject=(V)deSerializer.getValueSerializer().deserialize(value);
				return valueObject;
			}
		});
	}
	
	/**
	 * Method name: delete <BR>
	 * Description: <BR>
	 * Remark: <BR>
	 * @param key  void<BR>
	 */
	public void delete(String key){
		final byte[] rawKey = deSerializer.rawKey(key);
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) {
				connection.del(rawKey);
				return null;
			}
		});
	}
	
	public RedisTemplate<String, V> getRedisTemplate() {
		return redisTemplate;
	}	
}