qixiaoning
2025-07-08 fe724b50b3f1b3dfe2219eb9af4bcca96c89a158
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
package main
 
import (
    "basic.com/valib/bhomeclient.git"
    "basic.com/valib/bhomedbapi.git"
    "basic.com/valib/logger.git"
    "basic.com/valib/version.git"
    "context"
    "flag"
    "fmt"
    "os"
    "os/signal"
    "syscall"
    "vamicro/config"
    "vamicro/report-service/controllers"
    "vamicro/report-service/models"
    "vamicro/report-service/service"
    "vamicro/report-service/tasks"
)
 
var (
    procName = service.ProcName
    proc     = &bhomeclient.ProcInfo{
        Name: procName, //进程名称
        ID:   procName, //进程id
        Info: "",       //进程的描述信息,用于区分同一进程名称下多个进程
    }
    env = flag.String("e", "pro", "")
 
    dbPath string = "../config/report-service.db"
)
 
func init() {
    flag.Parse()
    vaversion.Usage()
 
    config.Init(*env)
    // 日志初始化
    var logFile = config.LogConf.Path + "vamicro-" + procName + ".log"
    logger.InitLogger(logFile, config.LogConf.Level, config.LogConf.MaxSize, config.LogConf.MaxBackups, config.LogConf.MaxAge)
    logger.Info("log init success !")
}
 
func main() {
    //防止多进程运行
    lockFile := "/tmp/run/report-service.lock"
    var lock *os.File
    lock, err := os.Create(lockFile)
    if err != nil {
        fmt.Println("create lock file error", err)
        os.Exit(1)
    }
 
    defer os.Remove(lockFile)
    defer lock.Close()
 
    err = syscall.Flock(int(lock.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
    if err != nil {
        fmt.Println("report service is running ")
        os.Exit(1)
    }
 
    defer syscall.Flock(int(lock.Fd()), syscall.LOCK_UN)
 
    var filterResultRecvChan = make(chan []byte)
 
    ctx, cancel := context.WithCancel(context.Background())
    models.Init(dbPath)
    defer models.CloseDB()
 
    fm, pubTopics := initFuncMap()
    var reg = &bhomeclient.RegisterInfo{
        Proc:     *proc,
        Channel:  nil,
        PubTopic: pubTopics,
        SubTopic: []string{"pub-topic-ruleProcess-filter-results"},
    }
 
    q := make(chan os.Signal, 1)
    signal.Notify(q, os.Interrupt, os.Kill, syscall.SIGTERM)
 
    ms, err := bhomeclient.NewMicroNode(ctx, q, config.Server.AnalyServerId, reg, logger.Debug)
    if err != nil {
        return
    }
 
    bhomedbapi.InitLog(logger.Debug)
    bhomedbapi.InitGetNetNode(ms.GetLocalNetNodeByTopic)
    bhomedbapi.InitDoReq(ms.RequestOnly)
 
    go ms.StartServer(fm)
 
    // 订阅报警数据
    go subMessage(ms, filterResultRecvChan)
 
    // 启动上报任务
    go tasks.Start(filterResultRecvChan)
 
    <-q
 
    fmt.Println("ctrl c")
    _ = ms.DeRegister()
    cancel()
    ms.Free()
}
 
const urlPrefix = "/data/api-v"
 
func initFuncMap() (map[string]bhomeclient.MicroFunc, []string) {
    funcMap := make(map[string]bhomeclient.MicroFunc)
 
    controller := new(controllers.ReportServiceController)
 
    funcMap[urlPrefix+"/report/getDevInfo"] = controller.GetDevInfo
    funcMap[urlPrefix+"/report/setDevInfo"] = controller.SetDevInfo
    funcMap[urlPrefix+"/report/getInterval"] = controller.GetIntervalConfig
    funcMap[urlPrefix+"/report/setInterval"] = controller.SetIntervalConfig
 
    var pubTopics []string
    for key, _ := range funcMap {
        pubTopics = append(pubTopics, key)
    }
 
    return funcMap, pubTopics
}
 
func subMessage(client *bhomeclient.MicroNode, MsgChan chan []byte) {
    for {
        select {
        case subMsg := <-client.SubCh:
            if subMsg != nil {
                logger.Debug("recive msg topic:", string(subMsg.Topic))
                MsgChan <- subMsg.Data
            }
        }
    }
}