package redis.clients.jedis;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantLock;

import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.SSLParameters;
import javax.net.ssl.SSLSocket;
import javax.net.ssl.SSLSocketFactory;

import com.sun.org.apache.bcel.internal.generic.Select;
import redis.clients.jedis.Protocol.Command;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisDataException;
import redis.clients.jedis.nio.NioTcpComponent;
import redis.clients.jedis.nio.Operation;
import redis.clients.jedis.nio.RedisNioOutputStream;
import redis.clients.util.IOUtils;
import redis.clients.util.RedisInputStream;
import redis.clients.util.RedisOutputStream;
import redis.clients.util.SafeEncoder;

public class Connection implements Closeable {

    private static final byte[][] EMPTY_ARGS = new byte[0][];

    private String host = Protocol.DEFAULT_HOST;
    private int port = Protocol.DEFAULT_PORT;
    private Socket socket;
    private RedisOutputStream outputStream;
    private RedisInputStream inputStream;
    private int pipelinedCommands = 0;
    private int connectionTimeout = Protocol.DEFAULT_TIMEOUT;
    private int soTimeout = Protocol.DEFAULT_TIMEOUT;
    private boolean broken = false;
    private boolean ssl;
    private SSLSocketFactory sslSocketFactory;
    private SSLParameters sslParameters;
    private HostnameVerifier hostnameVerifier;


    /**
     * nio >>------相关修改 ------ jie.pang
     **/

    private ByteBuffer rBf = ByteBuffer.allocate(1024);


    //写的队列
    private LinkedBlockingQueue<Operation> writeQueue = new LinkedBlockingQueue<Operation>();

    //等待读取的队列  (写出成功之后，移入到待读取的队列)
    private LinkedBlockingQueue<Operation> readQueue = new LinkedBlockingQueue<Operation>();

    //读取成功之后放入结果的队列
    private LinkedBlockingQueue<Operation>  resQueue = new LinkedBlockingQueue<Operation>();


//    private


    private volatile boolean nioConnection = false;

    private ReentrantLock nioConnectInitLock = new ReentrantLock();

    private NioTcpComponent tcpComponent;

    private SocketChannel channel;

    private SelectionKey selectionKey;


    /**
     * 初始化Nio connection
     */
    public void nioConnect() {

        if (nioConnection) {
            return;
        }

        try {
            //加锁防止并发执行初始化
            nioConnectInitLock.lock();
            tcpComponent = NioTcpComponent.instance();
            this.channel = tcpComponent.registChannel(this);
            nioConnection = true;
        } finally {
            nioConnectInitLock.unlock();
        }

    }


    public Connection sendNioCommand(final Command cmd, final byte[]... args) throws IOException {

        //连接
        nioConnect();

        //发送命令
        RedisNioOutputStream redisNioOutputStream = new RedisNioOutputStream();
        Protocol.sendCommand(redisNioOutputStream, cmd, args);
        //命令
        byte[] fullCommand = redisNioOutputStream.toBytes();

        Operation operation = new Operation(fullCommand);
        //放入队列成功
        if (writeQueue.offer(operation)) {
            //通知写数据
            tcpComponent.notifyWrite(channel, this);
        }

        return this;
    }

    //循环去写
    public void callBack2Write(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();

        Operation operation = writeQueue.peek();
        while (operation != null) {
            ByteBuffer wBf = operation.getByteBuffer();
            channel.write(wBf);
            if (wBf.hasRemaining()) {
                int write = channel.write(wBf);
                if (write == 0) {
                    wBf.compact();
                    return;
                }
            }

            //写完之后移除并放入待读取队列
            readQueue.offer(writeQueue.poll());
            operation = writeQueue.peek();
        }

        //移除写的事件监听
        key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
    }


    //循环去读取
    public void callBack2Read(SelectionKey key) throws IOException {

        //没有待读取的 opreation
        Operation operation = readQueue.peek();
        if (operation == null) {
            return;
        }

        int read = channel.read(rBf);
        //循环读取
        while (read > 0) {
            rBf.flip();
            while (rBf.hasRemaining()) {
                byte b = rBf.get();
//                System.out.print((char)b);
                boolean over = operation.processResult(b);
//                System.out.println("==="+operation.hashCode());
                if (over) {
//                    readQueue.poll();
                    resQueue.offer(readQueue.poll());
                    operation = readQueue.peek();
                }
            }
            rBf.clear();
            read = channel.read(rBf);
        }
    }


