Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
92cd652b0d |
58
internal/glob/glob.go
Normal file
58
internal/glob/glob.go
Normal file
@@ -0,0 +1,58 @@
|
|||||||
|
// @tag glob
|
||||||
|
package glob
|
||||||
|
|
||||||
|
// Match checks if a string matches a glob-style pattern.
|
||||||
|
// Supports:
|
||||||
|
// - * (asterisk): matches any sequence of characters (including empty)
|
||||||
|
// - ? (question): matches any single character
|
||||||
|
//
|
||||||
|
// This matches Redis PUBSUB CHANNELS pattern behavior.
|
||||||
|
func Match(pattern, s string) bool {
|
||||||
|
if pattern == "" {
|
||||||
|
return s == ""
|
||||||
|
}
|
||||||
|
if pattern == "*" {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
// Simple glob matching using dynamic programming
|
||||||
|
// Convert pattern to regex-like matching
|
||||||
|
patternLen := len(pattern)
|
||||||
|
sLen := len(s)
|
||||||
|
|
||||||
|
// dp[i][j] = true if pattern[0:i] matches s[0:j]
|
||||||
|
dp := make([][]bool, patternLen+1)
|
||||||
|
for i := range dp {
|
||||||
|
dp[i] = make([]bool, sLen+1)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Empty pattern matches empty string
|
||||||
|
dp[0][0] = true
|
||||||
|
|
||||||
|
// Handle leading * in pattern
|
||||||
|
for i := 1; i <= patternLen; i++ {
|
||||||
|
if pattern[i-1] == '*' {
|
||||||
|
dp[i][0] = dp[i-1][0]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Fill the DP table
|
||||||
|
for i := 1; i <= patternLen; i++ {
|
||||||
|
for j := 1; j <= sLen; j++ {
|
||||||
|
switch pattern[i-1] {
|
||||||
|
case '*':
|
||||||
|
// * matches: either consume nothing (stay at same position in pattern)
|
||||||
|
// or consume one character from string
|
||||||
|
dp[i][j] = dp[i-1][j] || dp[i][j-1]
|
||||||
|
case '?':
|
||||||
|
// ? matches any single character
|
||||||
|
dp[i][j] = dp[i-1][j-1]
|
||||||
|
default:
|
||||||
|
// Literal character must match exactly
|
||||||
|
dp[i][j] = dp[i-1][j-1] && pattern[i-1] == s[j-1]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return dp[patternLen][sLen]
|
||||||
|
}
|
||||||
69
internal/glob/glob_test.go
Normal file
69
internal/glob/glob_test.go
Normal file
@@ -0,0 +1,69 @@
|
|||||||
|
// @tag glob
|
||||||
|
// @tag test
|
||||||
|
package glob
|
||||||
|
|
||||||
|
import "testing"
|
||||||
|
|
||||||
|
func TestMatch(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
pattern string
|
||||||
|
s string
|
||||||
|
want bool
|
||||||
|
}{
|
||||||
|
{"", "", true},
|
||||||
|
{"", "foo", false},
|
||||||
|
{"*", "", true},
|
||||||
|
{"*", "foo", true},
|
||||||
|
{"*", "foo/bar", true},
|
||||||
|
{"foo", "foo", true},
|
||||||
|
{"foo", "bar", false},
|
||||||
|
{"foo*", "foo", true},
|
||||||
|
{"foo*", "foobar", true},
|
||||||
|
{"foo*", "fo", false},
|
||||||
|
{"*foo", "foo", true},
|
||||||
|
{"*foo", "barfoo", true},
|
||||||
|
{"*foo", "bar", false},
|
||||||
|
{"foo*bar", "foobar", true},
|
||||||
|
{"foo*bar", "foobazbar", true},
|
||||||
|
{"foo*bar", "foobaz", false},
|
||||||
|
{"foo?bar", "fooXbar", true},
|
||||||
|
{"foo?bar", "fooXXbar", false},
|
||||||
|
{"foo??bar", "fooXXbar", true},
|
||||||
|
{"?", "a", true},
|
||||||
|
{"?", "", false},
|
||||||
|
{"a?c", "abc", true},
|
||||||
|
{"a?c", "ac", false},
|
||||||
|
{"foo?*bar", "fooabcbar", true},
|
||||||
|
{"foo?*bar", "foobar", false},
|
||||||
|
{"*foo*bar*", "ahellofoocherrybarb", true},
|
||||||
|
{"ch*", "channel", true},
|
||||||
|
{"ch*", "ch", true},
|
||||||
|
{"ch*", "c", false},
|
||||||
|
{"*", "*", true},
|
||||||
|
{"**", "*", true},
|
||||||
|
{"*a*b", "aXb", true},
|
||||||
|
{"a*b*c", "aXbYc", true},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.pattern+"/"+tt.s, func(t *testing.T) {
|
||||||
|
if got := Match(tt.pattern, tt.s); got != tt.want {
|
||||||
|
t.Errorf("Match(%q, %q) = %v, want %v", tt.pattern, tt.s, got, tt.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func BenchmarkMatch(b *testing.B) {
|
||||||
|
patterns := []string{"*", "ch*", "foo*bar", "???", "*channel*"}
|
||||||
|
strings := []string{"channel", "channel123", "foobar", "abc", "somechannelname"}
|
||||||
|
|
||||||
|
b.ReportAllocs()
|
||||||
|
for i := 0; i < b.N; i++ {
|
||||||
|
for _, p := range patterns {
|
||||||
|
for _, s := range strings {
|
||||||
|
Match(p, s)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -31,6 +31,7 @@ type Client struct {
|
|||||||
conn net.Conn
|
conn net.Conn
|
||||||
name string // Connection name set via CLIENT SETNAME
|
name string // Connection name set via CLIENT SETNAME
|
||||||
replyMode ReplyMode // Reply mode (ON/OFF/SKIP)
|
replyMode ReplyMode // Reply mode (ON/OFF/SKIP)
|
||||||
|
selectedDB int // Selected database index (via SELECT command)
|
||||||
flags uint64 // Client flags (e.g., paused, no-touch)
|
flags uint64 // Client flags (e.g., paused, no-touch)
|
||||||
created time.Time // When the client connected
|
created time.Time // When the client connected
|
||||||
lastActive time.Time // Last activity time
|
lastActive time.Time // Last activity time
|
||||||
@@ -48,6 +49,7 @@ func NewClient(conn net.Conn) *Client {
|
|||||||
conn: conn,
|
conn: conn,
|
||||||
name: "",
|
name: "",
|
||||||
replyMode: ReplyOn,
|
replyMode: ReplyOn,
|
||||||
|
selectedDB: 0,
|
||||||
flags: 0,
|
flags: 0,
|
||||||
created: now,
|
created: now,
|
||||||
lastActive: now,
|
lastActive: now,
|
||||||
@@ -89,6 +91,27 @@ func (c *Client) GetReplyMode() ReplyMode {
|
|||||||
return c.replyMode
|
return c.replyMode
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SelectedDB returns the currently selected database index
|
||||||
|
func (c *Client) SelectedDB() int {
|
||||||
|
c.mu.RLock()
|
||||||
|
defer c.mu.RUnlock()
|
||||||
|
return c.selectedDB
|
||||||
|
}
|
||||||
|
|
||||||
|
// CreatedAt returns when the client connected
|
||||||
|
func (c *Client) CreatedAt() time.Time {
|
||||||
|
c.mu.RLock()
|
||||||
|
defer c.mu.RUnlock()
|
||||||
|
return c.created
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetSelectedDB sets the selected database index
|
||||||
|
func (c *Client) SetSelectedDB(db int) {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
c.selectedDB = db
|
||||||
|
}
|
||||||
|
|
||||||
// UpdateActivity updates the last activity timestamp
|
// UpdateActivity updates the last activity timestamp
|
||||||
func (c *Client) UpdateActivity() {
|
func (c *Client) UpdateActivity() {
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
@@ -212,7 +235,7 @@ func (c *Client) LocalAddr() string {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// InfoString returns CLIENT INFO formatted string (Redis format)
|
// InfoString returns CLIENT INFO formatted string (Redis format)
|
||||||
// Format: id= addr= laddr= fd= age= idle= flags= db= sub= psub= multi= qbuf= qbuf-free= obl= oll= omem= tot-mem= cmd= user= redir=
|
// Format: id= addr= laddr= fd= age= idle= flags= db= sub= psub= multi= qbuf= qbuf-free= obl= oll= omem= tot-mem= events= cmd= user= redir= resp=
|
||||||
func (c *Client) InfoString() string {
|
func (c *Client) InfoString() string {
|
||||||
c.mu.RLock()
|
c.mu.RLock()
|
||||||
defer c.mu.RUnlock()
|
defer c.mu.RUnlock()
|
||||||
@@ -229,16 +252,21 @@ func (c *Client) InfoString() string {
|
|||||||
flags += "S"
|
flags += "S"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Events - indicate connection is active for read and write
|
||||||
|
events := "rw"
|
||||||
|
|
||||||
// Format: Redis CLIENT INFO format
|
// Format: Redis CLIENT INFO format
|
||||||
return fmt.Sprintf("id=%d addr=%s laddr=%s fd=-1 age=%d idle=%d flags=%s db=0 sub=%d psub=%d multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 tot-mem=0 cmd=client user=default redir=-1",
|
return fmt.Sprintf("id=%d addr=%s laddr=%s fd=-1 age=%d idle=%d flags=%s db=%d sub=%d psub=%d multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 tot-mem=0 events=%s cmd=client user=default redir=-1 resp=2",
|
||||||
c.id,
|
c.id,
|
||||||
c.RemoteAddr(),
|
c.RemoteAddr(),
|
||||||
c.LocalAddr(),
|
c.LocalAddr(),
|
||||||
age,
|
age,
|
||||||
idle,
|
idle,
|
||||||
flags,
|
flags,
|
||||||
|
c.selectedDB,
|
||||||
len(c.subscribed),
|
len(c.subscribed),
|
||||||
len(c.patterns),
|
len(c.patterns),
|
||||||
|
events,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,7 +5,9 @@ package server
|
|||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/eleksir/pekoe/internal/glob"
|
||||||
"github.com/eleksir/pekoe/internal/logger"
|
"github.com/eleksir/pekoe/internal/logger"
|
||||||
"github.com/eleksir/pekoe/internal/protocol"
|
"github.com/eleksir/pekoe/internal/protocol"
|
||||||
"github.com/eleksir/pekoe/internal/pubsub"
|
"github.com/eleksir/pekoe/internal/pubsub"
|
||||||
@@ -132,12 +134,24 @@ func (s *Server) handlePubsub(args [][]byte, writer *ConnWriter) error {
|
|||||||
pattern = string(args[1])
|
pattern = string(args[1])
|
||||||
}
|
}
|
||||||
channels := s.channelList(pattern)
|
channels := s.channelList(pattern)
|
||||||
result := make([]protocol.Value, len(channels))
|
result := make([]protocol.Value, 0, len(channels))
|
||||||
for i, ch := range channels {
|
for _, ch := range channels {
|
||||||
result[i] = protocol.Bulk(ch)
|
if pattern == "" || glob.Match(pattern, ch) {
|
||||||
|
result = append(result, protocol.Bulk(ch))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
writer.WriteArray(result)
|
writer.WriteArray(result)
|
||||||
|
|
||||||
|
case "shardchannels":
|
||||||
|
// PUBSUB SHARDCHANNELS [pattern] - List active sharded channels
|
||||||
|
// Pekoe doesn't support sharded pubsub, return empty list
|
||||||
|
writer.WriteArray(nil)
|
||||||
|
|
||||||
|
case "shardnumsub":
|
||||||
|
// PUBSUB SHARDNUMSUB [shardchannel ...] - Get subscriber count for sharded channels
|
||||||
|
// Pekoe doesn't support sharded pubsub, return empty array
|
||||||
|
writer.WriteArray(nil)
|
||||||
|
|
||||||
case "numpat":
|
case "numpat":
|
||||||
// PUBSUB NUMPAT - Count of unique pattern subscriptions
|
// PUBSUB NUMPAT - Count of unique pattern subscriptions
|
||||||
writer.WriteInt(int64(s.hub.PatternCount()))
|
writer.WriteInt(int64(s.hub.PatternCount()))
|
||||||
@@ -147,7 +161,9 @@ func (s *Server) handlePubsub(args [][]byte, writer *ConnWriter) error {
|
|||||||
helpText := []string{
|
helpText := []string{
|
||||||
"PUBSUB <subcommand> [<argument> [<argument> ...]]>",
|
"PUBSUB <subcommand> [<argument> [<argument> ...]]>",
|
||||||
" PUBSUB CHANNELS [pattern] - List active channels (optionally filtered by pattern)",
|
" PUBSUB CHANNELS [pattern] - List active channels (optionally filtered by pattern)",
|
||||||
|
" PUBSUB SHARDCHANNELS [pattern] - List active sharded channels (not supported)",
|
||||||
" PUBSUB NUMSUB [channel ...] - Get subscriber count for channels",
|
" PUBSUB NUMSUB [channel ...] - Get subscriber count for channels",
|
||||||
|
" PUBSUB SHARDNUMSUB [shardchannel ...] - Get subscriber count for sharded channels (not supported)",
|
||||||
" PUBSUB NUMPAT - Get the number of pattern subscriptions",
|
" PUBSUB NUMPAT - Get the number of pattern subscriptions",
|
||||||
" PUBSUB HELP - Show this help message",
|
" PUBSUB HELP - Show this help message",
|
||||||
}
|
}
|
||||||
@@ -349,11 +365,15 @@ func (s *Server) buildClientList(args [][]byte) string {
|
|||||||
func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) (int, error) {
|
func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) (int, error) {
|
||||||
clients := s.hub.GetAllClients()
|
clients := s.hub.GetAllClients()
|
||||||
killed := 0
|
killed := 0
|
||||||
|
var notFoundErr error
|
||||||
|
|
||||||
// Parse arguments
|
// Parse arguments
|
||||||
var killID uint64
|
var killID uint64
|
||||||
|
var killIDMin uint64
|
||||||
|
var killIDMax uint64
|
||||||
var killAddr string
|
var killAddr string
|
||||||
var skipMe = true
|
var skipMe = true
|
||||||
|
var maxAgeMs int64
|
||||||
|
|
||||||
for i := 0; i < len(args); i++ {
|
for i := 0; i < len(args); i++ {
|
||||||
arg := strings.ToUpper(string(args[i]))
|
arg := strings.ToUpper(string(args[i]))
|
||||||
@@ -361,6 +381,12 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) (
|
|||||||
case "ID":
|
case "ID":
|
||||||
if i+2 < len(args) {
|
if i+2 < len(args) {
|
||||||
// ID min max format - kill all IDs in range
|
// ID min max format - kill all IDs in range
|
||||||
|
if _, err := fmt.Sscanf(string(args[i+1]), "%d", &killIDMin); err != nil {
|
||||||
|
return 0, fmt.Errorf("invalid client ID: %s", args[i+1])
|
||||||
|
}
|
||||||
|
if _, err := fmt.Sscanf(string(args[i+2]), "%d", &killIDMax); err != nil {
|
||||||
|
return 0, fmt.Errorf("invalid client ID: %s", args[i+2])
|
||||||
|
}
|
||||||
i += 2
|
i += 2
|
||||||
} else if i+1 < len(args) {
|
} else if i+1 < len(args) {
|
||||||
// Single ID
|
// Single ID
|
||||||
@@ -374,14 +400,35 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) (
|
|||||||
killAddr = string(args[i+1])
|
killAddr = string(args[i+1])
|
||||||
i++
|
i++
|
||||||
}
|
}
|
||||||
|
case "LADDR":
|
||||||
|
// LADDR not fully supported, just skip
|
||||||
|
if i+1 < len(args) {
|
||||||
|
i++
|
||||||
|
}
|
||||||
case "SKIPME":
|
case "SKIPME":
|
||||||
if i+1 < len(args) {
|
if i+1 < len(args) {
|
||||||
skipMe = strings.ToUpper(string(args[i+1])) == "YES"
|
skipMe = strings.ToUpper(string(args[i+1])) == "YES"
|
||||||
i++
|
i++
|
||||||
}
|
}
|
||||||
|
case "MAXAGE":
|
||||||
|
if i+1 < len(args) {
|
||||||
|
if _, err := fmt.Sscanf(string(args[i+1]), "%d", &maxAgeMs); err != nil {
|
||||||
|
return 0, fmt.Errorf("invalid maxage: %s", args[i+1])
|
||||||
|
}
|
||||||
|
if maxAgeMs <= 0 {
|
||||||
|
return 0, fmt.Errorf("maxage must be greater than 0")
|
||||||
|
}
|
||||||
|
i++
|
||||||
|
}
|
||||||
|
case "TYPE":
|
||||||
|
// TYPE filter handled separately in handleClient
|
||||||
|
if i+1 < len(args) {
|
||||||
|
i++
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
now := time.Now()
|
||||||
for _, c := range clients {
|
for _, c := range clients {
|
||||||
// Skip current client if requested
|
// Skip current client if requested
|
||||||
if skipMe && c.ID() == currentClient.ID() {
|
if skipMe && c.ID() == currentClient.ID() {
|
||||||
@@ -390,11 +437,29 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) (
|
|||||||
|
|
||||||
shouldKill := false
|
shouldKill := false
|
||||||
|
|
||||||
if killID > 0 && c.ID() == killID {
|
// Check ID (single or range)
|
||||||
|
if killID > 0 {
|
||||||
|
shouldKill = c.ID() == killID
|
||||||
|
} else if killIDMin > 0 || killIDMax > 0 {
|
||||||
|
id := c.ID()
|
||||||
|
if killIDMin > 0 && id < killIDMin {
|
||||||
|
shouldKill = false
|
||||||
|
} else if killIDMax > 0 && id > killIDMax {
|
||||||
|
shouldKill = false
|
||||||
|
} else {
|
||||||
shouldKill = true
|
shouldKill = true
|
||||||
}
|
}
|
||||||
if killAddr != "" && c.RemoteAddr() == killAddr {
|
}
|
||||||
shouldKill = true
|
|
||||||
|
// Check ADDR
|
||||||
|
if !shouldKill && killAddr != "" {
|
||||||
|
shouldKill = c.RemoteAddr() == killAddr
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check MAXAGE
|
||||||
|
if shouldKill && maxAgeMs > 0 {
|
||||||
|
age := now.Sub(c.CreatedAt()).Milliseconds()
|
||||||
|
shouldKill = age >= maxAgeMs
|
||||||
}
|
}
|
||||||
|
|
||||||
if shouldKill {
|
if shouldKill {
|
||||||
@@ -406,7 +471,39 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) (
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return killed, nil
|
// For single ID/ADDR that wasn't found, Redis returns error
|
||||||
|
if killed == 0 && (killID > 0 || killAddr != "") {
|
||||||
|
return 0, fmt.Errorf("No such client")
|
||||||
|
}
|
||||||
|
|
||||||
|
return killed, notFoundErr
|
||||||
|
}
|
||||||
|
|
||||||
|
// handleSelect processes SELECT command.
|
||||||
|
// SELECT db - Switch to a different database (0-15).
|
||||||
|
// For pekoe (pubsub-only), this only tracks the selected DB for client compatibility.
|
||||||
|
// Redis default max DB is 16 (0-15), configurable via max-databases in config.
|
||||||
|
func (s *Server) handleSelect(client *pubsub.Client, args [][]byte, writer *ConnWriter) error {
|
||||||
|
if len(args) != 1 {
|
||||||
|
writer.WriteError("ERR wrong number of arguments for 'select' command")
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
dbIndexStr := string(args[0])
|
||||||
|
var dbIndex int
|
||||||
|
if _, err := fmt.Sscanf(dbIndexStr, "%d", &dbIndex); err != nil {
|
||||||
|
writer.WriteError("ERR invalid DB index")
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if dbIndex < 0 || dbIndex >= MaxDatabases {
|
||||||
|
writer.WriteError("ERR DB index is out of range")
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
client.SetSelectedDB(dbIndex)
|
||||||
|
writer.WriteStatus("OK")
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// handlePsubscribe processes PSUBSCRIBE command.
|
// handlePsubscribe processes PSUBSCRIBE command.
|
||||||
|
|||||||
@@ -21,6 +21,8 @@ import (
|
|||||||
"github.com/eleksir/pekoe/internal/pubsub"
|
"github.com/eleksir/pekoe/internal/pubsub"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
const MaxDatabases = 16 // Redis default maximum databases (0-15)
|
||||||
|
|
||||||
// Sentinel errors for connection handling
|
// Sentinel errors for connection handling
|
||||||
var (
|
var (
|
||||||
ErrQuit = errors.New("client quit")
|
ErrQuit = errors.New("client quit")
|
||||||
@@ -276,6 +278,8 @@ func (s *Server) processCommand(client *pubsub.Client, cmd *Command, writer *Con
|
|||||||
return s.handleQuit(client, writer)
|
return s.handleQuit(client, writer)
|
||||||
case "client":
|
case "client":
|
||||||
return s.handleClient(client, cmd.Args[1:], writer)
|
return s.handleClient(client, cmd.Args[1:], writer)
|
||||||
|
case "select":
|
||||||
|
return s.handleSelect(client, cmd.Args[1:], writer)
|
||||||
default:
|
default:
|
||||||
// Check if client is in pubsub mode - only pubsub commands allowed
|
// Check if client is in pubsub mode - only pubsub commands allowed
|
||||||
if client != nil && (client.SubsCount() > 0 || client.PatternSubsCount() > 0) {
|
if client != nil && (client.SubsCount() > 0 || client.PatternSubsCount() > 0) {
|
||||||
|
|||||||
Reference in New Issue
Block a user