/**
 * @desc		TCP框架
 * ********************************************
 * @author		unphp <unphp@qq.com>
 * @date		2021-12-17
 * @copyright	ants 0.1
 * ********************************************
 */

package ants

import (
	"PPOSUtils/utils"
	"errors"
	//"fmt"
	"net"
	"strconv"
	"strings"
	"sync"
	"time"
)

//ClientConn TCP客户端链接
type ClientConn struct {
	lock         sync.RWMutex
	wg           sync.WaitGroup
	conn         *net.TCPConn
	host         string
	port         int
	user         string
	password     string
	workerID     string
	requestID    int64
	aliveTime    int64
	isLeader     bool
	leaderUpTime int64
	requestLock  chan int64
	writeLock    chan []byte
	responseData map[int64]AntsReqResult
	closeSignal  bool
	closerSlice  []func()
}

//AntsReqResult 请求的响应
type AntsReqResult struct {
	requestID    int64
	responseCode int64
	responseData []byte
}

func (that *ClientConn) AddCloser(closer func()) {
	if that.closerSlice == nil {
		that.closerSlice = []func(){}
	}
	that.closerSlice = append(that.closerSlice, closer)
}

func (that *ClientConn) Start() {
	that.wg.Add(2)
	go func() {
		if _err := that.tasker(); _err != nil {
			log.Infof("client[%s] will close, tasker return error (%s)", that.workerID, _err.Error())
		}
		that.CloseSign()
		that.wg.Done()
	}()
	go func() {
		if _err := that.keepAlive(); _err != nil {
			log.Infof("client[%s] will close, keepAlive return error (%s)", that.workerID, _err.Error())
		}
		that.CloseSign()
		that.wg.Done()
	}()
	that.wg.Wait()
	if that.closerSlice == nil {
		for _, _closer := range that.closerSlice {
			_closer()
		}
	}
	return
}

func (that *ClientConn) GetworkerID() string {
	return that.workerID
}

func (that *ClientConn) leaderSet(lsLeader bool) {
	that.lock.Lock()
	defer that.lock.Unlock()
	that.leaderUpTime++
	that.isLeader = lsLeader
}

func (that *ClientConn) leaderGet() (int64, bool) {
	that.lock.RLock()
	defer that.lock.RUnlock()
	return that.leaderUpTime, that.isLeader
}

func (that *ClientConn) Leader() (string, error) {
	that.requestLock <- 0
	defer func() {
		<-that.requestLock
	}()
	_requestID, _err := that.write("leader", nil)
	if _err != nil {
		log.Errorf("[ClientConn.Leader] conn write error:%s", _err.Error())
		return "0", _err
	}
	_result, _err := that.getReqResult(_requestID, 1)
	if _err != nil {
		log.Errorf("[ClientConn.Leader] result error:%s", _err.Error())
		return "0", _err
	}
	if _result.responseCode == 200 {
		log.Infof("[%s ClientConn.Leader] success, leader:%s", that.workerID, _result.responseData)
		return string(_result.responseData), nil
	}
	return "0", nil
}

func (that *ClientConn) LeaderCheck() (bool, error) {
	that.requestLock <- 0
	defer func() {
		<-that.requestLock
	}()
	_requestID, _err := that.write("isLeader", nil)
	if _err != nil {
		return false, _err
	}
	_result, _err := that.getReqResult(_requestID, 1)
	if _err != nil {
		return false, _err
	}
	if _result.responseCode == 200 {
		return true, nil
	}
	return false, nil
}

func (that *ClientConn) keepAlive() error {
	for {
		if that.Closed() {
			return nil
		}
		//最后一次报文通信时间
		_aliveTime := that.aliveTimeGet()
		if _aliveTime == 0 {
			if _err := that.Ping(); _err != nil {
				return _err
			}
			continue
		}
		//距离上次(报文通信时间)的时间(秒）
		_i := int64(0)
		_now := time.Now().Unix()
		if _aliveTime > 0 && (_now-_aliveTime) > 0 {
			_i = _now - _aliveTime
		}

		//间隔大于10秒，说明超时
		if _i > 10 {
			return errors.New(that.workerID + " client keepAlive timeout")
		}
		//大于6秒，心跳检测
		if _i >= 6 {
			if _err := that.Ping(); _err != nil {
				return _err
			}
			time.Sleep(time.Duration(_i) * time.Second)
			continue
		}
		//小于6秒，延后
		time.Sleep(time.Duration(5-_i) * time.Second)
	}
}