    public void callBackSetSelectionKey(SelectionKey key) {
        if (this.selectionKey == null) {
            this.selectionKey = key;
        }
    }

    /**
     * <<------nio 相关修改 ---------- jie.pang
     **/
    public Connection() {
    }

    public Connection(final String host) {
        this.host = host;
    }

    public Connection(final String host, final int port) {
        this.host = host;
        this.port = port;
    }

    public Connection(final String host, final int port, final boolean ssl) {
        this.host = host;
        this.port = port;
        this.ssl = ssl;
    }

    public Connection(final String host, final int port, final boolean ssl,
                      SSLSocketFactory sslSocketFactory, SSLParameters sslParameters,
                      HostnameVerifier hostnameVerifier) {
        this.host = host;
        this.port = port;
        this.ssl = ssl;
        this.sslSocketFactory = sslSocketFactory;
        this.sslParameters = sslParameters;
        this.hostnameVerifier = hostnameVerifier;
    }

    public Socket getSocket() {
        return socket;
    }

    public int getConnectionTimeout() {
        return connectionTimeout;
    }

    public int getSoTimeout() {
        return soTimeout;
    }

    public void setConnectionTimeout(int connectionTimeout) {
        this.connectionTimeout = connectionTimeout;
    }

    public void setSoTimeout(int soTimeout) {
        this.soTimeout = soTimeout;
    }

    public void setTimeoutInfinite() {
        try {
//            if (!isConnected()) {
//                connect();
//            }
            connect();
            channel.socket().setSoTimeout(0);
        } catch (SocketException ex) {
            broken = true;
            throw new JedisConnectionException(ex);
        }
    }

    public void rollbackTimeout() {
        try {
            socket.setSoTimeout(soTimeout);
        } catch (SocketException ex) {
            broken = true;
            throw new JedisConnectionException(ex);
        }
    }

    protected Connection sendCommand(final Command cmd, final String... args) {
        final byte[][] bargs = new byte[args.length][];
        for (int i = 0; i < args.length; i++) {
            bargs[i] = SafeEncoder.encode(args[i]);
        }
        return sendCommand(cmd, bargs);
    }

    protected Connection sendCommand(final Command cmd) {
        return sendCommand(cmd, EMPTY_ARGS);
    }

    protected Connection sendCommand(final Command cmd, final byte[]... args) {
        try {
            /*connect();
            Protocol.sendCommand(outputStream, cmd, args);*/

            pipelinedCommands++;
            return sendNioCommand(cmd, args);
        } catch (IOException ex) {
      /*
       * When client send request which formed by invalid protocol, Redis send back error message
       * before close connection. We try to read it to provide reason of failure.
       */
//            try {
//                String errorMessage = Protocol.readErrorLineIfPossible(inputStream);
//                if (errorMessage != null && errorMessage.length() > 0) {
//                    ex = new JedisConnectionException(errorMessage, ex.getCause());
//                }
//            } catch (Exception e) {
        /*
         * Catch any IOException or JedisConnectionException occurred from InputStream#read and just
         * ignore. This approach is safe because reading error message is optional and connection
         * will eventually be closed.
         */
//            }
            broken = true;
            throw new JedisConnectionException(ex);
            // Any other exceptions related to connection?
        }
    }

    public String getHost() {
        return host;
    }

