Pekoe - Redis Pub/Sub Server

Pekoe is a lightweight, Redis-compatible Pub/Sub message broker written in Go using only the standard library. It provides core Redis pub/sub functionality without the complexity of a full Redis implementation.

What is Pub/Sub?

Pub/Sub (Publish/Subscribe) is a messaging pattern where:

• Publishers send messages to channels
• Subscribers receive messages from channels they subscribed to
• The publisher doesn't know who (if anyone) receives the message

Publisher                  Channel              Subscribers
    |                          |                     |
    |--- "hello" to "news" --->|                     |
    |                          |--- "hello" -------->| Client A
    |                          |--- "hello" -------->| Client B
    |                          |--- "hello" -------->| Client C

Quick Start

Build

# Build the binary
make

# Or manually
go build -o pekoe ./cmd/pekoe

Run

# Start the server (config is auto-detected, see Configuration section)
pekoe

# Or specify a config file explicitly
pekoe --config /path/to/config.json

# Show version
pekoe --version

Connect with Redis Client

Use any Redis client to connect. Here are some examples:

Using redis-cli:

redis-cli -p 6379
> SUBSCRIBE news
Reading messages... (press Ctrl-C to exit)
1) "subscribe"
2) "news"
3) (integer) 1

# In another terminal, publish a message:
redis-cli -p 6379 PUBLISH news "Hello World"

Using Python (redis-py):

import redis

r = redis.Redis(port=6379)

# Subscribe to a channel
p = r.pubsub()
p.subscribe('news')

# In another client, publish
r.publish('news', 'Hello World')

# Receive the message
print(p.get_message())  # {'type': 'message', 'channel': 'news', 'data': 'Hello World'}

Using Go (go-redis):

package main

import (
    "context"
    "fmt"
    "time"
    
    "github.com/redis/go-redis/v9"
)

func main() {
    ctx := context.Background()
    
    // Publisher
    pub := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
    pub.Publish(ctx, "news", "Hello from Go!")
    
    // Subscriber
    sub := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
    ps := sub.Subscribe(ctx, "news")
    defer ps.Close()
    
    ch := ps.Channel()
    msg := <-ch
    fmt.Printf("Received: %s\n", msg.Payload)
}

Configuration

Copy config_example.json to your config location and edit:

{
  "bind": "0.0.0.0",
  "port": 6379,
  "max_clients": 10000,
  "read_timeout": 0,
  "write_timeout": 0,
  "max_message_bytes": 536870912,
  "log_level": "info"
}

Config File Lookup

When no --config flag is given, pekoe searches for a config file automatically:

Linux:

• Regular user: ~/.pekoe.json, then /etc/pekoe/config.json
• Root user: /etc/pekoe/config.json only

macOS and BSD:

• ~/.pekoe.json, then /usr/local/etc/pekoe/config.json

Windows:

• --config flag is required, no default lookup

On Linux, macOS, and BSD, if --config is given the specified file is used directly.

Configuration Options

Option	Type	Default	Description
bind	string	"0.0.0.0"	IP address to bind to
port	int	6379	Port number (1-65535)
max_clients	int	10000	Maximum concurrent connections
read_timeout	int	0	Read timeout in seconds (0 = no timeout)
write_timeout	int	0	Write timeout in seconds (0 = no timeout)
max_message_bytes	int	536870912	Max message size in bytes (512MB)
log_level	string	"info"	Logging level: debug, info, warn, error
log_file	string	""	Log file path (empty = stdout only)
pid_file	string	""	PID file path (empty = no PID file)

If you don't provide a config file, sensible defaults are used.

Supported Commands

SUBSCRIBE

Subscribe to one or more channels. The server sends a confirmation message for each channel.

# Subscribe to a single channel
SUBSCRIBE news

# Subscribe to multiple channels
SUBSCRIBE news sports weather

Response:

1) "subscribe"     # message type
2) "news"         # channel name
3) (integer) 1    # total subscriptions

UNSUBSCRIBE

Unsubscribe from channels. If no channel is specified, unsubscribe from all.

# Unsubscribe from specific channel
UNSUBSCRIBE news

# Unsubscribe from all channels
UNSUBSCRIBE

