package com.ctrip.xpipe.redis.keeper.handler;


import java.io.IOException;

import com.ctrip.xpipe.redis.keeper.KeeperRepl;
import com.ctrip.xpipe.redis.keeper.RedisClient;
import com.ctrip.xpipe.redis.keeper.RedisKeeperServer;
import com.ctrip.xpipe.redis.keeper.RedisSlave;
import com.ctrip.xpipe.redis.protocal.cmd.Psync;
import com.ctrip.xpipe.redis.protocal.protocal.SimpleStringParser;
import com.ctrip.xpipe.redis.keeper.store.DefaultRdbFileListener;

/**
 * @author wenchao.meng
 *
 * 2016年4月22日 上午11:49:02
 */
public class PsyncHandler extends AbstractCommandHandler{
	
	@Override
	protected void doHandle(String[] args, RedisClient redisClient) {
		
		RedisKeeperServer redisKeeperServer = redisClient.getRedisKeeperServer();
		
		//TODO  backup slower than redis slave, try wait...
		switch(redisKeeperServer.getClusterRole()){
			case BACKUP:
			case UNKNOWN:
				logger.error("[doHandle][server state not right, close connection.]" + redisKeeperServer.getClusterRole());
				try {
					redisClient.close();
				} catch (IOException e) {
					logger.error("[doHandle][close redisClient]" + redisClient, e);
				}
				return;
			case ACTIVE:
				break;
			default:
				throw new IllegalStateException("cluster role ilegal:" + redisKeeperServer.getClusterRole());
		}
		
		RedisSlave redisSlave  = redisClient.becomeSlave();
		if(redisSlave == null){
			logger.warn("[doHandle][psync client already slave]" + redisClient);
			try {
				redisClient.close();
			} catch (IOException e) {
				logger.error("[doHandle]" + redisClient, e);
			}
			return;
		}
		
		
		if(args[0].equals("?")){
			doFullSync(redisSlave);
		}else if(args[0].equals(redisKeeperServer.getKeeperRunid())){
			
			KeeperRepl keeperRepl = redisKeeperServer.getKeeperRepl();
			Long beginOffset = keeperRepl.getBeginOffset();
			Long endOffset = keeperRepl.getEndOffset();
			Long offset = Long.valueOf(args[1]);
			
			if((offset > (endOffset + 1)) || (offset < beginOffset)){
				if(logger.isInfoEnabled()){
					logger.info("[doHandle][offset out of range, do FullSync]" + beginOffset + "," + endOffset + "," + offset);
				}
				doFullSync(redisSlave);
			}else{
				doPartialSync(redisSlave, offset);
			}
		}else{
			doFullSync(redisSlave);
		}
	}

	private void doPartialSync(RedisSlave redisSlave, Long offset) {
		
		if(logger.isInfoEnabled()){
			logger.info("[doPartialSync]" + redisSlave);
		}
		
		SimpleStringParser simpleStringParser = new SimpleStringParser(Psync.PARTIAL_SYNC);
		redisSlave.sendMessage(simpleStringParser.format());
		redisSlave.beginWriteCommands(offset);
	}

	private void doFullSync(RedisSlave redisSlave) {

		try {
			if(logger.isInfoEnabled()){
				logger.info("[doFullSync]" + redisSlave);
			}
			RedisKeeperServer redisKeeperServer = redisSlave.getRedisKeeperServer();
			redisKeeperServer.readRdbFile(new DefaultRdbFileListener(redisSlave));
		} catch (IOException e) {
			logger.error("[doFullSync][close client]" + redisSlave, e);
			try {
				redisSlave.close();
			} catch (IOException e1) {
				logger.error("[doFullSync]" + redisSlave, e1);
			}
		}
	}

	@Override
	public String[] getCommands() {
		
		return new String[]{"psync", "sync"};
	}
}

@@ -270,16 +270,69 @@
 sSlave;%0A
+import com.ctrip.xpipe.redis.protocal.RedisProtocol;%0A
 import c
@@ -783,16 +783,50 @@
 rver();%0A
+%09%09boolean shouldContinue = false;%0A
 %09%09%0A%09%09//T
@@ -936,16 +936,197 @@
 BACKUP:%0A
+%09%09%09%09logger.error(%22%5BdoHandle%5D%5Bserver state backup, ask slave to wait.%5D%22 + redisKeeperServer.getClusterRole());%0A%09%09%09%09redisClient.sendMessage(RedisProtocol.CRLF.getBytes());%0A%09%09%09%09break;%0A
 %09%09%09case 
@@ -1180,17 +1180,15 @@
 ate 
-not right
+unknown
 , cl
@@ -1389,22 +1389,21 @@
 %09%09%7D%0A%09%09%09%09
+b
 re
-turn
+ak
 ;%0A%09%09%09cas
@@ -1412,16 +1412,43 @@
 ACTIVE:%0A
+%09%09%09%09shouldContinue = true;%0A
 %09%09%09%09brea
@@ -1563,24 +1563,65 @@
 ());%0A%09%09%7D%0A%09%09%0A
+%09%09if(!shouldContinue)%7B%0A%09%09%09return;%0A%09%09%7D%0A%09%09%0A
 %09%09RedisSlave

backup wait