// 模拟客户端请求
package tpro

import (
	"fmt"
	"io"
	"net"
	"reflect"
	"sync"
	"sync/atomic"

	"github.com/golang/protobuf/proto"
	"gitlab.dianchu.cc/Taotie/log"
)

type TPClient interface {
	Close()
	Start()
	DoRequest(msg []byte, seqId uint32, protoId uint16) error
	Do(request proto.Message, response proto.Message, protoId uint16)
	RegisterCallBack(protoId uint16, callBack func([]byte))
	RegisterAllCallBack(callBack func(protoId uint16, msg []byte))
	SetLogger(logger *log.Logger)
	GetSessionId() string
}
type tpClient struct {
	conn    net.Conn
	id      string
	close   chan bool
	server  func(ctx *ProtoContext) error
	logger  *log.Logger
	cbMutex sync.RWMutex

	seqIdMutex  sync.RWMutex
	cb          map[uint16]func([]byte) // 回调
	response    map[uint32]chan *ProtoContext
	seqId       uint32
	allCallBack func(protoId uint16, msg []byte)
}

func (t *tpClient) RegisterAllCallBack(callBack func(protoId uint16, msg []byte)) {
	t.allCallBack = callBack
}
func (t *tpClient) Do(request proto.Message, response proto.Message, protoId uint16) {
	req, err := proto.Marshal(request)
	if err != nil {
		t.logger.Error(err.Error())
		return
	}

	t.seqIdMutex.Lock()
	curSeqId := t.seqId
	if t.response[curSeqId] == nil { // 存在线程不安全
		t.response[curSeqId] = make(chan *ProtoContext, 1)
	}
	atomic.LoadUint32(&curSeqId) // 手动注入内存屏障，保证对chan的初始化在消息发送前完成,虽然是低概率事件
	// 准备接收数据
	tp := ProtoContext{
		Header: Header{
			Type:         PTRequest,
			IsCompressed: false,
			SeqId:        curSeqId,
			ProtoID:      protoId,
		},
		Body: Body{
			Length: 0,
			Msg:    req,
		},
	}

	if err := tp.EncodeToStream(t.conn); err != nil {
		t.logger.Error(err.Error())
		return
	}

	t.seqId += 1
	t.seqIdMutex.Unlock()
	// 打印发送消息日志
	t.logger.Info(fmt.Sprintf("request:structName:%v proto_id:%v seq_id:%v, type:%v ,is_compressed :%v ", reflect.TypeOf(request).String(), tp.ProtoID, tp.SeqId, tp.Type, tp.IsCompressed))
	t.logger.Info(fmt.Sprintf("msg: %v", request.String()))
	// TODO 超时机制
	resp := <-t.response[curSeqId]
	if resp.SeqId != curSeqId {
		t.logger.Error(fmt.Sprintf("proto_id %v && seq_id %v :response seqId is error!", protoId, curSeqId))
		return
	}
	// 序列化
	if err = proto.Unmarshal(resp.Msg, response); err != nil {
		t.logger.Error(fmt.Sprintf("proto_id %v && seq_id %v : proto unmarshal fail,err:%v", protoId, curSeqId, err))
		return
	}
	// 打印日志
	t.logger.Info(fmt.Sprintf("response:structName:%v proto_id:%v seq_id:%v, type:%v ,is_compressed :%v ", reflect.TypeOf(response).String(), resp.ProtoID, resp.SeqId, resp.Type, resp.IsCompressed))
	t.logger.Info(fmt.Sprintf("msg: %v", response.String()))
	t.callBack(protoId, resp.Msg)
}
func (t *tpClient) callBack(protoId uint16, response []byte) {
	if t.allCallBack != nil {
		t.allCallBack(protoId, response)
	}
	if t.cb[protoId] == nil {
		t.logger.Error(fmt.Sprintf("proto_id:%v callBack fun is nil", protoId))
	} else {
		t.cb[protoId](response)
	}
}
func (t *tpClient) RegisterCallBack(protoId uint16, callBack func(response []byte)) {
	t.cbMutex.Lock()
	defer t.cbMutex.Unlock()
	t.cb[protoId] = callBack
}

