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
|
||||
name string // Connection name set via CLIENT SETNAME
|
||||
replyMode ReplyMode // Reply mode (ON/OFF/SKIP)
|
||||
selectedDB int // Selected database index (via SELECT command)
|
||||
flags uint64 // Client flags (e.g., paused, no-touch)
|
||||
created time.Time // When the client connected
|
||||
lastActive time.Time // Last activity time
|
||||
@@ -48,6 +49,7 @@ func NewClient(conn net.Conn) *Client {
|
||||
conn: conn,
|
||||
name: "",
|
||||
replyMode: ReplyOn,
|
||||
selectedDB: 0,
|
||||
flags: 0,
|
||||
created: now,
|
||||
lastActive: now,
|
||||
@@ -89,6 +91,27 @@ func (c *Client) GetReplyMode() 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
|
||||
func (c *Client) UpdateActivity() {
|
||||
c.mu.Lock()
|
||||
@@ -212,7 +235,7 @@ func (c *Client) LocalAddr() string {
|
||||
}
|
||||
|
||||
// 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 {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
@@ -229,16 +252,21 @@ func (c *Client) InfoString() string {
|
||||
flags += "S"
|
||||
}
|
||||
|
||||
// Events - indicate connection is active for read and write
|
||||
events := "rw"
|
||||
|
||||
// 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.RemoteAddr(),
|
||||
c.LocalAddr(),
|
||||
age,
|
||||
idle,
|
||||
flags,
|
||||
c.selectedDB,
|
||||
len(c.subscribed),
|
||||
len(c.patterns),
|
||||
events,
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -5,7 +5,9 @@ package server
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/eleksir/pekoe/internal/glob"
|
||||
"github.com/eleksir/pekoe/internal/logger"
|
||||
"github.com/eleksir/pekoe/internal/protocol"
|
||||
"github.com/eleksir/pekoe/internal/pubsub"
|
||||
@@ -132,12 +134,24 @@ func (s *Server) handlePubsub(args [][]byte, writer *ConnWriter) error {
|
||||
pattern = string(args[1])
|
||||
}
|
||||
channels := s.channelList(pattern)
|
||||
result := make([]protocol.Value, len(channels))
|
||||
for i, ch := range channels {
|
||||
result[i] = protocol.Bulk(ch)
|
||||
result := make([]protocol.Value, 0, len(channels))
|
||||
for _, ch := range channels {
|
||||
if pattern == "" || glob.Match(pattern, ch) {
|
||||
result = append(result, protocol.Bulk(ch))
|
||||
}
|
||||
}
|
||||
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":
|
||||
// PUBSUB NUMPAT - Count of unique pattern subscriptions
|
||||
writer.WriteInt(int64(s.hub.PatternCount()))
|
||||
@@ -147,7 +161,9 @@ func (s *Server) handlePubsub(args [][]byte, writer *ConnWriter) error {
|
||||
helpText := []string{
|
||||
"PUBSUB <subcommand> [<argument> [<argument> ...]]>",
|
||||
" 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 SHARDNUMSUB [shardchannel ...] - Get subscriber count for sharded channels (not supported)",
|
||||
" PUBSUB NUMPAT - Get the number of pattern subscriptions",
|
||||
" 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) {
|
||||
clients := s.hub.GetAllClients()
|
||||
killed := 0
|
||||
var notFoundErr error
|
||||
|
||||
// Parse arguments
|
||||
var killID uint64
|
||||
var killIDMin uint64
|
||||
var killIDMax uint64
|
||||
var killAddr string
|
||||
var skipMe = true
|
||||
var maxAgeMs int64
|
||||
|
||||
for i := 0; i < len(args); i++ {
|
||||
arg := strings.ToUpper(string(args[i]))
|
||||
@@ -361,6 +381,12 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) (
|
||||
case "ID":
|
||||
if i+2 < len(args) {
|
||||
// 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
|
||||
} else if i+1 < len(args) {
|
||||
// Single ID
|
||||
@@ -374,14 +400,35 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) (
|
||||
killAddr = string(args[i+1])
|
||||
i++
|
||||
}
|
||||
case "LADDR":
|
||||
// LADDR not fully supported, just skip
|
||||
if i+1 < len(args) {
|
||||
i++
|
||||
}
|
||||
case "SKIPME":
|
||||
if i+1 < len(args) {
|
||||
skipMe = strings.ToUpper(string(args[i+1])) == "YES"
|
||||
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 {
|
||||
// Skip current client if requested
|
||||
if skipMe && c.ID() == currentClient.ID() {
|
||||
@@ -390,11 +437,29 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) (
|
||||
|
||||
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
|
||||
}
|
||||
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 {
|
||||
@@ -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.
|
||||
|
||||
@@ -21,6 +21,8 @@ import (
|
||||
"github.com/eleksir/pekoe/internal/pubsub"
|
||||
)
|
||||
|
||||
const MaxDatabases = 16 // Redis default maximum databases (0-15)
|
||||
|
||||
// Sentinel errors for connection handling
|
||||
var (
|
||||
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)
|
||||
case "client":
|
||||
return s.handleClient(client, cmd.Args[1:], writer)
|
||||
case "select":
|
||||
return s.handleSelect(client, cmd.Args[1:], writer)
|
||||
default:
|
||||
// Check if client is in pubsub mode - only pubsub commands allowed
|
||||
if client != nil && (client.SubsCount() > 0 || client.PatternSubsCount() > 0) {
|
||||
|
||||
Reference in New Issue
Block a user