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

package ants

import (
	"errors"
	//"fmt"
	"sync"
	"time"
)

type Node struct {
	nodeID           string
	host             string
	port             int
	user             string
	password         string
	getClientID      func(string) string
	lock             sync.RWMutex
	clientMp         map[string]*ClientConn
	clientWorkerPool chan string
}

func (that *Node) LeaderCheck() (bool, error) {
	_session, _err := that.NewSession()
	if _err != nil {
		return false, _err
	}
	defer _session.Close()
	return _session.LeaderCheck()
}

func (that *Node) Leader() (string, error) {
	_session, _err := that.NewSession()
	if _err != nil {
		return "0", _err
	}
	defer _session.Close()
	return _session.Leader()
}

func (that *Node) NewSession() (*NodeSession, error) {
	for {
		select {
		case _workerID := <-that.clientWorkerPool:
			_client, _ok := that.getClient(_workerID)
			if !_ok {
				continue
			}
			if _client.Closed() {
				that.delClient(_workerID)
				_client.Close()
				continue
			}
			return NewSession(_client, that), nil
		default:
			_client, _ok := that.newWork(that.getClientID(that.nodeID))
			if _ok {
				return NewSession(_client, that), nil
			} else {
				return nil, errors.New("server is busy")
			}
		}
	}
}

//RecycleConn 回收链接
func (that *Node) RecycleConn(client *ClientConn) bool {
	if client.Closed() {
		return false
	}
	that.setClient(client)
	_ticker := time.NewTicker(1 * time.Second)
	defer _ticker.Stop()
	select {
	case that.clientWorkerPool <- client.GetworkerID():
		return true
	case <-_ticker.C:
		//从链接池中删除
		that.delClient(client.GetworkerID())
		//关闭
		client.CloseSign()
		return false
	}
}

func (that *Node) setClient(client *ClientConn) {
	_workerID := client.GetworkerID()
	that.lock.Lock()
	defer that.lock.Unlock()
	that.clientMp[_workerID] = client
}

func (that *Node) delClient(clientWorkerID string) {
	that.lock.Lock()
	defer that.lock.Unlock()
	delete(that.clientMp, clientWorkerID)
}

func (that *Node) getClient(workerID string) (*ClientConn, bool) {
	that.lock.RLock()
	defer that.lock.RUnlock()
	_client, _ok := that.clientMp[workerID]
	return _client, _ok
}

func (that *Node) initStaticWorker() bool {
	_client, _ok := that.newWork(that.getClientID(that.nodeID))
	if !_ok {
		time.Sleep(time.Duration(500) * time.Millisecond)
		return false
	}
	time.Sleep(time.Duration(1) * time.Second)
	if that.RecycleConn(_client) {
		return false
	}
	return true
}

func (that *Node) newWork(workID string) (*ClientConn, bool) {
	_client := new(ClientConn)
	_client.host = that.host
	_client.port = that.port
	_client.user = that.user
	_client.password = that.password
	_client.workerID = workID
	_client.requestID = 1
	_client.requestLock = make(chan int64, 1)
	_client.writeLock = make(chan []byte, 1)
	_client.responseData = map[int64]AntsReqResult{}
	_client.AddCloser(func() {
		that.delClient(workID)
	})
	if _err := _client.auth(); _err != nil {
		log.Infof("client[%s] new client when auth fail (%s)", workID, _err.Error())
		_client.Close()
		return nil, false
	}
	//
	go func() {
		defer func() {
			if err := recover(); err != nil {
				return
			}
		}()
		defer _client.Close()
		_client.Start()
	}()
	return _client, true
}