# Unsubscribe from specific channels
UNSUBSCRIBE news sports

PSUBSCRIBE

Subscribe to channels matching a pattern. Supports wildcards:

• * matches any sequence of characters
• ? matches any single character

# Subscribe to all channels starting with "news"
PSUBSCRIBE news*

# Subscribe to multiple patterns
PSUBSCRIBE news.* sports.*

Response:

1) "psubscribe"
2) "news.*"
3) (integer) 1

PUNSUBSCRIBE

Unsubscribe from patterns. If no pattern is specified, unsubscribe from all.

# Unsubscribe from all patterns
PUNSUBSCRIBE

# Unsubscribe from specific pattern
PUNSUBSCRIBE news.*

PUBLISH

Publish a message to a channel. Returns the number of subscribers (both regular and pattern) who received it.

PUBLISH news "Hello World"
# Returns: (integer) 5  (5 subscribers received the message)

PING

Check server connectivity. Returns PONG or the optional message.

PING
# Returns: +PONG

PING hello
# Returns: +hello

PUBSUB

Introspection commands for monitoring:

NUMSUB - Get subscriber count for channels:

PUBSUB NUMSUB news sports
# Returns: 1) "news" 2) (integer) 3  3) "sports" 4) (integer) 5

CHANNELS - List all active channels, optionally filtered by pattern:

PUBSUB CHANNELS
# Returns: 1) "news" 2) "sports" 3) "weather"

PUBSUB CHANNELS news*
# Returns: 1) "news" 2) "news.sports"

NUMPAT - Count of pattern subscriptions:

PUBSUB NUMPAT
# Returns: (integer) 3

HELP - Get help on pubsub commands:

PUBSUB HELP

CLIENT

CLIENT command provides connection introspection and control:

CLIENT ID           # Returns unique connection ID
CLIENT SETNAME x    # Set connection name (max 255 chars, no spaces)
CLIENT GETNAME      # Get connection name (or nil)
CLIENT LIST         # List all connected clients
CLIENT INFO         # Get information about this connection
CLIENT KILL ID id   # Close connection by ID
CLIENT PAUSE 1000   # Pause client (no-op for compatibility)
CLIENT UNPAUSE      # Unpause client (no-op for compatibility)
CLIENT REPLY OFF    # Disable replies (ON|OFF|SKIP)
CLIENT TRACKING ON  # Client-side caching (no-op for compatibility)
CLIENT NO-EVICT ON  # No-evict mode (no-op for compatibility)
CLIENT NO-TOUCH ON  # No-touch mode (no-op for compatibility)
CLIENT CACHING YES  # Caching mode (no-op for compatibility)

CLIENT INFO format:

id=<id> addr=<addr> laddr=<local> fd=-1 age=<seconds> idle=<seconds> flags=<flags> db=0 sub=<count> psub=<count> multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 tot-mem=0 cmd=client user=default redir=-1

Client Library Compatibility

Pekoe has been tested with multiple Redis client libraries:

Library	Language	Status
go-redis/v9	Go	Fully compatible
rueidis	Go	Compatible (use DisableCache: true)
redigo	Go	Fully compatible

Go-redis Example

import "github.com/redis/go-redis/v9"

client := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
ps := client.Subscribe(ctx, "channel")

Rueidis Example

import "github.com/redis/rueidis"

client, _ := rueidis.NewClient(rueidis.ClientOption{
    InitAddress:  []string{"localhost:6379"},
    DisableCache: true,  // Required for pekoe
})

Redigo Example

import "github.com/gomodule/redigo/redis"

conn, _ := redis.Dial("tcp", "localhost:6379")
conn.Send("SUBSCRIBE", "channel")

Architecture

pekoe/
├── cmd/pekoe/       # Entry point
├── internal/        # Internal packages
│   ├── config/      # Configuration loading and validation
│   ├── logger/      # Timestamp-based logging with level filtering
│   ├── protocol/    # RESP protocol parsing and serialization
│   ├── pubsub/      # Pub/Sub core (Hub, Client)
│   ├── server/      # TCP server, connection handling, command routing
│   │   ├── server.go    # Server lifecycle, connection handling
│   │   ├── handlers.go  # Command handlers (subscribe, publish, client, etc.)
│   │   ├── conn.go      # RESP connection reader/writer
│   │   └── signals*.go  # Platform-specific signal handling
│   └── compatibility/  # Client library compatibility tests
├── contrib/         # Contribution files (init scripts, etc.)
├── Makefile         # Build automation
├── config.json      # Default configuration
└── README.md        # This file

