refactor(handlers/monitor): 仅在有订阅用户时才计算系统状态

go.mod

@@ -4,13 +4,14 @@ go 1.22.0
 
 require (
 	github.com/golang-jwt/jwt/v5 v5.2.1
-	github.com/issue9/assert/v4 v4.2.0
+	github.com/issue9/assert/v4 v4.3.0
 	github.com/issue9/cache v0.12.0
+	github.com/issue9/events v0.9.0
 	github.com/issue9/logs/v7 v7.6.0
 	github.com/issue9/mux/v8 v8.1.0
 	github.com/issue9/rands/v3 v3.0.1
 	github.com/issue9/unique/v2 v2.1.0
-	github.com/issue9/web v0.92.0
+	github.com/issue9/web v0.92.1
 	github.com/shirou/gopsutil/v3 v3.24.3
 	golang.org/x/text v0.14.0
 )

go.sum

@@ -11,8 +11,8 @@ github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
-github.com/issue9/assert/v4 v4.2.0 h1:XJGMFYW0xfESqFRPLWbSsr0xWdkofytvQbDfNb5n9fw=
-github.com/issue9/assert/v4 v4.2.0/go.mod h1:v7qDRXi7AsaZZNh8eAK2rkLJg5/clztqQGA1DRv9Lv4=
+github.com/issue9/assert/v4 v4.3.0 h1:W3XDKmttsfzihYGxJ9rJoL2ViJgWERB9IxfHcxjv65U=
+github.com/issue9/assert/v4 v4.3.0/go.mod h1:v7qDRXi7AsaZZNh8eAK2rkLJg5/clztqQGA1DRv9Lv4=
 github.com/issue9/cache v0.12.0 h1:NiDBuN9x22H4UJsOMDoEuIFA8r3qNqPqO9vyzzcvzoY=
 github.com/issue9/cache v0.12.0/go.mod h1:0s9j7qiKv4uWYqz0D2N2H7bIBvmtD+903h5GqnxW6i4=
 github.com/issue9/config v0.6.2 h1:znXvsk6gh0wm+fTEn0zUjjramKuOLY8Jt0ZTxp4GIkc=
@@ -21,6 +21,8 @@ github.com/issue9/conv v1.3.5 h1:UWeA+Zqp5vjNDLrmhLhaXPjS1hL8gh4quX6Shk5njKQ=
 github.com/issue9/conv v1.3.5/go.mod h1:lkZYMyrmxy+HK//N4eLmwUl9mCwj0zgeTjnxgl/w7hM=
 github.com/issue9/errwrap v0.3.2 h1:7KEme9Pfe75M+sIMcPCn/DV90wjnOcRbO4DXVAHj3Fw=
 github.com/issue9/errwrap v0.3.2/go.mod h1:KcCLuUGiffjooLCUjL89r1cyO8/HT/VRcQrneO53N3A=
+github.com/issue9/events v0.9.0 h1:P1uoLkEeZ1EcV9kTGzhJStglkJ1rNGzFYNhacWg4WS8=
+github.com/issue9/events v0.9.0/go.mod h1:7rdiy2zeMKDsgejo0JxYsIXj5fVCVUxX/l/ABmOqfiE=
 github.com/issue9/localeutil v0.26.5 h1:e78b6cOOtgzfb4g4U9uPLC8QyK6Lux+s7ZiQe+6iM1A=
 github.com/issue9/localeutil v0.26.5/go.mod h1:BJXJwcAT9CyyVZOlqfmq+B5FcPbqGxGjYnTYbVuiMM8=
 github.com/issue9/logs/v7 v7.6.0 h1:dvY1ctPROdd2YaOwYRNOkfbmMx+8OM0w53t8bWrWg9s=
@@ -41,8 +43,8 @@ github.com/issue9/term/v3 v3.2.8 h1:vAtsr9FLwrDQRyU7S8AOUI3f+QrYsXVHT/kee9MuMkk=
 github.com/issue9/term/v3 v3.2.8/go.mod h1:S/xLgjEXJNr7C6UhcGTEOV5k94mzHRcYMmtvhLJoX6A=
 github.com/issue9/unique/v2 v2.1.0 h1:lE9hstenrbBgDM/Iulv7d75qAYpgoDV8rSRoLmEBHOk=
 github.com/issue9/unique/v2 v2.1.0/go.mod h1:qZoDKnfu+7Q0yxhifVseRKD2Wea9Tc9zdXwALnFc54A=
