package com.alibaba.otter.canal.parse.driver.mysql.socket;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.util.internal.OutOfDirectMemoryError;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.SystemPropertyUtil;

import java.io.IOException;
import java.net.SocketAddress;

import org.apache.commons.lang.NotImplementedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 封装netty的通信channel和数据接收缓存，实现读、写、连接校验的功能。 2016-12-28
 * 
 * @author luoyaogui
 */
public class NettySocketChannel implements SocketChannel {

    private static final Logger logger                   = LoggerFactory.getLogger(SocketChannel.class);
    private static final int    WAIT_PERIOD              = 10;                                                                   // milliseconds
    private static final int    DEFAULT_INIT_BUFFER_SIZE = 1024 * 1024;                                                          // 1MB，默认初始缓存大小
    // 参考 mysql-connector-java-5.1.40.jar: com.mysql.jdbc.MysqlIO.maxThreeBytes
    // < 256 * 256 * 256 = 16MB
    private static final int    DEFAULT_MAX_BUFFER_SIZE  = 16 * DEFAULT_INIT_BUFFER_SIZE;                                        // 16MB，默认最大缓存大小
    private Channel             channel                  = null;
    private Object              lock                     = new Object();
    private ByteBuf             cache                    = PooledByteBufAllocator.DEFAULT.directBuffer(DEFAULT_INIT_BUFFER_SIZE); // 缓存大小
    private int                 maxDirectBuffer          = cache.maxCapacity();

    public Channel getChannel() {
        return channel;
    }

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

    public void writeCache(ByteBuf buf) throws InterruptedException, IOException {
        synchronized (lock) {
            while (true) {
                if (null == cache) {
                    throw new IOException("socket is closed !");
                }

                // source buffer is empty.
                if (!buf.isReadable()) {
                    break;
                }

                // 默认缓存大小不够用时需自动清理或扩充，否则将因缓存空间不足而造成I/O超时假象
                int length = buf.readableBytes();
                int deltaSize = length - cache.writableBytes();
                if (deltaSize > 0) {
                    // 首先避免频繁分配内存（扩容/收缩），其次避免频繁移动内存（清理）
                    if (cache.readerIndex() >= deltaSize) { // 可以清理
                        // 回收已读空间，重置读写指针
                        cache.discardReadBytes();
                        // 恢复自动扩充的过大缓存到默认初始缓存大小，释放空间
                        int oldCapacity = cache.capacity();
                        if (oldCapacity > DEFAULT_MAX_BUFFER_SIZE) { // 尝试收缩
                            int newCapacity = cache.writerIndex();
                            newCapacity = ((newCapacity - 1) / DEFAULT_INIT_BUFFER_SIZE + 1) * DEFAULT_INIT_BUFFER_SIZE; // 对齐
                            int quarter = (newCapacity >> 2); // 至少留空四分之一
                            quarter = ((quarter - 1) / DEFAULT_INIT_BUFFER_SIZE + 1) * DEFAULT_INIT_BUFFER_SIZE; // 对齐
                            newCapacity += quarter; // 留空四分之一
                            if (newCapacity < (oldCapacity >> 1)) { // 至少收缩二分之一
                                try {
                                    cache.capacity(newCapacity);
                                    logger.info("shrink cache capacity: {} - {} = {} bytes",
                                        oldCapacity,
                                        oldCapacity - newCapacity,
                                        newCapacity);
                                } catch (OutOfMemoryError ignore) {
                                    maxDirectBuffer = oldCapacity; // 未来不再超过当前容量，记录日志后继续
                                    logger.warn("cache OutOfMemoryError: {} bytes", newCapacity, ignore);
                                }
                            }
                        }
                    } else { // 尝试扩容
                        int oldCapacity = cache.capacity();
                        if (oldCapacity < maxDirectBuffer) {
                            int quarter = (oldCapacity >> 2); // 至少扩容四分之一
                            quarter = ((quarter - 1) / DEFAULT_INIT_BUFFER_SIZE + 1) * DEFAULT_INIT_BUFFER_SIZE; // 对齐
                            deltaSize = ((deltaSize - 1) / quarter + 1) * quarter; // 对齐
                            int newCapacity = oldCapacity + deltaSize;
                            if (newCapacity > maxDirectBuffer) {
                                newCapacity = maxDirectBuffer;
                            }
                            try {
                                cache.capacity(newCapacity);
                                logger.info("expand cache capacity: {} + {} = {} bytes",
                                    oldCapacity,
                                    newCapacity - oldCapacity,
                                    newCapacity);
                            } catch (OutOfDirectMemoryError e) {
                                // failed to allocate 885571168 byte(s) of
                                // direct memory (used: 1002946176, max:
                                // 1888485376)
                                long maxDirectMemory = SystemPropertyUtil.getLong("io.netty.maxDirectMemory", -1);
                                if (maxDirectMemory < 0) {
                                    maxDirectMemory = PlatformDependent.maxDirectMemory();
                                }
                                if (maxDirectBuffer > maxDirectMemory) {
                                    maxDirectBuffer = (int) maxDirectMemory;
                                    newCapacity = maxDirectBuffer;
                                    logger.warn("resize maxDirectBuffer: {} bytes", maxDirectBuffer, e);
                                    try {
                                        cache.capacity(newCapacity);
                                        logger.info("expand cache capacity: {} + {} = {} bytes",
                                            oldCapacity,
                                            newCapacity - oldCapacity,
                                            newCapacity);
                                    } catch (OutOfMemoryError ignore) {
                                        maxDirectBuffer = oldCapacity; // 未来不再超过当前容量，记录日志后继续
                                        logger.warn("cache OutOfMemoryError: {} bytes", newCapacity, ignore);
                                    }
                                } else {
                                    maxDirectBuffer = oldCapacity; // 未来不再超过当前容量，记录日志后继续
                                    logger.warn("cache OutOfDirectMemoryError: {} bytes", newCapacity, e);
                                }
                            } catch (OutOfMemoryError ignore) {
                                maxDirectBuffer = oldCapacity; // 未来不再超过当前容量，记录日志后继续
                                logger.warn("cache OutOfMemoryError: {} bytes", newCapacity, ignore);
                            }
                        }
                    }
                    deltaSize = length - cache.writableBytes();
                }

                if (deltaSize != length) {
                    // deltaSize <= 0 可全部写入，deltaSize > 0 只能部分写入
                    if (deltaSize <= 0) {
                        cache.writeBytes(buf, length);
                        break;
                    } else {
                        cache.writeBytes(buf, length - deltaSize);
                    }
                }
                // dest buffer is full.
                lock.wait(WAIT_PERIOD);
                // 回收已读空间，重置读写指针
                cache.discardReadBytes();
            }
        }
    }

