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

import com.ctrip.xpipe.api.monitor.EventMonitor;
import com.ctrip.xpipe.exception.XpipeException;
import com.ctrip.xpipe.netty.ByteBufReadAction;
import com.ctrip.xpipe.netty.ChannelTrafficStatisticsHandler;
import com.ctrip.xpipe.redis.keeper.RedisKeeperServer;
import com.ctrip.xpipe.redis.keeper.RedisMasterReplication;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;

/**
 * @author wenchao.meng
 *
 * 2016年4月21日 下午3:09:44
 */
public class NettySlaveHandler extends ChannelTrafficStatisticsHandler{

    private RedisKeeperServer redisKeeperServer;
    
	private RedisMasterReplication redisMasterReplication;
	
	public NettySlaveHandler(RedisMasterReplication redisMasterReplication, RedisKeeperServer redisKeeperServer, long trafficReportIntervalMillis) {
	    super(trafficReportIntervalMillis);
		this.redisMasterReplication = redisMasterReplication;
		this.redisKeeperServer = redisKeeperServer;
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		
		Channel channel = ctx.channel();

		redisMasterReplication.masterConnected(channel);
		super.channelActive(ctx);
	}

	
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		
		if(logger.isInfoEnabled()){
			logger.info("[channelInactive]" + ctx.channel());
		}
		
		redisMasterReplication.masterDisconntected(ctx.channel());
		super.channelInactive(ctx);
	}
	
	@Override
	protected void doChannelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		
		ByteBuf byteBuf = (ByteBuf) msg;
		byteBufReadPolicy.read(ctx.channel(), byteBuf, new ByteBufReadAction() {
			@Override
			public void read(Channel channel, ByteBuf byteBuf) throws XpipeException {
				redisMasterReplication.handleResponse(channel, byteBuf);
			}
		});
	}
	
	@Override
	protected void doReportTraffic(long readBytes, long writtenBytes, String remoteIp, int remotePort) {
        if(readBytes > 0) {
            String type = String.format("Keeper.In.%s", redisKeeperServer.getClusterId());
            String name = redisKeeperServer.getShardId();
            EventMonitor.DEFAULT.logEvent(type, name, readBytes);
        }
    }

    @Override
    protected void doWrite(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        
    }
}

@@ -1918,33 +1918,36 @@
 %09%7D%0A%09%0A%09@Override%0A
-%09
+    
 protected void d
@@ -2041,16 +2041,17 @@
       if
+ 
 (readByt
@@ -2176,16 +2176,42 @@
 g name =
+ String.format(%22%25s-%25s:%25s%22,
  redisKe
@@ -2233,16 +2233,39 @@
 hardId()
+, remoteIp, remotePort)
 ;%0A      

refine transaction name for Keeper.Out