jy-data-extract/udps/main.go

package main

import (
	"encoding/json"
	"flag"
	"log"
	mu "mfw/util"
	"net"
	qu "qfw/util"
	"time"

	"gopkg.in/mgo.v2/bson"
)

var udpclient mu.UdpClient //udp对象
var nextNodes []map[string]interface{}

var startDate, endDate, ip, port, stype string

func main() {
	//2015-11-03,2017-04-01
	//2017-04-01,2017-06-01
	//2017-06-01,2018-06-01
	//2018-06-01,2019-02-20
	flag.StringVar(&startDate, "start", "", "开始日期2006-01-02")
	flag.StringVar(&endDate, "end", "", "结束日期2006-01-02")
	flag.StringVar(&ip, "ip", "127.0.0.1", "ip")
	flag.StringVar(&port, "port", "", "dup端口")
	flag.StringVar(&stype, "stype", "", "stype")
	flag.Parse()
	log.Println(startDate, endDate, ip, port, stype)
	start, _ := time.ParseInLocation(qu.Date_Short_Layout, startDate, time.Local)
	end, _ := time.ParseInLocation(qu.Date_Short_Layout, endDate, time.Local)
	sid := bson.NewObjectIdWithTime(start)
	eid := bson.NewObjectIdWithTime(end)
	udpclient = mu.UdpClient{Local: ":1470", BufSize: 1024}
	udpclient.Listen(processUdpMsg)
	by, _ := json.Marshal(map[string]interface{}{
		"gtid":  sid,
		"lteid": eid,
		"stype": stype,
	})
	udpclient.WriteUdp(by, mu.OP_TYPE_DATA, &net.UDPAddr{
		IP:   net.ParseIP(ip),
		Port: qu.IntAll(port),
	})
	b := make(chan bool, 1)
	<-b
}

func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
	switch act {
	case mu.OP_TYPE_DATA:
		var mapInfo map[string]interface{}
		err := json.Unmarshal(data, &mapInfo)
		if err != nil {
			log.Println(err)
		} else {
			log.Println(mapInfo)
		}
	case mu.OP_NOOP: //下个节点回应
		log.Println("发送成功", string(data))
	}
}