package com.taobao.tddl.dbsync.binlog;

import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.SocketTimeoutException;
import java.sql.Connection;
import java.sql.SQLException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * TODO: Document It!!
 * 
 * <pre>
 * DirectLogFetcher fetcher = new DirectLogFetcher();
 * fetcher.open(conn, file, 0, 13);
 * 
 * while (fetcher.fetch()) {
 *     LogEvent event;
 *     do {
 *         event = decoder.decode(fetcher, context);
 * 
 *         // process log event.
 *     } while (event != null);
 * }
 * // connection closed.
 * </pre>
 * 
 * @author <a href="mailto:changyuan.lh@taobao.com">Changyuan.lh</a>
 * @version 1.0
 */
public final class DirectLogFetcher extends LogFetcher {

    protected static final Log logger                          = LogFactory.getLog(DirectLogFetcher.class);

    /** Command to dump binlog */
    public static final byte   COM_BINLOG_DUMP                 = 18;

    /** Packet header sizes */
    public static final int    NET_HEADER_SIZE                 = 4;
    public static final int    SQLSTATE_LENGTH                 = 5;

    /** Packet offsets */
    public static final int    PACKET_LEN_OFFSET               = 0;
    public static final int    PACKET_SEQ_OFFSET               = 3;

    /** Maximum packet length */
    public static final int    MAX_PACKET_LENGTH               = (256 * 256 * 256 - 1);

    /** BINLOG_DUMP options */
    public static final int    BINLOG_DUMP_NON_BLOCK           = 1;
    public static final int    BINLOG_SEND_ANNOTATE_ROWS_EVENT = 2;

    private Connection         conn;
    private OutputStream       mysqlOutput;
    private InputStream        mysqlInput;

    public DirectLogFetcher(){
        super(DEFAULT_INITIAL_CAPACITY, DEFAULT_GROWTH_FACTOR);
    }

    public DirectLogFetcher(final int initialCapacity){
        super(initialCapacity, DEFAULT_GROWTH_FACTOR);
    }

    public DirectLogFetcher(final int initialCapacity, final float growthFactor){
        super(initialCapacity, growthFactor);
    }

    private static final Object unwrapConnection(Object conn, Class<?> connClazz) throws IOException {
        while (!connClazz.isInstance(conn)) {
            try {
                Class<?> connProxy = Class.forName("org.springframework.jdbc.datasource.ConnectionProxy");
                if (connProxy.isInstance(conn)) {
                    conn = invokeMethod(conn, connProxy, "getTargetConnection");
                    continue;
                }
            } catch (ClassNotFoundException e) {
                // org.springframework.jdbc.datasource.ConnectionProxy not
                // found.
            }

            try {
                Class<?> connProxy = Class.forName("org.apache.commons.dbcp.DelegatingConnection");
                if (connProxy.isInstance(conn)) {
                    conn = getDeclaredField(conn, connProxy, "_conn");
                    continue;
                }
            } catch (ClassNotFoundException e) {
                // org.apache.commons.dbcp.DelegatingConnection not found.
            }

            try {
                if (conn instanceof java.sql.Wrapper) {
                    Class<?> connIface = Class.forName("com.mysql.jdbc.Connection");
                    conn = ((java.sql.Wrapper) conn).unwrap(connIface);
                    continue;
                }
            } catch (ClassNotFoundException e) {
                // com.mysql.jdbc.Connection not found.
            } catch (SQLException e) {
                logger.warn("Unwrap " + conn.getClass().getName() + " to " + connClazz.getName() + " failed: "
                            + e.getMessage(),
                    e);
            }

            return null;
        }
        return conn;
    }

    private static final Object invokeMethod(Object obj, Class<?> objClazz, String name) {
        try {
            Method method = objClazz.getMethod(name, (Class<?>[]) null);
            return method.invoke(obj, (Object[]) null);
        } catch (NoSuchMethodException e) {
            throw new IllegalArgumentException("No such method: \'" + name + "\' @ " + objClazz.getName(), e);
        } catch (IllegalAccessException e) {
            throw new IllegalArgumentException("Cannot invoke method: \'" + name + "\' @ " + objClazz.getName(), e);
        } catch (InvocationTargetException e) {
            throw new IllegalArgumentException("Invoke method failed: \'" + name + "\' @ " + objClazz.getName(),
                e.getTargetException());
        }
    }

