This commit is contained in:
Bel LaPointe
2026-02-03 20:37:00 -07:00
parent 057c5d0d25
commit 5db6a4360e
2 changed files with 41 additions and 17 deletions

View File

@@ -23,15 +23,30 @@ func adapt(ctx context.Context, config Config, conn net.Conn) error {
if err == io.EOF { if err == io.EOF {
return true, nil return true, nil
} }
return false, err return true, err
} }
if len(message) > 0 { if len(message) > 0 {
hashKey := message[max(0, len(message)-1)].(string) hashKey := message[max(0, len(message)-1)].(string)
hash := int(crc32.ChecksumIEEE([]byte(hashKey))) hash := int(crc32.ChecksumIEEE([]byte(hashKey)))
forward := config.forwards[hash%len(config.forwards)] forward := config.forwards[hash%len(config.forwards)]
result := forward.Do(ctx, message...) forwardCon := forward.Get()
_ = forward if forwardCon == nil {
return true, io.EOF
}
forwardConn := forwardCon.(net.Conn)
if _, err := forwardConn.Write(raw); err != nil {
return true, err
}
replyer := bufio.NewReader(forwardConn)
raw, _, err := readMessage(replyer)
if err != nil {
return true, err
}
log.Printf("%q", raw)
if _, err := conn.Write(raw); err != nil {
return true, err
}
} }
return false, nil return false, nil

View File

@@ -3,17 +3,18 @@ package src
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"net"
"os" "os"
"slices" "slices"
"strings" "strings"
"sync"
valkey "github.com/redis/go-redis/v9"
) )
type Config struct { type Config struct {
Listen string `json:"LISTEN"` Listen string `json:"LISTEN"`
Forwards string `json:"FORWARDS"` Forwards string `json:"FORWARDS"`
forwards []*valkey.Client forwards []*sync.Pool
} }
func NewConfig(ctx context.Context) (Config, error) { func NewConfig(ctx context.Context) (Config, error) {
@@ -38,17 +39,19 @@ func NewConfig(ctx context.Context) (Config, error) {
forwards := strings.Split(config.Forwards, ",") forwards := strings.Split(config.Forwards, ",")
forwards = slices.DeleteFunc(forwards, func(s string) bool { return s == "" }) forwards = slices.DeleteFunc(forwards, func(s string) bool { return s == "" })
config.forwards = make([]*valkey.Client, len(forwards)) if len(forwards) == 0 {
for i := range forwards { return config, fmt.Errorf("at least one $FORWARD required")
opt, err := valkey.ParseURL(forwards[i])
if err != nil {
config.Close()
return config, err
} }
config.forwards[i] = valkey.NewClient(opt) config.forwards = make([]*sync.Pool, len(forwards))
for i := range forwards {
config.forwards[i] = &sync.Pool{
New: func() any {
v, err := (&net.Dialer{}).DialContext(ctx, "tcp", forwards[i])
if err != nil { if err != nil {
config.Close() return nil
return config, err }
return v
},
} }
} }
@@ -58,7 +61,13 @@ func NewConfig(ctx context.Context) (Config, error) {
func (c Config) Close() { func (c Config) Close() {
for i := range c.forwards { for i := range c.forwards {
if c.forwards[i] != nil { if c.forwards[i] != nil {
c.forwards[i].Close() c.forwards[i].New = nil
for {
got := c.forwards[i].Get()
if got != nil {
got.(net.Conn).Close()
}
}
} }
} }
} }