    public void write(byte[]... buf) throws IOException {
        if (channel != null && channel.isWritable()) {
            channel.writeAndFlush(Unpooled.copiedBuffer(buf));
        } else {
            throw new IOException("write failed ! please checking !");
        }
    }

    public byte[] read(int readSize) throws IOException {
        return read(readSize, 0);
    }

    public byte[] read(int readSize, int timeout) throws IOException {
        int accumulatedWaitTime = 0;

        // 若读取内容较长，则自动扩充超时时间，以初始缓存大小为基准计算倍数
        if (timeout > 0 && readSize > DEFAULT_INIT_BUFFER_SIZE) {
            timeout *= (readSize / DEFAULT_INIT_BUFFER_SIZE + 1);
        }
        do {
            if (readSize > cache.readableBytes()) {
                if (null == channel) {
                    throw new IOException("socket has Interrupted !");
                }

                if (timeout > 0) {
                    accumulatedWaitTime += WAIT_PERIOD;
                    if (accumulatedWaitTime > timeout) {
                        StringBuilder sb = new StringBuilder("socket read timeout occured !");
                        sb.append(" readSize = ").append(readSize);
                        sb.append(", readableBytes = ").append(cache.readableBytes());
                        sb.append(", timeout = ").append(timeout);
                        throw new IOException(sb.toString());
                    }
                }

                synchronized (this) {
                    try {
                        wait(WAIT_PERIOD);
                    } catch (InterruptedException e) {
                        throw new IOException("socket has Interrupted !");
                    }
                }
            } else {
                byte[] back = new byte[readSize];
                synchronized (lock) {
                    cache.readBytes(back);
                }
                return back;
            }
        } while (true);
    }

    @Override
    public void read(byte[] data, int off, int len, int timeout) throws IOException {
        throw new NotImplementedException();
    }

    public boolean isConnected() {
        return channel != null ? true : false;
    }

    public SocketAddress getRemoteSocketAddress() {
        return channel != null ? channel.remoteAddress() : null;
    }

    public SocketAddress getLocalSocketAddress() {
        return channel != null ? channel.localAddress() : null;
    }

    public void close() {
        if (channel != null) {
            channel.close();
        }
        channel = null;
        // A fatal error has been detected by the Java Runtime Environment:
        // EXCEPTION_ACCESS_VIOLATION (0xc0000005)
        synchronized (lock) {
            cache.discardReadBytes();// 回收已占用的内存
            cache.release();// 释放整个内存
            cache = null;
        }
    }


}