    private static final Object getDeclaredField(Object obj, Class<?> objClazz, String name) {
        try {
            Field field = objClazz.getDeclaredField(name);
            field.setAccessible(true);
            return field.get(obj);
        } catch (NoSuchFieldException e) {
            throw new IllegalArgumentException("No such field: \'" + name + "\' @ " + objClazz.getName(), e);
        } catch (IllegalAccessException e) {
            throw new IllegalArgumentException("Cannot get field: \'" + name + "\' @ " + objClazz.getName(), e);
        }
    }

    /**
     * Connect MySQL master to fetch binlog.
     */
    public void open(Connection conn, String fileName, final int serverId) throws IOException {
        open(conn, fileName, BIN_LOG_HEADER_SIZE, serverId, false);
    }

    /**
     * Connect MySQL master to fetch binlog.
     */
    public void open(Connection conn, String fileName, final int serverId, boolean nonBlocking) throws IOException {
        open(conn, fileName, BIN_LOG_HEADER_SIZE, serverId, nonBlocking);
    }

    /**
     * Connect MySQL master to fetch binlog.
     */
    public void open(Connection conn, String fileName, final long filePosition, final int serverId) throws IOException {
        open(conn, fileName, filePosition, serverId, false);
    }

    /**
     * Connect MySQL master to fetch binlog.
     */
    public void open(Connection conn, String fileName, long filePosition, final int serverId, boolean nonBlocking)
                                                                                                                  throws IOException {
        try {
            this.conn = conn;
            Class<?> connClazz = Class.forName("com.mysql.jdbc.ConnectionImpl");
            Object unwrapConn = unwrapConnection(conn, connClazz);
            if (unwrapConn == null) {
                throw new IOException("Unable to unwrap " + conn.getClass().getName()
                                      + " to com.mysql.jdbc.ConnectionImpl");
            }

            // Get underlying IO streams for network communications.
            Object connIo = getDeclaredField(unwrapConn, connClazz, "io");
            if (connIo == null) {
                throw new IOException("Get null field:" + conn.getClass().getName() + "#io");
            }
            mysqlOutput = (OutputStream) getDeclaredField(connIo, connIo.getClass(), "mysqlOutput");
            mysqlInput = (InputStream) getDeclaredField(connIo, connIo.getClass(), "mysqlInput");

            if (filePosition == 0) filePosition = BIN_LOG_HEADER_SIZE;
            sendBinlogDump(fileName, filePosition, serverId, nonBlocking);
            position = 0;
        } catch (IOException e) {
            close(); /* Do cleanup */
            logger.error("Error on COM_BINLOG_DUMP: file = " + fileName + ", position = " + filePosition);
            throw e;
        } catch (ClassNotFoundException e) {
            close(); /* Do cleanup */
            throw new IOException("Unable to load com.mysql.jdbc.ConnectionImpl", e);
        }
    }

    /**
     * Put a byte in the buffer.
     * 
     * @param b the byte to put in the buffer
     */
    protected final void putByte(byte b) {
        ensureCapacity(position + 1);

        buffer[position++] = b;
    }

    /**
     * Put 16-bit integer in the buffer.
     * 
     * @param i16 the integer to put in the buffer
     */
    protected final void putInt16(int i16) {
        ensureCapacity(position + 2);

        byte[] buf = buffer;
        buf[position++] = (byte) (i16 & 0xff);
        buf[position++] = (byte) (i16 >>> 8);
    }

    /**
     * Put 32-bit integer in the buffer.
     * 
     * @param i32 the integer to put in the buffer
     */
    protected final void putInt32(long i32) {
        ensureCapacity(position + 4);

        byte[] buf = buffer;
        buf[position++] = (byte) (i32 & 0xff);
        buf[position++] = (byte) (i32 >>> 8);
        buf[position++] = (byte) (i32 >>> 16);
        buf[position++] = (byte) (i32 >>> 24);
    }

    /**
     * Put a string in the buffer.
     * 
     * @param s the value to put in the buffer
     */
    protected final void putString(String s) {
        ensureCapacity(position + (s.length() * 2) + 1);

        System.arraycopy(s.getBytes(), 0, buffer, position, s.length());
        position += s.length();
        buffer[position++] = 0;
    }

