mirror of
https://github.com/redis/go-redis.git
synced 2025-07-28 06:42:00 +03:00
Support Monitor Command (#2830)
* Add monitor command * Add monitor commadn and tests * insure goroutine shutdown * fix data race * linting * change timeout explanation --------- Co-authored-by: Chayim <chayim@users.noreply.github.com>
This commit is contained in:
83
command.go
83
command.go
@ -8,6 +8,7 @@ import (
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9/internal"
|
||||
@ -5381,3 +5382,85 @@ func (cmd *InfoCmd) Item(section, key string) string {
|
||||
return cmd.val[section][key]
|
||||
}
|
||||
}
|
||||
|
||||
type MonitorStatus int
|
||||
|
||||
const (
|
||||
monitorStatusIdle MonitorStatus = iota
|
||||
monitorStatusStart
|
||||
monitorStatusStop
|
||||
)
|
||||
|
||||
type MonitorCmd struct {
|
||||
baseCmd
|
||||
ch chan string
|
||||
status MonitorStatus
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func newMonitorCmd(ctx context.Context, ch chan string) *MonitorCmd {
|
||||
return &MonitorCmd{
|
||||
baseCmd: baseCmd{
|
||||
ctx: ctx,
|
||||
args: []interface{}{"monitor"},
|
||||
},
|
||||
ch: ch,
|
||||
status: monitorStatusIdle,
|
||||
mu: sync.Mutex{},
|
||||
}
|
||||
}
|
||||
|
||||
func (cmd *MonitorCmd) String() string {
|
||||
return cmdString(cmd, nil)
|
||||
}
|
||||
|
||||
func (cmd *MonitorCmd) readReply(rd *proto.Reader) error {
|
||||
ctx, cancel := context.WithCancel(cmd.ctx)
|
||||
go func(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
err := cmd.readMonitor(rd, cancel)
|
||||
if err != nil {
|
||||
cmd.err = err
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}(ctx)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cmd *MonitorCmd) readMonitor(rd *proto.Reader, cancel context.CancelFunc) error {
|
||||
for {
|
||||
cmd.mu.Lock()
|
||||
st := cmd.status
|
||||
cmd.mu.Unlock()
|
||||
if pk, _ := rd.Peek(1); len(pk) != 0 && st == monitorStatusStart {
|
||||
line, err := rd.ReadString()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cmd.ch <- line
|
||||
}
|
||||
if st == monitorStatusStop {
|
||||
cancel()
|
||||
break
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cmd *MonitorCmd) Start() {
|
||||
cmd.mu.Lock()
|
||||
defer cmd.mu.Unlock()
|
||||
cmd.status = monitorStatusStart
|
||||
}
|
||||
|
||||
func (cmd *MonitorCmd) Stop() {
|
||||
cmd.mu.Lock()
|
||||
defer cmd.mu.Unlock()
|
||||
cmd.status = monitorStatusStop
|
||||
}
|
||||
|
Reference in New Issue
Block a user