package models

import (
	"basic.com/pubsub/protomsg.git"
	"basic.com/valib/logger.git"
	"github.com/gogo/protobuf/proto"
	"nanomsg.org/go-mangos"
	"nanomsg.org/go-mangos/protocol/req"
	"nanomsg.org/go-mangos/transport/ipc"
	"nanomsg.org/go-mangos/transport/tcp"
	"strconv"
	"time"
	"vamicro/config"
)

var syncSdkCompareCacheChan = make(chan []byte, 512)

func PublishSdkCompareCacheChange(b []byte) {
	syncSdkCompareCacheChan <- b
}

func InitSync() {
	go func() {
		logger.Debug("syncDbTablePerson:recv doUpdateCompareCacheRequest msg")
		var sock mangos.Socket
		var err error
		if sock, err = req.NewSocket(); err != nil {
			logger.Debug("comp cache can't new req socket:%s", err.Error())
			return
		}
		sock.AddTransport(ipc.NewTransport())
		sock.AddTransport(tcp.NewTransport())
		var url = "tcp://" + config.DbPersonCompare.Ip + ":" + strconv.Itoa(config.DbPersonCompare.Port)
		logger.Debug("doUpdateCompareCacheRequest url:", url)
		if err = sock.Dial(url); err != nil {
			logger.Debug("comp cache can't dial on req socket:%s", err.Error())
			return
		}
		sock.SetOption(mangos.OptionMaxRecvSize, 1024*1024*100)
		sock.SetOption(mangos.OptionRecvDeadline, time.Second*2)
		defer sock.Close()

		for {
			select {
			case b := <-syncSdkCompareCacheChan:
				logger.Debug("SyncSdkCompareCache in,len(b):", len(b))
				// doUpdateCompareCacheRequest(b)
				if err = sock.Send(b); err != nil {
					logger.Debug("comp cache can't send message on push socket:%s", err.Error())
					continue
				}
				if msg, err := sock.Recv(); err != nil {
					logger.Debugf("comp cache sock.Recv receive err:%s\n", err.Error())
					continue
				} else {
					logger.Debug("comp cache sock.Recv msg:", string(msg))
				}
			}
		}
	}()
}

func SyncSdkCompareCache(analyServerId string, cacheChangeType protomsg.EsCacheChanged, tableId string, personId string, faceFeature string, dbAction protomsg.DbAction, enable int32, carNo string) {
	msg := protomsg.EsPersonCacheChange{
		Type:     cacheChangeType,
		TableId:  []string{tableId},
		PersonId: personId,
		Feature:  faceFeature,
		Action:   dbAction,
		Enable:   enable,
		CarNo:    carNo,
	}
	b, err := proto.Marshal(&msg)
	logger.Debug("SyncSdkCompareCache proto.err:", err)
	if err != nil {
		return
	}
	logger.Debug("SyncSdkCompareCache analyServerId:", analyServerId)

	PublishSdkCompareCacheChange(b)
}

func doUpdateCompareCacheRequest(b []byte) {
	logger.Debug("syncDbTablePerson:recv doUpdateCompareCacheRequest msg")
	var sock mangos.Socket
	var err error
	if sock, err = req.NewSocket(); err != nil {
		logger.Debug("comp cache can't new req socket:%s", err.Error())
		return
	}
	sock.AddTransport(ipc.NewTransport())
	sock.AddTransport(tcp.NewTransport())
	var url = "tcp://" + config.DbPersonCompare.Ip + ":" + strconv.Itoa(config.DbPersonCompare.Port)
	logger.Debug("doUpdateCompareCacheRequest url:", url)
	if err = sock.Dial(url); err != nil {
		logger.Debug("comp cache can't dial on req socket:%s", err.Error())
		return
	}
	sock.SetOption(mangos.OptionMaxRecvSize, 1024*1024*100)
	sock.SetOption(mangos.OptionRecvDeadline, time.Second*2)
	if err = sock.Send(b); err != nil {
		logger.Debug("comp cache can't send message on push socket:%s", err.Error())
		return
	}
	if msg, err := sock.Recv(); err != nil {
		logger.Debug("comp cache sock.Recv receive err:%s", err.Error())
		return
	} else {
		logger.Debug("comp cache sock.Recv msg:", string(msg))
	}
	sock.Close()
}

func ReInitDbPersonCompareData() {
	logger.Debug("加入集群后重新初始化底库比对数据")
	var sock mangos.Socket
	var err error
	if sock, err = req.NewSocket(); err != nil {
		logger.Debug("ReInitDbPersonCompareData can't new req socket:%s", err.Error())
		return
	}
	sock.AddTransport(ipc.NewTransport())
	sock.AddTransport(tcp.NewTransport())
	var url = "tcp://" + config.DbPersonCompare.Ip + ":" + strconv.Itoa(config.DbPersonCompare.Port)
	if err = sock.Dial(url); err != nil {
		logger.Debug("ReInitDbPersonCompareData can't dial on req socket:%s", err.Error())
		return
	}
	sock.SetOption(mangos.OptionMaxRecvSize, 1024*1024)
	//sock.SetOption(mangos.OptionRecvDeadline, time.Second*10)
	msg := protomsg.CompareEvent{
		EventType: protomsg.CompareEventType_ReInitCache,
		Payload:   []byte(""),
	}
	b, err := proto.Marshal(&msg)
	if err != nil {
		logger.Debug("ReInitDbPersonCompareData msg marshal err", err)
	}
	if err = sock.Send(b); err != nil {
		logger.Debug("ReInitDbPersonCompareData can't send message on push socket:%s", err.Error())
		return
	}
	if msg, err := sock.Recv(); err != nil {
		logger.Debug("ReInitDbPersonCompareData sock.Recv receive err:%s", err.Error())
		return
	} else {
		logger.Debug("ReInitDbPersonCompareData sock.Recv msg:", string(msg))
	}
	sock.Close()
}