    protected final void sendBinlogDump(String fileName, final long filePosition, final int serverId,
                                        boolean nonBlocking) throws IOException {
        position = NET_HEADER_SIZE;

        putByte(COM_BINLOG_DUMP);
        putInt32(filePosition);
        int binlog_flags = nonBlocking ? BINLOG_DUMP_NON_BLOCK : 0;
        binlog_flags |= BINLOG_SEND_ANNOTATE_ROWS_EVENT;
        putInt16(binlog_flags); // binlog_flags
        putInt32(serverId); // slave's server-id
        putString(fileName);

        final byte[] buf = buffer;
        final int len = position - NET_HEADER_SIZE;
        buf[0] = (byte) (len & 0xff);
        buf[1] = (byte) (len >>> 8);
        buf[2] = (byte) (len >>> 16);

        mysqlOutput.write(buffer, 0, position);
        mysqlOutput.flush();
    }

    /**
     * {@inheritDoc}
     * 
     * @see com.taobao.tddl.dbsync.binlog.LogFetcher#fetch()
     */
    public boolean fetch() throws IOException {
        try {
            // Fetching packet header from input.
            if (!fetch0(0, NET_HEADER_SIZE)) {
                logger.warn("Reached end of input stream while fetching header");
                return false;
            }

            // Fetching the first packet(may a multi-packet).
            int netlen = getUint24(PACKET_LEN_OFFSET);
            int netnum = getUint8(PACKET_SEQ_OFFSET);
            if (!fetch0(NET_HEADER_SIZE, netlen)) {
                logger.warn("Reached end of input stream: packet #" + netnum + ", len = " + netlen);
                return false;
            }

            // Detecting error code.
            final int mark = getUint8(NET_HEADER_SIZE);
            if (mark != 0) {
                if (mark == 255) // error from master
                {
                    // Indicates an error, for example trying to fetch from
                    // wrong
                    // binlog position.
                    position = NET_HEADER_SIZE + 1;
                    final int errno = getInt16();
                    String sqlstate = forward(1).getFixString(SQLSTATE_LENGTH);
                    String errmsg = getFixString(limit - position);
                    throw new IOException("Received error packet:" + " errno = " + errno + ", sqlstate = " + sqlstate
                                          + " errmsg = " + errmsg);
                } else if (mark == 254) {
                    // Indicates end of stream. It's not clear when this would
                    // be sent.
                    logger.warn("Received EOF packet from server, apparent" + " master disconnected.");
                    return false;
                } else {
                    // Should not happen.
                    throw new IOException("Unexpected response " + mark + " while fetching binlog: packet #" + netnum
                                          + ", len = " + netlen);
                }
            }

            // The first packet is a multi-packet, concatenate the packets.
            while (netlen == MAX_PACKET_LENGTH) {
                if (!fetch0(0, NET_HEADER_SIZE)) {
                    logger.warn("Reached end of input stream while fetching header");
                    return false;
                }

                netlen = getUint24(PACKET_LEN_OFFSET);
                netnum = getUint8(PACKET_SEQ_OFFSET);
                if (!fetch0(limit, netlen)) {
                    logger.warn("Reached end of input stream: packet #" + netnum + ", len = " + netlen);
                    return false;
                }
            }

            // Preparing buffer variables to decoding.
            origin = NET_HEADER_SIZE + 1;
            position = origin;
            limit -= origin;
            return true;
        } catch (SocketTimeoutException e) {
            close(); /* Do cleanup */
            logger.error("Socket timeout expired, closing connection", e);
            throw e;
        } catch (InterruptedIOException e) {
            close(); /* Do cleanup */
            logger.warn("I/O interrupted while reading from client socket", e);
            throw e;
        } catch (IOException e) {
            close(); /* Do cleanup */
            logger.error("I/O error while reading from client socket", e);
            throw e;
        }
    }

    private final boolean fetch0(final int off, final int len) throws IOException {
        ensureCapacity(off + len);

        for (int count, n = 0; n < len; n += count) {
            if (0 > (count = mysqlInput.read(buffer, off + n, len - n))) {
                // Reached end of input stream
                return false;
            }
        }

        if (limit < off + len) limit = off + len;
        return true;
    }

    /**
     * {@inheritDoc}
     * 
     * @see com.taobao.tddl.dbsync.binlog.LogFetcher#close()
     */
    public void close() throws IOException {
        try {
            if (conn != null) conn.close();

            conn = null;
            mysqlInput = null;
            mysqlOutput = null;
        } catch (SQLException e) {
            logger.warn("Unable to close connection", e);
        }
    }
}
