package softbus

import (
	"context"
	"fmt"
	"time"

	"github.com/golang/protobuf/proto"
)

const (
	// RegKey fixed key for hb to servermanager
	RegKey = 12
	// GetTopicInfoTypeTopic topic
	GetTopicInfoTypeTopic = "gettopic"
	// GetTopicInfoTypeChannel channel
	GetTopicInfoTypeChannel = "getchannel"
)

type subOReply struct {
	sock *DgramSocket
	info *ProcInfo
}

type sockRe struct {
	sock *DgramSocket
	peer int
}

// TransInfo 传输的数据和必要的记录
type TransInfo struct {
	info *MsgInfo
	port int
}

// Handle handler
type Handle struct {
	ctx context.Context
	// 创建channel对应的reply,等待读取其中的内容,server
	// 其中必须有一个作为Request函数的server
	m map[string]*subOReply
	// 创建心跳连接,client,仅发送心跳信息
	sockHB *sockRe
	// 创建更新主题连接,client,仅发送主题更新信息
	sockUp *sockRe
	// 创建订阅的socket
	sockSub *subOReply
	// 创建reply服务Request函数
	sockRep *sockRe
	// 创建一个万能socket发送给任意server
	sockWorker *sockRe

	chSub   chan TransInfo
	chReply chan TransInfo
}

func garbageCollect(ctx context.Context, h *Handle) {

	<-ctx.Done()

	for _, v := range h.m {
		v.sock.Close()
	}
	h.sockHB.sock.Close()
	h.sockUp.sock.Close()
	h.sockSub.sock.Close()
	h.sockRep.sock.Close()
	h.sockWorker.sock.Close()
}

func recvRoutine(ctx context.Context, sock *DgramSocket, ch chan<- TransInfo) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			if data, peer, err := sock.RecvFrom(); err == nil {
				var info *MsgInfo
				if err := proto.Unmarshal(data, info); err == nil {
					ch <- TransInfo{
						info: info,
						port: peer,
					}
				}
			}
		}
	}
}

// Register reg
func Register(ctx context.Context, info *RegisterInfo) *Handle {
	m := make(map[string]*subOReply)

	// 首先请求一堆key
	sockReg := OpenDgramSocket()
	if sockReg == nil {
		return nil
	}
	defer sockReg.Close()

	var msg, rdata []byte
	var err error
loop:
	for {
		select {
		case <-ctx.Done():
			return nil
		default:

			if msg == nil {
				if msg, err = proto.Marshal(info); err != nil {
					time.Sleep(100 * time.Millisecond)
					continue
				}
			}
			if rdata, err = sockReg.SendAndRecv(msg, RegKey); err == nil {
				break loop
			}
		}
	}

	// 得到key,赋值
	var regReply RegisterInfoReply
	if err := proto.Unmarshal(rdata, &regReply); err != nil {
		return nil
	}

	// 收发req/rep channel
	for _, v := range info.Channel {
		if k, ok := regReply.ChannelKey[v]; ok {
			s := OpenDgramSocket()
			s.Bind(int(k))
			m[v] = &subOReply{
				sock: s,
				info: info.ProcInfo,
			}
		}
	}

	chSize := 5
	chSub := make(chan TransInfo, chSize)
	chReply := make(chan TransInfo, chSize)

	// heartbeat使用一个socket
	sockHB := OpenDgramSocket()
	hbr := &sockRe{
		sock: sockHB,
		peer: int(regReply.HeartbeatKey),
	}
	// 更新主题使用一个
	sockUp := OpenDgramSocket()
	upr := &sockRe{
		sock: sockUp,
		peer: int(regReply.UpdateTopicKey),
	}

	// sub使用一个socket
	sockSub := OpenDgramSocket()
	sockSub.Bind(int(regReply.SubTopicKey))
	// 启动接收线程
	go recvRoutine(ctx, sockSub, chSub)
	sub := &subOReply{
		sock: sockSub,
		info: info.ProcInfo,
	}
	// reply使用一个,服务Request
	sockReply := OpenDgramSocket()
	sockReply.Bind(int(regReply.ReplyKey))
	// 启动接收线程
	go recvRoutine(ctx, sockReply, chSub)
	rer := &sockRe{
		sock: sockReply,
		peer: -1,
	}

	// 万能socket,仅作为客户端使用
	sockW := OpenDgramSocket()
	swr := &sockRe{
		sock: sockW,
		peer: -1,
	}
	handle := &Handle{
		ctx:        ctx,
		m:          m,
		sockHB:     hbr,
		sockUp:     upr,
		sockSub:    sub,
		sockRep:    rer,
		sockWorker: swr,
		chSub:      chSub,
		chReply:    chReply,
	}

	go garbageCollect(ctx, handle)

	return handle
}

// GetTopicInfo get topic info
func (h *Handle) GetTopicInfo(topic, typ string) int {
	if v, ok := h.m[topic]; ok {
		return v.sock.Port()
	}
	return -1
}

func (h *Handle) send2(sr *sockRe, data []byte, logID string) error {
	if r := sr.sock.SendTo(data, sr.peer); r != 0 {
		return fmt.Errorf("%s SendTo Failed: %d", logID, r)
	}
	return nil
}

// HeartBeat hb
func (h *Handle) HeartBeat(info *HeartbeatInfo) error {
	msg, err := proto.Marshal(info)
	if err == nil {
		return h.send2(h.sockHB, msg, "HeartBeat")
	}
	return err
}

// SendOnly no recv
func (h *Handle) SendOnly(key int, info *MsgInfo) error {
	msg, err := proto.Marshal(info)
	if err == nil {
		if r := h.sockWorker.sock.SendTo(msg, key); r != 0 {
			return fmt.Errorf("SendOnly Failed: %d", r)
		}
	}
	return err
}

// Pub func
func (h *Handle) Pub(info *MsgInfo) error {
	msg, err := proto.Marshal(info)
	if err == nil {
		return h.send2(h.sockUp, msg, "Pub")
	}
	return err
}

// Request req sync
func (h *Handle) Request(key int, info *MsgInfo) *MsgInfo {
	msg, err := proto.Marshal(info)
	if err != nil {
		return nil
	}

	// 同步接口,需要等待返回值
	var ret *MsgInfo
loop:
	for {
		select {
		case <-h.ctx.Done():
			return nil
		default:
			if data, err := h.sockWorker.sock.SendAndRecv(msg, key); err == nil {
				if err := proto.Unmarshal(data, ret); err == nil {
					break loop
				}
			}
		}
	}
	return ret
}

// Reply request
func (h *Handle) Reply(key int, info *MsgInfo) error {
	msg, err := proto.Marshal(info)
	if err == nil {
		if r := h.sockRep.sock.SendTo(msg, key); r != 0 {
			return fmt.Errorf("Reply Failed: %d", r)
		}
	}
	return err
}

// GetMesg get mesg for sub or reply
func (h *Handle) GetMesg() (subMsg *MsgInfo, replyMsg *MsgInfo, replyKey int) {
	if len(h.chSub) > 1 {
		m := <-h.chSub
		subMsg = m.info
	}

	if len(h.chReply) > 1 {
		m := <-h.chReply
		replyMsg = m.info
		replyKey = m.port
	}
	return
}