Key Components

Hub (pubsub/hub.go)

• Maintains channel-to-subscribers mapping
• Maintains pattern subscriptions
• Handles message broadcasting to both regular and pattern subscribers

Client (pubsub/client.go)

• Represents a connected subscriber
• Tracks which channels/patterns they're subscribed to
• Sends messages to the client

Server (server/server.go, server/handlers.go)

• Listens on TCP socket
• Parses Redis protocol commands
• Routes commands to appropriate handlers in handlers.go
• Enforces pubsub mode (only pubsub commands allowed when subscribed)
• Supports CLIENT command for compatibility

Protocol (protocol/protocol.go)

• Parses RESP (Redis Serialization Protocol)
• Serializes responses back to clients

Logger (logger/logger.go)

• Structured logging with timestamp, level prefix, and key=value fields
• Level filtering (debug, info, warn, error)
• Configurable output (stdout or file) via config file

Testing

# Run all tests with race detection
make test

# Or directly with go
go test -v -race ./...

# Run compatibility tests only
go test -v ./internal/compatibility/...

# Run specific test
go test -v -run TestGoRedisBasicSubscribe ./internal/compatibility/...

Compatibility Tests

The compatibility test suite validates pekoe against multiple Redis client libraries:

• go-redis tests: 27 tests covering basic subscribe, publish, patterns, pubsub commands
• rueidis tests: 9 tests covering connectivity, publish, unicode messages
• redigo tests: 19 tests covering raw protocol interactions

Test coverage includes:

• Basic SUBSCRIBE/PUBLISH/UNSUBSCRIBE
• Pattern PSUBSCRIBE/PUNSUBSCRIBE with wildcards
• Multiple channel/pattern subscriptions
• PUBSUB NUMSUB, CHANNELS, NUMPAT commands
• PING command (with and without message)
• Binary-safe messages
• Empty and long messages
• Unicode content
• Concurrent operations
• Special characters in channel names
• Mixed channel and pattern subscriptions

Logging

Pekoe logs in a structured key=value format with timestamp and log level prefix.

Log format:

YYYY Mmm dd hh:mm:ss LEVEL pekoe[pid]: key=value key=value ...

Example:

2024 Apr  2 14:30:15 INFO pekoe[12345]: event=startup addr=0.0.0.0:6379 max_clients=10000
2024 Apr  2 14:30:20 DEBUG pekoe[12345]: client=192.168.1.100:54321 command=subscribe args=1
2024 Apr  2 14:30:20 DEBUG pekoe[12345]: client=192.168.1.100:54321 event=subscribe channel=news
2024 Apr  2 14:30:25 ERROR pekoe[12345]: client=192.168.1.100:54321 error=unknown_command command=invalidcmd

Log levels:

Level	Description
debug	All messages including DEBUG
info	INFO and above (default)
warn	WARN, ERROR, CRITICAL
error	ERROR, CRITICAL only

Events logged:

Event	Level	Description
event=startup	INFO	Server started
event=shutdown	INFO	Shutdown signal received
event=shutdown_complete	INFO	Server stopped
event=subscribe	DEBUG	Client subscribed to channel
event=unsubscribe	DEBUG	Client unsubscribed from channel
event=psubscribe	DEBUG	Client subscribed to pattern
event=punsubscribe	DEBUG	Client unsubscribed from pattern
event=publish	DEBUG	Message published
event=quit	INFO	Client sent QUIT
event=log_reopened	INFO	Log file reopened (after SIGUSR1/SIGHUP)
event=sighup_ignored	INFO	SIGHUP received (reload not implemented)
error=unknown_command	ERROR	Unknown command received
error=invalid_in_pubsub_mode	ERROR	Invalid command in pubsub mode
error=protocol	ERROR	Protocol error
error=processing	ERROR	Error processing command
error=flush	ERROR	Error flushing response
error=log_reopen_error	ERROR	Failed to reopen log file
event=max_clients_reached	WARN	Connection rejected (max clients)
event=accept_temporary	WARN	Accept error (transient, retrying)
event=accept_error	ERROR	Accept error (fatal)
event=listener_close_error	ERROR	Error closing listener on shutdown