func (that *ClientConn) NewRequest(path string, data []byte) (int64, []byte, error) {
	if that.Closed() {
		return 403, nil, errors.New(that.workerID + " client conn closed")
	}
	//阻塞上锁
	that.requestLock <- 1
	defer func() {
		<-that.requestLock
	}()
	//
	_path := that.fiexLeninfo(path, 60)
	_requestID, _err := that.write("req", append(_path, data...))
	if _err != nil {
		return 403, nil, _err
	}
	//
	_responseData, _err := that.getReqResult(_requestID, 0)
	if _err != nil {
		return 403, nil, _err
	}
	if _responseData.responseCode == 200 {
		return 200, _responseData.responseData, nil
	}
	return _responseData.responseCode, nil, errors.New("testing ... ")
}

//Ping ...
func (that *ClientConn) Ping() error {
	_, _err := that.write("client-ping", nil)
	if _err != nil {
		return _err
	}
	return nil
}

//Lister ...
func (that *ClientConn) tasker() error {
	for {
		//
		if that.Closed() {
			return nil
		}
		//读取
		_cmdType, _cmdLength, _err := that.readHeader()
		if _err != nil {
			return _err
		}
		_body, _err := that.readBody(_cmdLength)
		if _err != nil {
			return _err
		}
		_slice := strings.Split(_cmdType, "|")
		switch _slice[0] {
		case "req":
			if len(_slice) != 3 {
				if len(_slice) > 1 {
					go that.setReqResult(utils.ToInt64(_slice[1]), 0, _body)
				}
			}
			go that.setReqResult(utils.ToInt64(_slice[1]), utils.ToInt64(_slice[2]), _body)
		case "isLeader":
			if len(_slice) != 3 {
				if len(_slice) > 1 {
					go that.setReqResult(utils.ToInt64(_slice[1]), 0, _body)
				}
			}
			go that.setReqResult(utils.ToInt64(_slice[1]), utils.ToInt64(_slice[2]), _body)
		case "leader":
			if len(_slice) != 3 {
				if len(_slice) > 1 {
					go that.setReqResult(utils.ToInt64(_slice[1]), 0, _body)
				}
			}
			go that.setReqResult(utils.ToInt64(_slice[1]), utils.ToInt64(_slice[2]), _body)
		default:
			//
		}
	}
}

func (that *ClientConn) readBody(length int) ([]byte, error) {
	_body := []byte{}
	if length > 0 {
		_byte, _err := that.read(length)
		if _err != nil {
			return nil, _err
		}
		_body = _byte
	}
	return _body, nil
}

func (that *ClientConn) aliveTimeGet() int64 {
	that.lock.RLock()
	defer that.lock.RUnlock()
	return that.aliveTime
}

func (that *ClientConn) aliveTimeUp() {
	that.lock.Lock()
	defer that.lock.Unlock()
	that.aliveTime = time.Now().Unix()
}

func (that *ClientConn) setReqResult(requestID, code int64, data []byte) {
	that.lock.Lock()
	defer that.lock.Unlock()
	//responseData 只保留最近5条响应
	_result := AntsReqResult{
		requestID:    requestID,
		responseCode: code,
		responseData: data,
	}
	if requestID > 5 {
		_responseData := map[int64]AntsReqResult{
			requestID: _result,
		}
		_i := int64(1)
		//
		for {
			if _i > 5 {
				break
			}
			_requestID := requestID - _i
			_v, _ok := _responseData[_requestID]
			if _ok {
				_responseData[_requestID] = _v
			}
			_i++
		}
		that.responseData = _responseData
		return
	}
	that.responseData[requestID] = _result
}

func (that *ClientConn) getReqResult(requestID, times int64) (AntsReqResult, error) {
	that.lock.RLock()
	_v, _ok := that.responseData[requestID]
	that.lock.RUnlock()
	if !_ok {
		if times > 50 {
			return AntsReqResult{}, errors.New(that.workerID + "client response timeout")
		}
		time.Sleep(200 * time.Millisecond)
		return that.getReqResult(requestID, times+1)
	}
	return _v, nil
}

//connect 连接
func (that *ClientConn) connect() (*net.TCPConn, error) {
	if that.conn == nil {
		rAddr := &net.TCPAddr{
			IP:   net.ParseIP(that.host),
			Port: that.port,
			Zone: "tcp",
		}
		_conn, _err := net.DialTCP("tcp", nil, rAddr)
		if _err != nil {
			//loger.Error(_err.Error())
			that.conn = nil
			return that.conn, _err
		}
		that.conn = _conn
	}
	return that.conn, nil
}

