package pubsub

import (
	"basic.com/valib/gopherdiscovery.git"
	"encoding/json"
	"fmt"
	"time"
)

type mangosPubSub struct {
	url          string
	heartBeatUrl string
	pubCh        chan Message  //publish msg chan
	surveyors    gopherdiscovery.StringSet

	svr *gopherdiscovery.DiscoveryServer
	cli *gopherdiscovery.DiscoveryClient

	cliInfo map[string][]byte

	recvCh chan Message  //recv msg chan
}


func newPub(publishUrl string,heartBeatUrl string) (*mangosPubSub,error) {
	var discoveryServer *gopherdiscovery.DiscoveryServer
	var err error
	var (
		defaultOpts = gopherdiscovery.Options{
			SurveyTime:   1500 * time.Millisecond,
			PollTime:     2 * time.Second,
		}
	)

	discoveryServer, err = gopherdiscovery.Server(heartBeatUrl, publishUrl, defaultOpts)

	fmt.Println("newPub err:",err)
	pub := &mangosPubSub{
		url:          publishUrl,
		heartBeatUrl: heartBeatUrl,
		surveyors:    gopherdiscovery.NewStringSet(),
		pubCh:        make(chan Message, 50),
		svr: discoveryServer,
	}
	var msgCache = make(map[string]Message)
	//clientMsgCh := make(map[string]chan Message)
	go func() {
		for {
			select {
			case msg := <-pub.pubCh:
				msgCache[msg.Topic] = msg
				//if len(clientMsgCh) > 0 {
				//	for _, ch := range clientMsgCh {
				//		ch <- msg
				//	}
				//}
				fmt.Println("<-pub.pubCh,pub.surveyors.Len:",pub.surveyors.Cardinality())
				if pub.surveyors.Cardinality() >0 {
					sendB, _ := json.Marshal(msg)
					discoveryServer.PublishMsg(string(sendB))
				}
			default:
				nodeIds := discoveryServer.AliveNodes()

				//for _,nodeId := range nodeIds {
				//	if _,ok := clientMsgCh[nodeId]; !ok {
				//		clientMsgCh[nodeId] = make(chan Message)
				//	}
				//}
				removedNodes := pub.surveyors.Difference(nodeIds)
				addedNodes := nodeIds.Difference(pub.surveyors)

				if addedNodes.Cardinality() >0 { //有新节点上线的时候，需要发一次消息，节点离线的时候不用管
					fmt.Println("removedNodes:", removedNodes, "addedNodes:", addedNodes)
					if len(msgCache) > 0 {
						for _,cMsg := range msgCache {
							sendB, _ := json.Marshal(cMsg)
							discoveryServer.PublishMsg(string(sendB))
						}
					}
				}

				pub.surveyors = nodeIds
				pub.cliInfo = discoveryServer.SvInfo()
				time.Sleep(10 * time.Millisecond)
			}
		}
	}()

	return pub,nil
}

const (
	msgTopicAll = "sub-msg-all-topic"
)

func newSub(subcribeUrl string,heartBeatUrl string, topics []string,procId string) (*mangosPubSub,error) {
	client, err := gopherdiscovery.ClientWithSub(heartBeatUrl, subcribeUrl, procId)
	if err !=nil {
		return nil,err
	}
	heartMsg := client.HeartBeatMsg()
	_= <-heartMsg
	fmt.Println("heat beat with server success")
	sub := &mangosPubSub{
		url:subcribeUrl,
		heartBeatUrl: heartBeatUrl,
		recvCh: make(chan Message,50),
		surveyors: gopherdiscovery.NewStringSet(),
		cli: client,
	}
	var receivedCache = make(map[string]string)
	go func() {
		peers, _ := client.Peers()
		for msg := range peers {
			//判断是否是想要的主题消息
			var recvMsg Message
			if err := json.Unmarshal(msg, &recvMsg);err ==nil {
				if b,matchedTopic := matchTopic(&recvMsg, topics);b {
					if lastMsgId,ok := receivedCache[matchedTopic];ok {
						if lastMsgId != recvMsg.Id {
							receivedCache[matchedTopic] = recvMsg.Id
							sub.recvCh <- recvMsg
						}
					} else {
						receivedCache[matchedTopic] = recvMsg.Id
						sub.recvCh <- recvMsg
					}
				}
			}
		}
	}()

	return sub,nil
}

func matchTopic(msg *Message,subTopics []string) (bool,string) {
	if subTopics ==nil || len(subTopics) ==0 {
		return true,msgTopicAll
	}
	for _,t := range subTopics {
		if msg.Topic == t {
			return true,msg.Topic
		}
	}
	return false,""
}

func (ps *mangosPubSub) Surveyor() []string {
	return ps.surveyors.ToSlice()
}

func (ps *mangosPubSub) Publish(msg Message) {
	ps.pubCh <- msg
}

func (ps *mangosPubSub) Recv() chan Message {
	return ps.recvCh
}

func (ps *mangosPubSub) GetCliInfo() map[string][]byte {
	if ps.svr != nil {
		return ps.cliInfo
	}
	return nil
}

func (ps *mangosPubSub) SetResp(r []byte) {
	if ps.cli != nil {
		ps.cli.SetResp(r)
	}
}