Log Rotation

Pekoe supports log re-opening via signals for log rotation. Configure a PID file to enable easy signal sending.

Configuration example:

{
  "log_file": "/var/log/pekoe.log",
  "pid_file": "/var/run/pekoe.pid"
}

Events logged:

• event=pid_file_created - PID file created on startup
• event=pid_file_removed - PID file removed on shutdown
• event=pid_file_error - Error creating PID file (non-fatal)

Unix/Linux:

# Move the current log file
mv /var/log/pekoe.log /var/log/pekoe.log.1

# Send SIGUSR1 to reopen the log file
kill -USR1 $(cat /var/run/pekoe.pid)

Windows:

# On Windows, SIGHUP is used for log rotation since SIGUSR1 is not available
# Send SIGHUP to reopen the log file
# Note: Signal handling on Windows requires appropriate tools

Example logrotate configuration:

/var/log/pekoe.log {
    daily
    rotate 7
    compress
    delaycompress
    missingok
    notifempty
    postrotate
        kill -USR1 $(cat /var/run/pekoe.pid 2>/dev/null) 2>/dev/null || true
    endscript
}

Why Pekoe?

1. Minimal dependencies - Uses only Go standard library
2. Lightweight - Only implements pub/sub, no extra features
3. Redis compatible - Use existing Redis clients
4. High performance - Fine-grained locking for concurrent operations
5. Pattern subscriptions - Support for PSUBSCRIBE/PUNSUBSCRIBE
6. Easy to understand - Clean, well-documented code
7. Tested - Comprehensive compatibility tests with multiple client libraries

Use Cases

• Real-time notifications - Push updates to connected clients
• Chat systems - Message distribution between users
• Live updates - Stock prices, sports scores, news feeds
• Microservices - Service-to-service messaging
• Load balancing - Distribute work across workers

Error Handling

Pekoe handles errors gracefully:

• Protocol errors - Returns ERR message, closes connection
• Client disconnection - Automatically cleans up subscriptions
• Full channels - Returns 0 for publish to empty channels
• Invalid commands - Returns unknown command error
• Invalid patterns - Returns error for malformed patterns
• Case-insensitive commands - Commands are case-insensitive per Redis spec

Platform Support

Primary platform is Linux (amd64). Pekoe is proven to work on Alpine Linux 3.16 x86_64.

Cross-compilation targets (via make build-all):

• linux/amd64, linux/arm64
• darwin/amd64, darwin/arm64
• freebsd/amd64
• windows/amd64

Building

# Clone the repository
git clone https://github.com/eleksir/pekoe.git
cd pekoe

# Build the binary
make

# Run tests with race detection
make test

# Run linters
make lint

# Cross-compile for all platforms
make build-all

# Install to /usr/local/bin (requires root)
make install

Makefile Targets

Target	Description
make	Clean and build (default)
make build	Build the binary
make test	Run tests with race detection
make clean	Remove built binaries
make build-all	Cross-compile for all platforms
make lint	Run linters (go vet, golangci-lint if available)
make install	Build and install to /usr/local/bin/
make help	Show available targets

Installation

# Install to /usr/local/bin (requires root)
make install

# Or manually
cp pekoe /usr/local/bin/

# Create config directory and copy default config
mkdir -p /etc/pekoe
cp config_example.json /etc/pekoe/config.json

Run

# Start with auto-detected config
./pekoe

# Start with explicit config file
./pekoe --config /path/to/config.json

Alpine Linux Init Script

For Alpine Linux, copy the init script from contrib/pekoe.initd:

cp contrib/pekoe.initd /etc/init.d/pekoe
chmod +x /etc/init.d/pekoe
rc-update add pekoe default

Limitations

• No persistence (messages are not stored)
• No clustering or replication
• No authentication

These are intentional design choices to keep Pekoe simple and focused on pub/sub messaging.