package extract

import (
	"encoding/json"
	"fmt"
	db "jy/mongodbutil"
	ju "jy/util"
	mu "mfw/util"
	"net"
	qu "qfw/util"
	"sync"

	log "github.com/donnie4w/go-logger/logger"
	"gopkg.in/mgo.v2/bson"
)

var ClearUdpclient mu.UdpClient //清理udp对象

//udp通知清理
func ClearUdp() {
	nextNodes = qu.ObjArrToMapArr(ju.Config["nextNode"].([]interface{}))
	ClearUdpclient = mu.UdpClient{Local: ":" + qu.ObjToString(ju.Config["clearudpport"]), BufSize: 1024}
	ClearUdpclient.Listen(clearProcessUdpMsg)
}

func clearProcessUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
	switch act {
	case mu.OP_TYPE_DATA:
		var rep map[string]interface{}
		err := json.Unmarshal(data, &rep)
		if err != nil {
			log.Debug(err)
			Udpclient.WriteUdp([]byte("false"), mu.OP_NOOP, ra) //回应上一个节点
		} else {
			sid, _ := rep["gtid"].(string)
			eid, _ := rep["lteid"].(string)
			//stype, _ := rep["stype"].(string)
			log.Debug("======", sid, eid)
			go Udpclient.WriteUdp([]byte("ok"), mu.OP_NOOP, ra) //回应上一个节点
			ClearByUdp(sid, eid)
			log.Debug("udp通知清理完成,eid=", eid)
		}
	case mu.OP_NOOP: //下个节点回应
		var rep map[string]interface{}
		err := json.Unmarshal(data, &rep)
		if err != nil {
			log.Debug(err)
		} else {
			log.Debug(rep)
		}
	}
}

//根据id区间清理
func ClearByUdp(sid, eid string) {
	clt := &ClearTask{}
	clt.Id = qu.ObjToString(ju.Config["cleartaskid"])
	//加载任务信息
	clt.InitClearTaskInfo()
	clt.ClearTaskInfo.FDB = db.MgoFactory(2, 3, 120, clt.ClearTaskInfo.FromDbAddr, clt.ClearTaskInfo.FromDB)
	//初始化脚本信息
	clt.InitClearLuas()
	//更新结果表清理后的字段值
	go clt.UpdateResultVal(true)

	query := bson.M{"_id": bson.M{"$gte": bson.ObjectIdHex(sid), "$lte": bson.ObjectIdHex(eid)}}
	log.Debug("query---", query, clt.ClearTaskInfo.FromColl)
	count := clt.ClearTaskInfo.FDB.Count(clt.ClearTaskInfo.FromColl, query)
	log.Debug("count---", count)
	pageNum := (count + PageSize - 1) / PageSize
	limit := PageSize
	if count < PageSize {
		limit = count
	}
	//一次查询5000条数据进行清理
	index := 0
	wg := sync.WaitGroup{}
	for i := 0; i < pageNum; i++ {
		query = bson.M{"_id": bson.M{"$gte": bson.ObjectIdHex(sid)}}
		fmt.Printf("page=%d,query=%v", i+1, query)
		list, _ := clt.ClearTaskInfo.FDB.Find(clt.ClearTaskInfo.FromColl, query, nil, Fields2, false, 0, limit)
		for _, v := range *list {
			_id := qu.BsonIdToSId(v["_id"])
			clt.ClearTaskInfo.ProcessPool <- true
			wg.Add(1)
			go func(val map[string]interface{}) {
				defer wg.Done()
				clt.ClearProcess(&val)
			}(v)
			//sid = qu.BsonIdToSId(v["_id"])
			index++
			if index%1000 == 0 {
				log.Debug("index：", index, ",页码：", i+1, ",_id：", _id)
			}
			sid = _id
			if sid >= eid {
				break
			}
		}
	}
	wg.Wait()
	clt.UpdateResultVal(false)
}
