中心服务改造：支持自动转推功能

Services/CenterService/Controller/WasuChecker.go

 package Controller
 
 import (
-	log "github.com/sirupsen/logrus"
-	"src/Common/DateTime"
+	cmap "github.com/orcaman/concurrent-map"
 	"src/Common/Unit"
 	"src/Config"
 	"src/Model/Wasu"
@@ -11,43 +10,38 @@ import (
 
 type WasuChecker struct {
 	Config *Config.Config
+	queue cmap.ConcurrentMap
 }
 
 func (this *WasuChecker) Start() {
+	this.queue =cmap.New()
+	//定时遍历SP，检查SP是否启动
 	go func() {
 	START:
-		s, ok := this.Config.Sys.Get("wasu_check_seconds")
+		s, ok := this.Config.Sys.Get("config_sp_check_seconds")
 		if ok {
-			this.checkIncrementList()
-			n := Unit.Time2Seconds(s.(string))
+			this.check()
+			n := Unit.Time2Seconds(s.(string))+10
 			time.Sleep(time.Duration(n) * time.Second)
 		}
 		goto START
 	}()
 }
 
-func (this *WasuChecker) checkIncrementList() {
-	log.Infoln("WasuChecker.checkIncrementList Starting...")
-	mediaModel := new(Wasu.MediaListModel)
-	mediaModel.Config = this.Config
-	date := DateTime.Format("YYYYMMDD", time.Now())
-	//date:="20171213"
-	mediaList := mediaModel.FetchList(date, 1)
-	ok := mediaModel.UpdateMediaList(mediaList)
-	if !ok {
-		log.Errorln("mediaModel.UpdateMediaList err: date=", date, "page=", 1)
-	}
-
-	if mediaList != nil {
-		pageNum := mediaList.PageNum
-		if pageNum >= 2 {
-			for page := 2; page <= pageNum; page++ {
-				mediaList := mediaModel.FetchList(date, page)
-				ok := mediaModel.UpdateMediaList(mediaList)
-				if !ok {
-					log.Errorln("mediaModel.UpdateMediaList err: date=", date, "page=", page)
-				}
-			}
+func (this *WasuChecker) check() {
+	for _, tmp := range this.Config.Sps.Items() {
+		sp := tmp.(map[string]interface{})
+		spid:= sp["id"].(string);
+		if !this.queue.Has(spid) {
+			this.queue.Set(spid, "running")
+			spModel := new(Wasu.SpModel)
+			spModel.Config = this.Config
+			spModel.Spid = spid
+			//st, _ := DateTime.Parse("YYYY-MM-DD hh:mm:ss", sp["push_offset"].(string))
+			//spModel.StartTime = DateTime.Format("YYYYMMDDhhmmss", st)
+			go spModel.Start()
 		}
 	}
 }
+
+

Services/CenterService/Model/Sp/SpModel.go

@@ -16,7 +16,7 @@ type SpModel struct {
 
 func (this *SpModel) Start() {
 	START:
-		log.Infoln("启动 SP检测：",this.Spid," offset:",this.StartTime)
+		log.Infoln("启动SP推送媒资检测：",this.Spid," offset:",this.StartTime)
 		config,ok := this.Config.Sps.Get(this.Spid)
 		n:=60
 		if ok {

Services/CenterService/Model/Wasu/MediaListModel.go

@@ -28,6 +28,8 @@ import (
 **/
 
 type Media struct {
+	Spid string `json:"spid"`
+	PushMode string `json:"push_mode"` //是否自动转推：0手动，1自动
 	JsonUrl string `json:"jsonUrl"`
 	//Md5			string 	`json:"md5"`
 	//NewsId		string	`json:"newsId"`
@@ -45,10 +47,24 @@ type MediaList struct {
 
 type MediaListModel struct {
 	Config *Config.Config
+	Spid string
 }
 
 func (this *MediaListModel) FetchList(date string, page int) *MediaList {
-	wasu_api_incrementList, ok := this.Config.Sys.Get("wasu_api_incrementList")
+	//wasu_api_incrementList, ok := this.Config.Sys.Get("wasu_api_incrementList")
+	wasu_api_incrementList := ""
+	config,ok := this.Config.Sps.Get(this.Spid)
+	if ok {
+		conf,ok:=config.(map[string]interface{})
+		if ok {
+			wasu_api_incrementList=conf["wasu_api_incrementList"].(string)
+		}
+	}
+
+	if wasu_api_incrementList == "" {
+		logrus.Error("PULL SP ：",this.Spid," 缺少配置项：wasu_api_incrementList")
+	}
+
 	mediaList := new(MediaList)
 	if ok {
 		url := fmt.Sprintf("%s?date=%s&page=%d", wasu_api_incrementList, date, page)
@@ -78,6 +94,24 @@ func (this *MediaListModel) UpdateMediaList(mediaList *MediaList) bool {
 }
 
 func (this *MediaListModel) UpdateMedia(media *Media) bool {
+	media.Spid = this.Spid
+	wasu_incrementList_push_mode:="";
+	config,ok := this.Config.Sps.Get(this.Spid)
+	if ok {
+		conf,ok:=config.(map[string]interface{})
+		if ok {
+			wasu_incrementList_push_mode=conf["wasu_incrementList_push_mode"].(string)
+		}
+	}
+
+	if wasu_incrementList_push_mode == "" {
+		logrus.Error("PULL SP ：",this.Spid," 缺少配置项：wasu_incrementList_push_mode")
+		wasu_incrementList_push_mode="0"
+	}
+
+	media.PushMode=wasu_incrementList_push_mode;
+
+
 	center_api_media_update, ok := this.Config.Sys.Get("center_api_media_update")
 	b := false
 	if ok {

Services/CenterService/Model/Wasu/SpModel.go

+package Wasu
+
+import (
+	log "github.com/sirupsen/logrus"
+	"src/Common/DateTime"
+	"src/Common/Unit"
+	"src/Config"
+	"time"
+)
+
+type SpModel struct {
+	Config *Config.Config
+	Spid string
+}
+
+func (this *SpModel) Start() {
+	START:
+		log.Infoln("启动SP拉取增量媒资检测：",this.Spid)
+		config,ok := this.Config.Sps.Get(this.Spid)
+		n:=3600
+		if ok {
+			conf,ok:=config.(map[string]interface{})
+			if ok {
+				wasu_check_seconds:=conf["wasu_check_seconds"].(string)
+				if wasu_check_seconds !="" {
+					n = Unit.Time2Seconds(wasu_check_seconds)
+				}
+				this.pull();
+			}
+
+		}
+		time.Sleep(time.Duration(n) * time.Second)
+		goto START
+}
+
+func (this *SpModel) pull() {
+	log.Infoln("SpModel.pull Starting...")
+	mediaModel := new(MediaListModel)
+	mediaModel.Config = this.Config
+	mediaModel.Spid = this.Spid
+	date := DateTime.Format("YYYYMMDD", time.Now())
+	//date:="20190627"
+	mediaList := mediaModel.FetchList(date, 1)
+	ok := mediaModel.UpdateMediaList(mediaList)
+	if !ok {
+		log.Errorln("mediaModel.UpdateMediaList err: date=", date, "page=", 1)
+	}
+
+	if mediaList != nil {
+		pageNum := mediaList.PageNum
+		if pageNum >= 2 {
+			for page := 2; page <= pageNum; page++ {
+				mediaList := mediaModel.FetchList(date, page)
+				ok := mediaModel.UpdateMediaList(mediaList)
+				if !ok {
+					log.Errorln("mediaModel.UpdateMediaList err: date=", date, "page=", page)
+				}
+			}
+		}
+	}
+}
\ No newline at end of file