parent
e9e0ee0772
commit
151828f569
|
@ -29,6 +29,12 @@ func main() {
|
|||
time.Sleep(3 * time.Second) // 等待应用mqtt连接
|
||||
for i := 0; i < 4; i++ {
|
||||
ac.startIotQcService() // 启动IOT驱采服务
|
||||
resp, err := ac.iotLogReq(&dto.IotServiceLogReq{Count: 10})
|
||||
if err != nil {
|
||||
slog.Error("应用请求日志错误", "err", err)
|
||||
} else {
|
||||
slog.Info("应用请求日志成功", "resp", resp)
|
||||
}
|
||||
time.Sleep(10 * time.Second)
|
||||
ac.stopIotQcService() // 停止IOT驱采服务
|
||||
time.Sleep(5 * time.Second)
|
||||
|
@ -189,6 +195,28 @@ func (app *AppClient) iotStopReq(req *dto.IotQcServiceStopReq) (*dto.IotQcServic
|
|||
return result, nil
|
||||
}
|
||||
|
||||
func (app *AppClient) iotLogReq(req *dto.IotServiceLogReq) (*dto.IotServiceLogResp, error) {
|
||||
h, err := app.newRpcHandler("%s/iotlogresp")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
|
||||
defer cancel()
|
||||
b, err := proto.Marshal(req)
|
||||
if err != nil {
|
||||
return nil, errors.Join(fmt.Errorf("序列化IOT服务日志请求消息失败"), err)
|
||||
}
|
||||
resp, err := h.Request(ctx, &paho.Publish{
|
||||
Topic: mqtt.GetLogReqTopic(),
|
||||
Payload: b,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errors.Join(fmt.Errorf("发送IOT服务日志请求错误"), err)
|
||||
}
|
||||
result := &dto.IotServiceLogResp{}
|
||||
proto.Unmarshal(resp.Payload, result)
|
||||
return result, nil
|
||||
}
|
||||
func (app *AppClient) RegIotCjDataHandler(h func(*dto.IotCj)) {
|
||||
app.cfg.Router.RegisterHandler(mqtt.GetCjTopic(), func(p *paho.Publish) {
|
||||
cj := &dto.IotCj{}
|
||||
|
|
|
@ -0,0 +1,87 @@
|
|||
package log
|
||||
|
||||
import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"sync"
|
||||
)
|
||||
|
||||
var historyList *logHistoryList
|
||||
|
||||
func GetTailsOf(num int) []string {
|
||||
if historyList == nil {
|
||||
return nil
|
||||
}
|
||||
return historyList.GetTailsOf(num)
|
||||
}
|
||||
|
||||
// 初始化IOT日志配置和历史纪录
|
||||
func InitLog() {
|
||||
level := slog.LevelInfo
|
||||
historyList = newLogHistoryList(1000) // 最多保存1000条(根据情况调整)
|
||||
w := newIotLogWriter(historyList)
|
||||
slog.SetDefault(slog.New(slog.NewJSONHandler(w, &slog.HandlerOptions{
|
||||
Level: level,
|
||||
AddSource: false,
|
||||
})))
|
||||
}
|
||||
|
||||
type iotLogWriter struct {
|
||||
io.Writer
|
||||
logs *logHistoryList
|
||||
}
|
||||
|
||||
// 输出到Stdout
|
||||
func newIotLogWriter(logs *logHistoryList) *iotLogWriter {
|
||||
return &iotLogWriter{Writer: os.Stdout, logs: logs}
|
||||
}
|
||||
|
||||
func (w *iotLogWriter) Write(p []byte) (n int, err error) {
|
||||
w.logs.Add(string(p))
|
||||
return w.Writer.Write(p)
|
||||
}
|
||||
|
||||
type logHistoryList struct {
|
||||
logs []string
|
||||
capacity int
|
||||
mutex sync.Mutex
|
||||
}
|
||||
|
||||
func newLogHistoryList(capacity int) *logHistoryList {
|
||||
return &logHistoryList{
|
||||
logs: make([]string, capacity),
|
||||
capacity: capacity,
|
||||
}
|
||||
}
|
||||
|
||||
func (cl *logHistoryList) Add(item string) {
|
||||
cl.mutex.Lock()
|
||||
defer cl.mutex.Unlock()
|
||||
|
||||
if len(cl.logs) >= cl.capacity {
|
||||
// 如果容量已满,删除最早的元素
|
||||
cl.logs = cl.logs[1:]
|
||||
}
|
||||
cl.logs = append(cl.logs, item)
|
||||
}
|
||||
|
||||
func (cl *logHistoryList) Len() int {
|
||||
cl.mutex.Lock()
|
||||
defer cl.mutex.Unlock()
|
||||
|
||||
return len(cl.logs)
|
||||
}
|
||||
|
||||
// 获取最新的num条记录
|
||||
func (cl *logHistoryList) GetTailsOf(num int) []string {
|
||||
cl.mutex.Lock()
|
||||
defer cl.mutex.Unlock()
|
||||
|
||||
end := len(cl.logs)
|
||||
start := end - num
|
||||
if start < 0 {
|
||||
start = 0
|
||||
}
|
||||
return cl.logs[start:end]
|
||||
}
|
|
@ -0,0 +1,14 @@
|
|||
package log
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestGetLogHistory(t *testing.T) {
|
||||
list := newLogHistoryList(10)
|
||||
for i := 0; i < 10; i++ {
|
||||
list.Add("test" + strconv.Itoa(i))
|
||||
}
|
||||
t.Log(list.GetTailsOf(20))
|
||||
}
|
8
main.go
8
main.go
|
@ -1,11 +1,9 @@
|
|||
package main
|
||||
|
||||
import "joylink.club/iot/server"
|
||||
import (
|
||||
"joylink.club/iot/server"
|
||||
)
|
||||
|
||||
func main() {
|
||||
// slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{
|
||||
// Level: slog.LevelDebug,
|
||||
// AddSource: false,
|
||||
// })))
|
||||
server.StartIotQcServer()
|
||||
}
|
||||
|
|
|
@ -2,9 +2,11 @@ package server
|
|||
|
||||
import (
|
||||
"joylink.club/iot/dto"
|
||||
"joylink.club/iot/log"
|
||||
)
|
||||
|
||||
// 获取IOT日志请求响应
|
||||
func GetIotLog(req *dto.IotServiceLogReq) *dto.IotServiceLogResp {
|
||||
return &dto.IotServiceLogResp{}
|
||||
tails := log.GetTailsOf(int(req.Count))
|
||||
return &dto.IotServiceLogResp{Logs: tails}
|
||||
}
|
||||
|
|
|
@ -8,6 +8,7 @@ import (
|
|||
|
||||
"joylink.club/iot/config"
|
||||
"joylink.club/iot/dto"
|
||||
"joylink.club/iot/log"
|
||||
"joylink.club/iot/mqtt"
|
||||
"joylink.club/iot/service"
|
||||
)
|
||||
|
@ -130,6 +131,7 @@ func (s *IotQcServer) stopIotQcMappingService(req *dto.IotQcServiceStopReq) *dto
|
|||
}
|
||||
|
||||
func StartIotQcServer() {
|
||||
log.InitLog()
|
||||
iqcs = &IotQcServer{
|
||||
tasks: []service.IScheduledTask{},
|
||||
state: &dto.IotServiceState{
|
||||
|
|
Loading…
Reference in New Issue