mysteel_watch/services/index_queue.go

package services

import (
	"eta/mysteel_watch/cache"
	"fmt"
	"time"
)

// the service for log
//func AutoRefresh() {
//	sub := global.Redis.Subscribe(context.TODO(), "autoRefresh")
//
//	defer func() {
//		sub.Close()
//		if err := recover(); err != nil {
//			fmt.Println("[AutoRefresh]", err)
//		}
//	}()
//	for {
//		msg, err := sub.ReceiveMessage(context.TODO())
//		if err != nil {
//			fmt.Println("sub err:" + err.Error())
//		}
//		fmt.Println("sub:", msg.Payload)
//		IndexHandle(msg.Payload)
//
//		//global.Rc.Brpop(utils.REFRESH_INDEX, func(b []byte) {
//		//	filePath := string(b)
//		//	fmt.Println("filePath:", filePath)
//		//	IndexHandle(filePath)
//		//})
//	}
//}

// AutoRefresh 调用python刷新指标
func AutoRefresh() {
	for {
		el := cache.RefreshList.Front()
		// 如果没取到，那么就睡眠1s
		if el == nil {
			time.Sleep(1 * time.Second)
			continue
		}
		filePath := el.Value.(string)
		IndexHandle(filePath)

		// 处理完后就移除该list
		cache.RefreshList.Remove(el)
		cache.FilePathMutex.Lock()
		delete(cache.FilePathMap, filePath)
		cache.FilePathMutex.Unlock()
	}
}

// IndexHandle
// @Description: 指标处理
// @author: Roc
// @datetime 2023-11-27 09:41:59
// @param filePath string
func IndexHandle(filePath string) {
	//filePath = strings.Replace(filePath, `"`, ``, -1)
	fmt.Println("开始刷新文件:", filePath)
	time.Sleep(1 * time.Second)
	MysteelChemicalRefresh(filePath)
}