package com.ctrip.xpipe.redis.proxy.session;

import com.ctrip.xpipe.observer.AbstractLifecycleObservable;
import com.ctrip.xpipe.redis.core.proxy.endpoint.ProxyEndpoint;
import com.ctrip.xpipe.redis.proxy.Session;
import com.ctrip.xpipe.redis.proxy.Tunnel;
import com.ctrip.xpipe.redis.proxy.model.SessionMeta;
import com.ctrip.xpipe.redis.proxy.session.state.SessionClosed;
import com.ctrip.xpipe.utils.ChannelUtil;
import com.ctrip.xpipe.utils.VisibleForTesting;
import com.google.common.collect.Lists;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author chen.zhu
 * <p>
 * May 24, 2018
 */
public abstract class AbstractSession extends AbstractLifecycleObservable implements Session {

    protected final static Logger logger = LoggerFactory.getLogger(AbstractSession.class);

    protected final static String SESSION_STATE_CHANGE = "Session.State.Change";

    protected ProxyEndpoint endpoint;

    protected Channel channel;

    private Tunnel tunnel;

    protected long trafficReportIntervalMillis;

    private List<SessionEventHandler> handlers = Lists.newArrayList();

    private volatile SessionWritableState writableState = SessionWritableState.WRITABLE;

    private AtomicBoolean logRead = new AtomicBoolean(false);

    private AtomicBoolean logWrite = new AtomicBoolean(false);

    protected AbstractSession(Tunnel tunnel, long trafficReportIntervalMillis) {
        this.tunnel = tunnel;
        this.trafficReportIntervalMillis = trafficReportIntervalMillis;
    }

    @Override
    public Tunnel tunnel() {
        return tunnel;
    }

    @Override
    public void addSessionEventHandler(SessionEventHandler handler) {
        handlers.add(handler);
    }

    @Override
    public void makeReadable() {
        ChannelUtil.triggerChannelAutoRead(getChannel());
    }

    @Override
    public void makeUnReadable() {
        ChannelUtil.closeChannelAutoRead(getChannel());
    }

    @Override
    public void setWritableState(SessionWritableState state) {
        if(state == writableState) {
            return;
        }
        writableState = state;
        switch (state) {
            case WRITABLE:
                onSessionWritable();
                break;
            case UNWRITABLE:
                onSessionNotWritable();
                break;

                default: break;
        }
    }

    @Override
    public boolean logRead() {
        return logRead.get();
    }

    @Override
    public boolean logWrite() {
        return logWrite.get();
    }

    @Override
    public void markReadLoggability(boolean isLoggable) {
        logRead.set(isLoggable);
    }

    @Override
    public void markWriteLoggability(boolean isLoggable) {
        logWrite.set(isLoggable);
    }

    protected void onSessionInit() {
        for(SessionEventHandler handler : handlers) {
            handler.onInit();
        }
    }

    protected void onSessionEstablished() {
        for(SessionEventHandler handler : handlers) {
            handler.onEstablished();
        }
    }

    private void onSessionWritable() {
        for(SessionEventHandler handler : handlers) {
            handler.onWritable();
        }
    }

    private void onSessionNotWritable() {
        for(SessionEventHandler handler : handlers) {
            handler.onNotWritable();
        }
    }

    @Override
    public void tryWrite(ByteBuf byteBuf) {
        getSessionState().tryWrite(byteBuf);
    }

    public void doWrite(ByteBuf byteBuf) {
        if(logger.isDebugEnabled()) {
            logger.debug("[doWrite] {}: {}", getSessionType(), ByteBufUtil.prettyHexDump(byteBuf));
        }
        ChannelFuture future = getChannel().write(byteBuf.retain(), getChannel().voidPromise());
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if(isFlushable()) {
                    getChannel().flush();
                }
            }
        });
    }

    private boolean isFlushable() {
        return getChannel().isWritable();
    }

    protected void setSessionState(SessionState newState) {
        if(!getSessionState().isValidNext(newState)) {
            logger.error("[setSessionState] Set state failed, state relationship not match, old: {}, new: {}",
                    getSessionState(), newState.name());
            return;
        }
        doSetSessionState(newState);
    }

    protected abstract void doSetSessionState(SessionState newState);

    @Override
    public Channel getChannel() {
        return channel;
    }

    @Override
    public SessionMeta getSessionMeta() {
        try {
            return new SessionMeta(this, endpoint, getSessionState());
        } catch (Exception e) {
            return new SessionMeta(getSessionType().name(), "unknown", "unknown", getSessionState().name());
        }
    }

    @Override
    public SESSION_TYPE getSessionType() {
        return null;
    }

    @Override
    public void release() {
        if(channel != null) {
            channel.close();
        }
        setSessionState(new SessionClosed(this));
    }

    @Override
    protected void doInitialize() throws Exception {
        super.doInitialize();
    }

    @Override
    protected void doStart() throws Exception {
        super.doStart();
    }

    @VisibleForTesting
    public void setChannel(Channel channel) {
        AbstractSession.this.channel = channel;
    }

}

@@ -604,94 +604,8 @@
 el;%0A
-import io.netty.channel.ChannelFuture;%0Aimport io.netty.channel.ChannelFutureListener;%0A
 impo
@@ -3816,31 +3816,8 @@
     
- ChannelFuture future =
  get
@@ -3827,24 +3827,32 @@
 nnel().write
+AndFlush
 (byteBuf.ret
@@ -3891,377 +3891,8 @@
 ));%0A
-        future.addListener(new ChannelFutureListener() %7B%0A            @Override%0A            public void operationComplete(ChannelFuture future) throws Exception %7B%0A                if(isFlushable()) %7B%0A                    getChannel().flush();%0A                %7D%0A            %7D%0A        %7D);%0A    %7D%0A%0A    private boolean isFlushable() %7B%0A        return getChannel().isWritable();%0A
     

Revert "add flushable check to session"