package com.waytous.anticollision.tcp

import com.blankj.utilcode.util.LogUtils
import com.waytous.anticollision.BuildConfig
import com.waytous.anticollision.utils.NamedThreadFactory
import io.netty.buffer.UnpooledByteBufAllocator
import java.io.IOException
import java.io.InputStream
import java.io.OutputStream
import java.net.InetSocketAddress
import java.net.Socket
import java.net.SocketTimeoutException
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.locks.ReentrantLock

/**
 * 缓冲区8M
 * */
const val BUFFER_SIZE = 8192

/**
 * 默认超时时间
 * */
const val DEFAULT_TIMEOUT = 32

/**
 * 每次希望读取的数据长度
 * */
const val READ_SIZE = 1024

/**
 * 链接状态
 * @author male
 * */
internal enum class State {
    /**
     * 链接断开
     * */
    Disconnected,

    /**
     * 正在链接
     * */
    Connecting,

    /**
     * 已经链接到服务器
     * */
    Connected
}

/**
 * tcp链接管理类
 * @param connectListener
 * @param receiveInterval
 * @author male
 * */
internal class TcpManager(
    private val connectListener: ConnectListener,
    private val receiveInterval: Int = 2
) {

    /**
     * 接收线程
     * */
    private val receiveExecutor by lazy {
        Executors.newSingleThreadExecutor(NamedThreadFactory("Receive"))
    }

    /**
     * 发送线程
     * */
    private val sendExecutors: ExecutorService by lazy {
        Executors.newCachedThreadPool(NamedThreadFactory("Send"))
    }

    /**
     * 一条tcp链接
     * */
    private val connection by lazy {
        Connection()
    }

    private val mCancel by lazy {
        AtomicBoolean(true)
    }

    /**
     * access tcp server
     * @param host
     * @param port
     * @param timeout
     * */
    fun connect(host: String, port: Int, timeout: Int = 40) {
        receiveExecutor.execute {
            val errCode = connection.connect(host, port, timeout)
            if (errCode != Error.NOError) {
                connectListener.onDisconnect(errCode)
                return@execute
            }
            connectListener.onConnect()
            mCancel.set(false)
            receive(receiveInterval)
        }
    }

    /**
     * disconnect from tcp server
     * @param error
     * */
    fun disconnect(error: Error = Error.UserDisconnected) {
        if (connection.connectStatus.get() == State.Disconnected) {
            return
        }
        receiveExecutor.execute {
            connection.disconnect()
            mCancel.set(true)
            connectListener.onDisconnect(error)
        }
    }

    /**
     * 发送消息
     * @param data
     * */
    fun send(data: ByteArray) {
        if (connection.connectStatus.get() == State.Connected) {
            sendExecutors.execute { connection.send(data) }
        }
    }

    /**
     * 发送消息
     * @param timeout
     * */
    private fun receive(timeout: Int) {
        if (mCancel.get()) {
            return
        }
        val (errCode, length) = connection.receive()
        if (errCode == Error.NOError) {
            if (length > 0) {
                val buf = UnpooledByteBufAllocator.DEFAULT.buffer(length)
                buf.writeBytes(connection.buffer, 0, length)
                connectListener.onDataReceived(buf)
                connection.buffer.fill(0, 0, length - 1)
            }
            receiveExecutor.execute { receive(timeout) }
        } else {
            mCancel.set(true)
            connectListener.onDisconnect(errCode)
        }
    }

    data class ReceivedResult(val error: Error, val length: Int)

    /**
     * tcp链接
     * */
    class Connection {

        /**
         * 接收线程锁
         * */
        private val receiveLock by lazy {
            ReentrantLock()
        }

        /**
         * 发送线程锁
         * */
        private val sendLock by lazy {
            ReentrantLock()
        }

        private val socket by lazy {
            Socket().apply {
                keepAlive = true
                receiveBufferSize = BUFFER_SIZE
            }
        }

        /**
         * 数据接收流
         * */
        private lateinit var inputStream: InputStream

        /**
         * 数据发送流
         * */
        private lateinit var outputStream: OutputStream

        /**
         * 链接状态
         * */
        val connectStatus by lazy {
            AtomicReference(State.Disconnected)
        }

        /**
         * 接收缓冲区
         * */
        val buffer by lazy {
            ByteArray(BUFFER_SIZE)
        }

        /**
         * 链接到指定服务器
         * @param host
         * @param port
         * @param timeout
         * @return Error
         * */
        fun connect(host: String, port: Int, timeout: Int, receiveInterval: Int = 2): Error {
            try {
                receiveLock.lock()
                sendLock.lock()
                if (socket.isConnected && connectStatus.get() > State.Disconnected) {
                    return Error.NOError
                }
                connectStatus.set(State.Connecting)
                socket.soTimeout = receiveInterval
                socket.connect(
                    InetSocketAddress(host, port),
                    if (timeout > 0) timeout else DEFAULT_TIMEOUT
                )
                inputStream = socket.getInputStream()
                outputStream = socket.getOutputStream()
                connectStatus.set(State.Connected)
                return Error.NOError
            } catch (e: IOException) {
                connectStatus.set(State.Disconnected)
                return Error.IOError
            } catch (e: SocketTimeoutException) {
                connectStatus.set(State.Disconnected)
                return Error.Timeout
            } catch (e: IllegalArgumentException) {
                connectStatus.set(State.Disconnected)
                return Error.InvalidParam
            } catch (e: Exception) {
                connectStatus.set(State.Disconnected)
                return Error.ServerUnknownError
            } finally {
                sendLock.unlock()
                receiveLock.unlock()
            }
        }

        /**
         * 发送数据
         * @param data
         * @return Error
         * */
        fun send(data: ByteArray): Error {
            if (data.isEmpty()) return Error.NOError
            try {
                sendLock.lock()
                outputStream.write(data)
                outputStream.flush()
            } catch (e: IOException) {
                if (BuildConfig.DEBUG) {
                    LogUtils.e("send data error:${e.message}")
                } else {
                    LogUtils.file("send data error:${e.message}")
                }
            } catch (e: Exception) {
                if (BuildConfig.DEBUG) {
                    LogUtils.e("send data error:${e.message}")
                } else {
                    LogUtils.file("send data error:${e.message}")
                }
            } finally {
                sendLock.unlock()
            }
            return Error.NOError
        }

        private fun closeSocket() {
            socket.close()
        }

        /**
         * 接收数据
         * @return ReceivedResult
         * */
        fun receive(): ReceivedResult {
            try {
                receiveLock.lock()
                if (connectStatus.get() != State.Connected) return ReceivedResult(
                    Error.NotConnected,
                    0
                )
                if (inputStream.available() <= 0) return ReceivedResult(Error.NOError, 0)
                val len = inputStream.read(buffer)
                return if (len > 0) {
                    ReceivedResult(Error.NOError, len)
                } else {
                    if (len == 0) {
                        connectStatus.set(State.Disconnected)
                        ReceivedResult(Error.StreamClosed, len)
                    } else {
                        cleanup()
                        connectStatus.set(State.Disconnected)
                        ReceivedResult(Error.IOError, len)
                    }
                }
            } catch (e: IOException) {
                connectStatus.set(State.Disconnected)
                return ReceivedResult(Error.IOError, -1)
            } catch (e: SocketTimeoutException) {
                connectStatus.set(State.Disconnected)
                return ReceivedResult(Error.Timeout, 0)
            } catch (e: IllegalArgumentException) {
                connectStatus.set(State.Disconnected)
                return ReceivedResult(Error.InvalidParam, 0)
            } catch (e: Exception) {
                connectStatus.set(State.Disconnected)
                return ReceivedResult(Error.ServerUnknownError, 0)
            } finally {
                receiveLock.unlock()
            }
        }

        /**
         * 断开链接
         * */
        fun disconnect() = cleanup()

        /**
         * 断开链接
         * */
        private fun cleanup() = try {
            receiveLock.lock()
            sendLock.lock()
            closeSocket()
            connectStatus.set(State.Disconnected)
        } catch (e: Exception) {
            if (BuildConfig.DEBUG) {
                LogUtils.e("cleanup error:${e.message}")
            } else {
                LogUtils.file("cleanup error:${e.message}")
            }
        } finally {
            sendLock.unlock()
            receiveLock.unlock()
        }
    }

}