-github.com/issue9/web v0.92.0 h1:yheONHWMXXf3/qdbhKiRfUfSJIHK13zsnkSVd5pfQoM=
-github.com/issue9/web v0.92.0/go.mod h1:IIVT/CPr/uyZcYj+RVR2yk7qMIvHk7LDycEQU5VlABo=
+github.com/issue9/web v0.92.1 h1:FcxfncJiK60BW9KR9QNgx8+klKEIAEY90NOusz5Rf1g=
+github.com/issue9/web v0.92.1/go.mod h1:IIVT/CPr/uyZcYj+RVR2yk7qMIvHk7LDycEQU5VlABo=
 github.com/jellydator/ttlcache/v3 v3.2.0 h1:6lqVJ8X3ZaUwvzENqPAobDsXNExfUJd61u++uW8a3LE=
 github.com/jellydator/ttlcache/v3 v3.2.0/go.mod h1:hi7MGFdMAwZna5n2tuvh63DvFLzVKySzCVW6+0gA2n4=
 github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU=

handlers/monitor/monitor.go

@@ -10,49 +10,66 @@ import (
 	"encoding/json"
 	"time"
 
+	"github.com/issue9/events"
 	"github.com/issue9/web"
 	"github.com/issue9/web/mimetype/sse"
 )
 
 // Monitor 系统状态检测
 type Monitor struct {
-	dur time.Duration
-	s   *sse.Server[int]
+	dur    time.Duration
+	server *sse.Server[string]
+	event  *events.Event[*Stats]
+	enable bool
 }
 
 // New 声明 [Monitor]
 //
 // dur 为发送状态数据的时间间隔；
 func New(s web.Server, dur time.Duration) *Monitor {
-	return &Monitor{
-		dur: dur,
-		s:   sse.NewServer[int](s, 0, 0, 10),
+	m := &Monitor{
+		dur:    dur,
+		server: sse.NewServer[string](s, 0, 0, 10),
+		event:  events.New[*Stats](),
+		enable: true,
 	}
+
+	s.Services().AddTicker(web.Phrase("monitor system stat"), func(now time.Time) error {
+		if !m.enable {
+			return nil
+		}
+
+		stats, err := calcState(dur, now)
+		if err != nil {
+			return err
+		}
+
+		m.event.Publish(true, stats)
+		return nil
+	}, dur, true, false)
+
+	return m
 }
 
 // Handle 输出监视信息
 //
 // NOTE: 这是一个 SSE 连接，需要保证 content-type 的正确性。
 func (m *Monitor) Handle(ctx *web.Context) web.Responser {
-	source, wait := m.s.NewSource(0, ctx)
-	event := source.NewEvent("monitor", json.Marshal)
+	source, wait := m.server.NewSource(ctx.Server().UniqueID(), ctx) // 只要保证是唯一 ID 就行
 	var cancel context.CancelFunc
 
 	defer func() {
 		wait()
-
-		if cancel != nil { // 退出时删除 ticker 事件
-			cancel()
-		}
+		cancel() // 在 wait 之后
+		m.enable = m.event.Len() > 0
 	}()
 
-	cancel = ctx.Server().Services().AddTicker(web.Phrase("monitor system stats"), func(now time.Time) error {
-		stats, err := calcState(m.dur, now)
-		if err != nil {
-			return err
+	event := source.NewEvent("monitor", json.Marshal)
+	cancel = m.event.Subscribe(func(data *Stats) {
+		if err := event.Sent(data); err != nil {
+			ctx.Logs().ERROR().Error(err)
 		}
-		return event.Sent(stats)
-	}, m.dur, true, false)
+	})
 
 	return nil
 }

handlers/monitor/monitor_test.go

@@ -39,13 +39,12 @@ func TestMonitor(t *testing.T) {
 	r.Get("/stats", m.Handle)
 
 	stats := make(chan *sse.Message, 10)
-	req, err := http.NewRequest(http.MethodGet, "http://localhost:8080/stats", nil)
-	a.NotError(err).NotNil(req)
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
 	defer cancel()
-	err = sse.OnMessage(ctx, s.Logs().ERROR(), req, nil, stats)
-	a.NotError(err)
-
-	a.Contains((<-stats).Data[0], `"cpu":`).
-		Contains((<-stats).Data[0], `"mem":`)
+	err = sse.OnMessage(ctx, s.Logs().ERROR(), "http://localhost:8080/stats", nil, stats)
+	a.When(err == nil, func(a *assert.Assertion) {
+		s := <-stats
+		a.Contains(s.Data[0], `"cpu":`).
+			Contains(s.Data[0], `"mem":`)
+	}, "返回了错误信息 %v", err)
 }

handlers/monitor/types.go

@@ -17,20 +17,22 @@ import (
 
 type Stats struct {
 	XMLName struct{}  `json:"-" yaml:"-" xml:"stats"`
-	OS      *Info     `json:"os" yaml:"os" xml:"os"`                // 系统级别的状态信息
-	Process *Info     `json:"process" yaml:"process" xml:"process"` // 当前进程的状态信息
+	OS      *OS       `json:"os" yaml:"os" xml:"os"`                // 系统级别的状态信息
+	Process *Process  `json:"process" yaml:"process" xml:"process"` // 当前进程的状态信息
 	Created time.Time `json:"created" yaml:"created" xml:"created"` // 此条记录的创建时间
 }
 
-type Info struct {
-	CPU float64 `json:"cpu" yaml:"cpu" xml:"cpu"` // CPU 使用百分比
-	Mem uint64  `json:"mem" yaml:"mem" xml:"mem"` // 内存使用量，以 byte 为单位。
-
-	// 网络相关数据
-	Net *Net `json:"net,omitempty" yaml:"net,omitempty" xml:"net,omitempty"`
+type OS struct {
+	CPU float64 `json:"cpu" yaml:"cpu" xml:"cpu"`                               // CPU 使用百分比
+	Mem uint64  `json:"mem" yaml:"mem" xml:"mem"`                               // 内存使用量，以 byte 为单位。
+	Net *Net    `json:"net,omitempty" yaml:"net,omitempty" xml:"net,omitempty"` // 网络相关数据
+}
 
-	// 在全局模式之下为空
-	Goroutines int `json:"goroutines,omitempty" yaml:"goroutines,omitempty" xml:"goroutines,omitempty"`
+type Process struct {
+	CPU        float64 `json:"cpu" yaml:"cpu" xml:"cpu"`    // CPU 使用百分比
+	Mem        uint64  `json:"mem" yaml:"mem" xml:"mem"`    // 内存使用量，以 byte 为单位。
+	Conn       int     `json:"conn" yaml:"conn" xml:"conn"` // 连接数量
+	Goroutines int     `json:"goroutines,omitempty" yaml:"goroutines,omitempty" xml:"goroutines,omitempty"`
 }
 
 type Net struct {
@@ -53,7 +55,7 @@ func calcState(interval time.Duration, now time.Time) (*Stats, error) {
 	return &Stats{OS: all, Process: p, Created: now}, nil
 }
 
-func calcProcess() (*Info, error) {
+func calcProcess() (*Process, error) {
 	p, err := process.NewProcess(int32(os.Getpid()))
 	if err != nil {
 		return nil, err
@@ -74,24 +76,15 @@ func calcProcess() (*Info, error) {
 		return nil, err
 	}
 
-	netIO, err := p.IOCounters()
-	if err != nil {
-		return nil, err
-	}
-
-	return &Info{
-		CPU: cpus,
-		Mem: mems.RSS,
-		Net: &Net{
-			Conn: len(conns),
-			Sent: netIO.WriteBytes,
-			Recv: netIO.ReadBytes,
-		},
+	return &Process{
+		CPU:        cpus,
+		Mem:        mems.RSS,
+		Conn:       len(conns),
 		Goroutines: runtime.NumGoroutine(),
 	}, nil
 }
 
-func calcOS(interval time.Duration) (*Info, error) {
+func calcOS(interval time.Duration) (*OS, error) {
 	cpus, err := cpu.Percent(interval, false)
 	if err != nil {
 		return nil, err
@@ -112,7 +105,7 @@ func calcOS(interval time.Duration) (*Info, error) {
 		return nil, err
 	}
 
-	return &Info{
+	return &OS{
 		CPU: cpus[0],
 		Mem: mems.Used,
 		Net: &Net{