func (that *ClientConn) auth() error {
	defer func() {
	}()
	if _, _err := that.connect(); _err != nil {
		return _err
	}
	_cmdSlice := []string{
		string(that.fiexLeninfo(that.workerID, 20)),
		that.user,
		that.password,
	}
	_cmdType := "client-auth"
	_, _err := that.write(_cmdType, []byte(strings.Join(_cmdSlice, "")))
	if _err != nil {
		return _err
	}
	//
	_responseCmdType, _cmdLength, _err := that.readHeader()
	if _err != nil {
		return _err
	}
	if _cmdLength > 0 {
		_, _err := that.read(_cmdLength)
		if _err != nil {
			return _err
		}
	}
	_responseCmdTypeSlice := strings.Split(_responseCmdType, "|")
	if len(_responseCmdTypeSlice) != 3 {
		return errors.New("auth fail 001")
	}
	if _responseCmdTypeSlice[0] != "client-auth" {
		return errors.New("auth fail 002")
	}
	if _responseCmdTypeSlice[2] != "200" {
		return errors.New("auth fail 003")
	}
	return nil
}

//readHeader 解析请求报文头，得到报文内容字节的长度
func (that *ClientConn) readHeader() (string, int, error) {
	//解析头
	_headByte, _err := that.read(30)
	if _err != nil {
		return "", 0, _err
	}
	_cmdType := that.trim(_headByte[0:25])
	//
	_bodyLength := 0
	_cmdBodyLength := that.trim(_headByte[25:])
	if _cmdBodyLength != "" {
		_n, _err := strconv.Atoi(that.trim(_headByte[25:]))
		if _err != nil {
			return "", 0, _err
		}
		_bodyLength = _n
	}
	return _cmdType, _bodyLength, nil
}

func (that *ClientConn) read(bodyLength int) ([]byte, error) {
	if bodyLength > 0 {
		_bytes, _err := that.readByte([]byte{}, bodyLength, 1)
		return _bytes, _err
	}
	return []byte{}, nil
}

//readByte 递归接收完报文
func (that *ClientConn) readByte(readByte []byte, length int, readTime int) ([]byte, error) {
	if that.Closed() {
		return readByte, errors.New(that.workerID + " client conn closed")
	}
	if length > 0 {
		_datainfo := make([]byte, length)
		if readTime > 1000 {
			return readByte, errors.New("read lost data more than 1000 times! ")
		}
		_n, _err := that.conn.Read(_datainfo)
		if _err != nil {
			if _n > 0 {
				readByte = append(readByte, _datainfo[0:_n]...)
			}
			that.CloseSign()
			return readByte, _err
		}
		lostLen := length - _n
		if _n > 0 {
			readByte = append(readByte, _datainfo[0:_n]...)
		}
		if lostLen > 0 {
			readTime = readTime + 1
			return that.readByte(readByte, lostLen, readTime)
		}
		return readByte, nil
	}
	return readByte, nil
}

//connWrite 发送报文
func (that *ClientConn) write(cmdType string, cmdByte []byte) (int64, error) {
	if that.Closed() {
		return 0, errors.New(that.workerID + " received close signal")
	}
	//阻塞上锁
	that.writeLock <- []byte{}
	defer func() {
		//释放锁
		<-that.writeLock
		if _err := that.conn.SetWriteDeadline(time.Time{}); _err != nil {
			that.CloseSign()
			return
		}
	}()
	//
	that.requestID = that.requestID + 1
	_cmdType := []string{
		cmdType,
		utils.ToString(that.requestID),
	}
	_sendByte := that.fiexLeninfo(strings.Join(_cmdType, "|"), 25)
	_len := 0
	if cmdByte != nil && len(cmdByte) > 0 {
		_len = len(cmdByte)
		_sendByte = append(_sendByte, that.fiexLeninfo(strconv.Itoa(_len), 5)...)
		_sendByte = append(_sendByte, cmdByte...)
	} else {
		_sendByte = append(_sendByte, that.fiexLeninfo(strconv.Itoa(_len), 5)...)
	}
	that.aliveTimeUp()
	//设置超时
	if _err := that.conn.SetWriteDeadline(time.Now().Add(time.Duration(3) * time.Second)); _err != nil {
		that.CloseSign()
		return 0, _err
	}
	if _, _err := that.conn.Write(_sendByte); _err != nil {
		that.CloseSign()
		return 0, _err
	}
	return that.requestID, nil
}

//返回的“信息头”---包含将要返回多少字节给客户端
func (that *ClientConn) fiexLeninfo(str string, length int) []byte {
	minsend := make([]byte, length)
	first := append([]byte(str), minsend...)
	first = first[0:length]
	return first
}

func (that *ClientConn) trim(b []byte) string {
	emptyByte := make([]byte, 1)
	return strings.TrimSpace(strings.Trim(string(b), string(emptyByte)))
}

func (that *ClientConn) Closed() bool {
	that.lock.RLock()
	defer that.lock.RUnlock()
	return that.closeSignal
}

func (that *ClientConn) CloseSign() {
	that.lock.Lock()
	defer that.lock.Unlock()
	that.closeSignal = true
}

func (that *ClientConn) Close() {
	if that.conn != nil {
		that.conn.Close()
	}
	that.conn = nil
}