    public void setHost(final String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(final int port) {
        this.port = port;
    }

    public void connect() {
         nioConnect();
//        if (!isConnected()) {
//            try {
//                socket = new Socket();
//                // ->@wjw_add
//                socket.setReuseAddress(true);
//                socket.setKeepAlive(true); // Will monitor the TCP connection is
//                // valid
//                socket.setTcpNoDelay(true); // Socket buffer Whetherclosed, to
//                // ensure timely delivery of data
//                socket.setSoLinger(true, 0); // Control calls close () method,
//                // the underlying socket is closed
//                // immediately
//                // <-@wjw_add
//
//                socket.connect(new InetSocketAddress(host, port), connectionTimeout);
//                socket.setSoTimeout(soTimeout);
//
//                if (ssl) {
//                    if (null == sslSocketFactory) {
//                        sslSocketFactory = (SSLSocketFactory) SSLSocketFactory.getDefault();
//                    }
//                    socket = (SSLSocket) sslSocketFactory.createSocket(socket, host, port, true);
//                    if (null != sslParameters) {
//                        ((SSLSocket) socket).setSSLParameters(sslParameters);
//                    }
//                    if ((null != hostnameVerifier) &&
//                            (!hostnameVerifier.verify(host, ((SSLSocket) socket).getSession()))) {
//                        String message = String.format(
//                                "The connection to '%s' failed ssl/tls hostname verification.", host);
//                        throw new JedisConnectionException(message);
//                    }
//                }
//
//                outputStream = new RedisOutputStream(socket.getOutputStream());
//                inputStream = new RedisInputStream(socket.getInputStream());
//            } catch (IOException ex) {
//                broken = true;
//                throw new JedisConnectionException(ex);
//            }
//        }
    }


    @Override
    public void close() {
        disconnect();
    }

    public void disconnect() {

        if (channel != null) {
            IOUtils.closeQuietly(channel);
        }

        if (selectionKey != null) {
            selectionKey.cancel();
        }


//        if (isConnected()) {
//            try {
//                outputStream.flush();
//                socket.close();
//            } catch (IOException ex) {
//                broken = true;
//                throw new JedisConnectionException(ex);
//            } finally {
//                IOUtils.closeQuietly(socket);
//            }
//        }
    }

    public boolean isConnected() {

        return channel!=null;

//        return socket != null && socket.isBound() && !socket.isClosed() && socket.isConnected()
//                && !socket.isInputShutdown() && !socket.isOutputShutdown();
    }

    public String getStatusCodeReply() {
//        flush();
        pipelinedCommands--;

        try {
            Operation operation = resQueue.take();
            byte[] resp = (byte[]) operation.getOperationFuture().get();
            return SafeEncoder.encode(resp);
        } catch (InterruptedException e) {
            throw new JedisConnectionException(e);
        } catch (ExecutionException e) {
            throw new JedisConnectionException(e);
        }

//        final byte[] resp = (byte[]) readProtocolWithCheckingBroken();
//        if (null == resp) {
//            return null;
//        } else {
//            return SafeEncoder.encode(resp);
//        }
    }

    public String getBulkReply() {

        final byte[] result = getBinaryBulkReply();
        if (null != result) {
            return SafeEncoder.encode(result);
        } else {
            return null;
        }
    }

    public byte[] getBinaryBulkReply() {
        flush();
        pipelinedCommands--;
        return (byte[]) readProtocolWithCheckingBroken();
    }

    public Long getIntegerReply() {
        flush();
        pipelinedCommands--;
        return (Long) readProtocolWithCheckingBroken();
    }

    public List<String> getMultiBulkReply() {
        return BuilderFactory.STRING_LIST.build(getBinaryMultiBulkReply());
    }

    @SuppressWarnings("unchecked")
    public List<byte[]> getBinaryMultiBulkReply() {
        flush();
        pipelinedCommands--;
        return (List<byte[]>) readProtocolWithCheckingBroken();
    }

    public void resetPipelinedCount() {
        pipelinedCommands = 0;
    }

    @SuppressWarnings("unchecked")
    public List<Object> getRawObjectMultiBulkReply() {
        return (List<Object>) readProtocolWithCheckingBroken();
    }

    public List<Object> getObjectMultiBulkReply() {
        flush();
        pipelinedCommands--;
        return getRawObjectMultiBulkReply();
    }

    @SuppressWarnings("unchecked")
    public List<Long> getIntegerMultiBulkReply() {
        flush();
        pipelinedCommands--;
        return (List<Long>) readProtocolWithCheckingBroken();
    }

    public List<Object> getAll() {
        return getAll(0);
    }

    public List<Object> getAll(int except) {
        List<Object> all = new ArrayList<Object>();
        flush();
        while (pipelinedCommands > except) {
            try {
                all.add(readProtocolWithCheckingBroken());
            } catch (JedisDataException e) {
                all.add(e);
            }
            pipelinedCommands--;
        }
        return all;
    }

    public Object getOne() {
        flush();
        pipelinedCommands--;
        return readProtocolWithCheckingBroken();
    }

    public boolean isBroken() {
        return broken;
    }

    protected void flush() {
//        try {
//            outputStream.flush();
//        } catch (IOException ex) {
//            broken = true;
//            throw new JedisConnectionException(ex);
//        }
    }

    protected Object readProtocolWithCheckingBroken() {

        Operation operation = null;
        try {
            operation = resQueue.take();
            Object o = operation.getOperationFuture().get();
            return o;
        } catch (InterruptedException e) {
            throw new JedisConnectionException(e);
        } catch (ExecutionException e) {
            throw new JedisConnectionException(e);
        }
//        try {
//            return Protocol.read(inputStream);
//        } catch (JedisConnectionException exc) {
//            broken = true;
//            throw exc;
//        }
    }
}