func (t *tpClient) SetLogger(logger *log.Logger) {
	t.logger = logger
}

// todo 临时实现 后续要和newclient合并或者优化
func Client(addr string) (tpro *tpClient, err error) {
	var (
		c    = make(chan bool, 1)
		conn net.Conn
	)
	conn, err = net.Dial("tcp", addr)
	if err != nil {
		return nil, err
	}
	t := &tpClient{
		conn:       conn,
		close:      c,
		logger:     defaultLog(),
		cbMutex:    sync.RWMutex{},
		seqIdMutex: sync.RWMutex{},
		cb:         make(map[uint16]func([]byte)), // 回调
		response:   make(map[uint32]chan *ProtoContext),
		seqId:      uint32(0),
	}
	return t, nil
}

func NewClient(addr string, id string) (tpro *tpClient, err error) {
	var (
		c      = make(chan bool, 1)
		conn   net.Conn
		msg    []byte
		result *ProtoContext
	)
	conn, err = net.Dial("tcp", addr)
	if err != nil {
		return nil, err
	}

	t := &tpClient{
		conn:       conn,
		close:      c,
		logger:     defaultLog(),
		cbMutex:    sync.RWMutex{},
		seqIdMutex: sync.RWMutex{},
		cb:         make(map[uint16]func([]byte)), // 回调
		response:   make(map[uint32]chan *ProtoContext),
		seqId:      uint32(1), // 0已经被使用
	}
	cRequest := &ConnRequest{Data: map[string]string{
		"role_id": id,
	}}
	msg, err = proto.Marshal(cRequest)
	if err != nil {
		log.Info(err.Error())
		return
	}
	if err = t.DoRequest(msg, 0, 0); err != nil {
		log.Info(err.Error())
		return
	}
	result, err = NewPContextForStream(t.conn)
	if err != nil {
		log.Info(err.Error())
		return
	}
	if result.SeqId != 0 {
		log.Info(fmt.Sprintf("服务端握手信息错误,seq!=0"))
		return
	}
	connResponse := &ConnResponse{}
	if err = proto.Unmarshal(result.Msg, connResponse); err != nil {
		log.Info(err.Error())
	}
	if connResponse.Code != 0 {
		log.Info(fmt.Sprintf("服务端握手返回code不为0"))
		return
	}
	return t, nil
}

//
func (t *tpClient) Close() {
	t.conn.Close()
}

// GetSessionId 获取链接id
func (t *tpClient) GetSessionId() string {
	return t.id
}

// Start 启动tpro客户端
func (t *tpClient) Start() {
	go func() {
		for {
			c, err := NewPContextForStream(t.conn)
			if err == io.EOF {
				return
			}
			if err != nil {
				t.logger.Error(fmt.Sprintf("read Response fail:\n err:%v", err))
				return
			}
			if c.Type == PTResponse {
				t.response[c.SeqId] <- c
			} else if c.Type == PTPush {

				t.callBack(c.ProtoID, c.Msg)
			} else {
				t.logger.Error(fmt.Sprintf("response type is error:%v", c))
			}
		}
	}()
}

func (t *tpClient) DoRequest(msg []byte, seqId uint32, protoId uint16) error {
	tp := ProtoContext{
		Header: Header{
			Type:         PTRequest,
			IsCompressed: false,
			SeqId:        seqId,
			ProtoID:      protoId,
		},
		Body: Body{
			Length: 0,
			Msg:    msg,
		},
	}
	if err := tp.EncodeToStream(t.conn); err != nil {
		return err
	}
	fmt.Println(tp.Encode())
	fmt.Println(tp.Msg)
	return nil
}
func defaultLog() *log.Logger {
	return log.NewNop()

}
