redis_ipc

package module
v0.6.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Dec 20, 2025 License: AGPL-3.0 Imports: 12 Imported by: 0

README

LibreScoot Redis IPC Library

Redis-based IPC library for Go with type-safe generics, functional options, and LibreScoot-specific patterns for hash-based state management.

Features

  • Generics: Type-safe subscriptions and queue handlers
  • Functional Options: Flexible client configuration
  • Connection Callbacks: React to connect/disconnect events
  • Context Propagation: All operations accept context.Context
  • Graceful Shutdown: Wait for handlers to complete
  • Hash State Pattern: Atomic HSET + PUBLISH with change detection
  • Fault Set Management: Redis SET with pub/sub notifications
  • Stream Publishing: XADD with configurable max length
  • Stream Consumption: XREAD with optional consumer groups

Installation

go get github.com/librescoot/redis-ipc

Quick Start

import ipc "github.com/librescoot/redis-ipc"

// Create client with options
client, err := ipc.New(
    ipc.WithAddress("localhost"),
    ipc.WithPort(6379),
    ipc.WithOnConnect(func() { log.Println("connected") }),
    ipc.WithOnDisconnect(func(err error) { log.Println("disconnected") }),
)
if err != nil {
    log.Fatal(err)
}
defer client.Close()

Typed Subscriptions

type VehicleState struct {
    State   string `json:"state"`
    Speed   int    `json:"speed"`
}

// Subscribe with automatic JSON decoding
sub, err := ipc.Subscribe(client, "vehicle:events", func(msg VehicleState) error {
    log.Printf("State: %s, Speed: %d", msg.State, msg.Speed)
    return nil
})
defer sub.Unsubscribe()

// Publish typed messages
ctx := context.Background()
ipc.PublishTyped(client, ctx, "vehicle:events", VehicleState{State: "ready", Speed: 0})

Queue Processing

type Command struct {
    Action string `json:"action"`
}

// Handle queue items with automatic restart on errors
handler := ipc.HandleRequests(client, "scooter:commands", func(cmd Command) error {
    log.Printf("Command: %s", cmd.Action)
    return nil
})
defer handler.Stop()

// Send to queue
ipc.SendRequest(client, ctx, "scooter:commands", Command{Action: "unlock"})

LibreScoot Hash Pattern

The LibreScoot pattern stores state in Redis hashes and notifies via pub/sub:

  • Publisher: HSET vehicle state "ready" → PUBLISH vehicle "state"
  • Consumer: SUBSCRIBE vehicle → receives "state" → HGET vehicle state
HashPublisher
// Create publisher for "vehicle" hash (publishes to "vehicle" channel)
vehicle := client.NewHashPublisher("vehicle")

// Or with a custom channel name
vehicle := client.NewHashPublisherWithChannel("vehicle", "state")

// Set field and publish atomically
vehicle.Set(ctx, "state", "ready")

// Only publish if value changed
changed, _ := vehicle.SetIfChanged(ctx, "state", "ready")

// Batch updates with selective publishing
vehicle.SetManyIfChanged(ctx, map[string]any{
    "state":      "parked",
    "kickstand":  "down",
    "brake:left": "off",
})

// Set with automatic timestamp field
vehicle.SetWithTimestamp(ctx, "state", "ready")
// Sets both "state" and "state:timestamp"

// Delete a single field
vehicle.Delete(ctx, "old-field")

// Clear entire hash
vehicle.Clear(ctx)

// Atomic replace: DEL + HMSET + PUBLISH
vehicle.ReplaceAll(ctx, map[string]any{
    "state": "ready",
    "speed": 0,
})

// Set without publishing (silent update)
vehicle.Set(ctx, "internal-state", "value", ipc.NoPublish())
vehicle.SetMany(ctx, fields, ipc.NoPublish())

// Batch update with single notification
vehicle.SetManyPublishOne(ctx, map[string]any{
    "lat": "52.520",
    "lon": "13.405",
}, "location")  // Publishes only "location", not each field
HashWatcher
// Create watcher for "battery:0" hash
watcher := client.NewHashWatcher("battery:0")

// Or with a custom channel name
watcher := client.NewHashWatcherWithChannel("battery:0", "battery")

// Register field-specific handlers
watcher.OnField("state", func(value string) error {
    log.Printf("Battery state: %s", value)
    return nil
})

watcher.OnField("charge", func(value string) error {
    log.Printf("Battery charge: %s%%", value)
    return nil
})

// Catch-all for unhandled fields
watcher.OnAny(func(field, value string) error {
    log.Printf("%s = %s", field, value)
    return nil
})

// Typed handler with automatic JSON decoding
ipc.OnFieldTyped(watcher, "config", func(cfg Config) error {
    return applyConfig(cfg)
})

// Start watching
watcher.Start()
defer watcher.Stop()

// Fetch initial state
all, _ := watcher.FetchAll(ctx)

// Fetch a single field
state, _ := watcher.Fetch(ctx, "state")
StartWithSync

Subscribe first, fetch current state, then process messages. This avoids race conditions:

watcher := client.NewHashWatcher("vehicle")
watcher.OnField("state", handleState)

// Subscribes, fetches HGETALL, calls handlers, then processes messages
watcher.StartWithSync(ctx)
Debouncing

Coalesce rapid updates - only the last value is passed to handlers after the quiet period:

watcher := client.NewHashWatcher("vehicle")
watcher.SetDebounce(500 * time.Millisecond)  // Wait 500ms after last update
watcher.OnField("speed", handleSpeed)        // Called once after rapid changes settle
watcher.Start()
FaultSet
// Manage fault codes in a Redis SET with pub/sub notification
faults := client.NewFaultSet("battery:0:fault", "battery:0", "fault")

faults.Add(ctx, 35)     // SADD + PUBLISH
faults.Remove(ctx, 35)  // SREM + PUBLISH
faults.Has(ctx, 35)     // SISMEMBER
faults.All(ctx)         // SMEMBERS
faults.Clear(ctx)       // DEL + PUBLISH

Redis Streams

StreamPublisher

Publish events to a Redis stream using XADD:

// Create publisher with default max length (1000)
stream := client.NewStreamPublisher("events:faults")

// Or with custom max length
stream := client.NewStreamPublisher("events:faults", ipc.WithMaxLen(5000))

// Publish a map
id, err := stream.Add(ctx, map[string]any{
    "group":       "battery:0",
    "code":        "35",
    "description": "NFC Reader Error",
})

// Publish a typed struct (JSON-encoded to fields)
type FaultEvent struct {
    Group       string `json:"group"`
    Code        int    `json:"code"`
    Description string `json:"description"`
}

id, err := ipc.StreamAdd(stream, ctx, FaultEvent{
    Group:       "battery:0",
    Code:        35,
    Description: "NFC Reader Error",
})
StreamConsumer

Consume stream messages using XREAD:

// Create consumer
consumer := client.NewStreamConsumer("events:faults",
    ipc.WithBlockTimeout(1 * time.Second),
)

// Set handler
consumer.Handle(func(id string, values map[string]string) error {
    log.Printf("Fault %s: group=%s code=%s", id, values["group"], values["code"])
    return nil
})

// Start from beginning ("0") or only new messages ("$")
consumer.Start(ctx, "0")

// Or use typed handler
ipc.StreamHandle(consumer, func(id string, evt FaultEvent) error {
    log.Printf("Fault: %+v", evt)
    return nil
})
consumer.Start(ctx, "$")
Consumer Groups

For multi-instance consumption with acknowledgment:

consumer := client.NewStreamConsumer("events:faults",
    ipc.WithBlockTimeout(1 * time.Second),
    ipc.WithConsumerGroup("uplink-service", "instance-1"),
)

consumer.Handle(handler)

// ">" means only undelivered messages to this group
consumer.Start(ctx, ">")

Transactions

tx := client.NewTxGroup()
tx.Add("HSET", "vehicle", "state", "ready").
   Add("HSET", "vehicle", "state:timestamp", time.Now().Unix()).
   Add("PUBLISH", "vehicle", "state")

results, err := tx.Exec(ctx)

Message Router

For JSON envelope-based routing ({"type": "...", "data": {...}}):

router := client.NewRouter("events")

ipc.Handle(router, "state", func(s StateMsg) error { ... })
ipc.Handle(router, "error", func(e ErrorMsg) error { ... })

router.Start()
defer router.Stop()

// Publish routed messages
ipc.PublishRouted(client, ctx, "events", "state", StateMsg{...})

Configuration Options

client, err := ipc.New(
    ipc.WithURL("redis://localhost:6379"),  // Or just "localhost:6379" or "localhost"
    // Or use separate address/port:
    // ipc.WithAddress("localhost"),
    // ipc.WithPort(6379),
    ipc.WithRetryInterval(5 * time.Second),
    ipc.WithMaxRetries(3),
    ipc.WithPoolSize(3),
    ipc.WithDialTimeout(2 * time.Second),
    ipc.WithLogger(slog.Default()),
    ipc.WithCodec(ipc.JSONCodec{}),  // or ipc.StringCodec{}
    ipc.WithOnConnect(func() { ... }),
    ipc.WithOnDisconnect(func(err error) { ... }),
)

Direct Redis Operations

All operations accept context.Context:

// Strings
client.Get(ctx, "key")
client.Set(ctx, "key", "value", 0)
client.Incr(ctx, "counter")

// Hashes
client.HGet(ctx, "hash", "field")
client.HSet(ctx, "hash", "field", "value")
client.HGetAll(ctx, "hash")

// Lists
client.LPush(ctx, "queue", "value")
client.BRPop(ctx, time.Second, "queue")

// Pub/Sub
client.Publish(ctx, "channel", "message")

// Keys
client.Exists(ctx, "key")
client.Del(ctx, "key")
client.Expire(ctx, "key", time.Hour)

// Health check
client.Ping(ctx)

// Raw command
client.Do(ctx, "PING")

// Access underlying go-redis client
client.Raw().Scan(ctx, ...)

Best Practices

Reuse Publishers and Watchers

Create publishers once and store them in your service struct:

// Good: create once, reuse
type Service struct {
    client   *ipc.Client
    powerPub *ipc.HashPublisher
    battPub  *ipc.HashPublisher
}

func NewService(client *ipc.Client) *Service {
    return &Service{
        client:   client,
        powerPub: client.NewHashPublisher("power-manager"),
        battPub:  client.NewHashPublisher("battery:0"),
    }
}

func (s *Service) UpdatePowerState(ctx context.Context, state string) error {
    return s.powerPub.Set(ctx, "state", state)
}

// Bad: creates new publisher on every call (wasteful)
func (s *Service) UpdatePowerStateBad(ctx context.Context, state string) error {
    pub := s.client.NewHashPublisher("power-manager")
    return pub.Set(ctx, "state", state)
}
Use SetManyIfChanged for Bulk Updates

When updating multiple fields, SetManyIfChanged only publishes changed fields:

// Returns list of actually-changed fields (useful for logging)
changed, err := pub.SetManyIfChanged(ctx, map[string]any{
    "state":  newState,
    "speed":  newSpeed,
    "charge": newCharge,
})
if len(changed) > 0 {
    log.Printf("Updated fields: %v", changed)
}
Use ReplaceAll for Complete Hash Replacement

When you need to atomically clear and repopulate a hash (e.g., inhibitor lists):

// Atomic: DEL + HMSET + PUBLISH in one transaction
pub.ReplaceAll(ctx, map[string]any{
    "inhibitor1": "reason1",
    "inhibitor2": "reason2",
})

// Clear the hash entirely
pub.ReplaceAll(ctx, nil)  // or pub.Clear(ctx)
Use StartWithSync for Initial State

When you need current values before processing updates:

watcher := client.NewHashWatcher("vehicle")
watcher.OnField("state", handleState)

// StartWithSync: Subscribe → HGETALL → call handlers → process messages
// Ensures no messages are missed during initialization
watcher.StartWithSync(ctx)

// vs Start(): Just subscribes, doesn't fetch initial state
// watcher.Start()

License

AGPL-3.0, see LICENSE for details.

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func PublishRouted 

func PublishRouted[T any](c *Client, ctx context.Context, channel, msgType string, data T) error

PublishRouted publishes a message with type routing

func PublishTyped 

func PublishTyped[T any](c *Client, ctx context.Context, channel string, message T) error

PublishTyped publishes a typed message using the client's codec

func SendRequest 

func SendRequest[T any](c *Client, ctx context.Context, queue string, message T) error

SendRequest sends a typed message to a queue

func StreamAdd 

func StreamAdd[T any](sp *StreamPublisher, ctx context.Context, msg T) (string, error)

StreamAdd publishes a typed message to the stream using JSON encoding.

Types

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client is the main Redis IPC client

func New 

func New(opts ...Option) (*Client, error)

New creates a new Redis IPC client with functional options

func (*Client) BLPop 

func (c *Client) BLPop(ctx context.Context, timeout time.Duration, keys ...string) ([]string, error)

BLPop blocks until it can remove and return the first element of a list

func (*Client) BRPop 

func (c *Client) BRPop(ctx context.Context, timeout time.Duration, keys ...string) ([]string, error)

BRPop blocks until it can remove and return the last element of a list

func (*Client) Close 

func (c *Client) Close() error

Close gracefully shuts down the client, waiting for handlers to finish

func (*Client) CloseWithTimeout 

func (c *Client) CloseWithTimeout(timeout time.Duration) error

CloseWithTimeout shuts down the client with a maximum wait time

func (*Client) Codec 

func (c *Client) Codec() Codec

Codec returns the client's codec

func (*Client) Connected 

func (c *Client) Connected() bool

Connected returns the current connection status

func (*Client) Decr 

func (c *Client) Decr(ctx context.Context, key string) (int64, error)

Decr decrements the integer value of a key by one

func (*Client) Del 

func (c *Client) Del(ctx context.Context, keys ...string) (int64, error)

Del deletes keys

func (*Client) Do 

func (c *Client) Do(ctx context.Context, cmd string, args ...interface{}) (interface{}, error)

Do executes a Redis command

func (*Client) Exists 

func (c *Client) Exists(ctx context.Context, keys ...string) (int64, error)

Exists checks if keys exist

func (*Client) Expire 

func (c *Client) Expire(ctx context.Context, key string, expiration time.Duration) (bool, error)

Expire sets a key's time to live

func (*Client) Get 

func (c *Client) Get(ctx context.Context, key string) (string, error)

Get retrieves the value of a key

func (*Client) HGet 

func (c *Client) HGet(ctx context.Context, key, field string) (string, error)

HGet retrieves the value of a hash field

func (*Client) HGetAll 

func (c *Client) HGetAll(ctx context.Context, key string) (map[string]string, error)

HGetAll retrieves all fields and values of a hash

func (*Client) HSet 

func (c *Client) HSet(ctx context.Context, key, field string, value interface{}) error

HSet sets the value of a hash field

func (*Client) Hash added in v0.6.0 

func (c *Client) Hash(name string) *HashPublisher

Hash returns a cached HashPublisher for the given hash name. Publishers are created on first access and reused thereafter. This is more ergonomic than NewHashPublisher when you don't need to store the publisher in a struct field.

func (*Client) Incr 

func (c *Client) Incr(ctx context.Context, key string) (int64, error)

Incr increments the integer value of a key by one

func (*Client) LPop 

func (c *Client) LPop(ctx context.Context, key string) (string, error)

LPop removes and returns the first element of a list

func (*Client) LPush 

func (c *Client) LPush(ctx context.Context, key string, values ...interface{}) (int64, error)

LPush inserts values at the head of a list

func (*Client) Logger 

func (c *Client) Logger() *slog.Logger

Logger returns the client's logger

func (*Client) NewFaultSet 

func (c *Client) NewFaultSet(setKey, channel, payload string) *FaultSet

NewFaultSet creates a fault set manager. Example: NewFaultSet("battery:0:fault", "battery:0", "fault")

func (*Client) NewHashPublisher 

func (c *Client) NewHashPublisher(hash string) *HashPublisher

NewHashPublisher creates a publisher for a hash. By convention, the channel name equals the hash name.

func (*Client) NewHashPublisherWithChannel 

func (c *Client) NewHashPublisherWithChannel(hash, channel string) *HashPublisher

NewHashPublisherWithChannel creates a publisher with a custom channel name.

func (*Client) NewHashWatcher 

func (c *Client) NewHashWatcher(hash string) *HashWatcher

NewHashWatcher creates a watcher for a hash. By convention, the channel name equals the hash name.

func (*Client) NewHashWatcherWithChannel 

func (c *Client) NewHashWatcherWithChannel(hash, channel string) *HashWatcher

NewHashWatcherWithChannel creates a watcher with a custom channel name.

func (*Client) NewRouter 

func (c *Client) NewRouter(channel string) *Router

NewRouter creates a message router for a channel

func (*Client) NewStreamConsumer 

func (c *Client) NewStreamConsumer(stream string, opts ...StreamConsumerOption) *StreamConsumer

NewStreamConsumer creates a consumer for a Redis stream.

func (*Client) NewStreamPublisher 

func (c *Client) NewStreamPublisher(stream string, opts ...StreamPublisherOption) *StreamPublisher

NewStreamPublisher creates a publisher for a Redis stream.

func (*Client) NewTxGroup 

func (c *Client) NewTxGroup() *TxGroup

NewTxGroup creates a new transaction group

func (*Client) Ping added in v0.5.0 

func (c *Client) Ping(ctx context.Context) error

Ping checks if the Redis server is reachable

func (*Client) Publish 

func (c *Client) Publish(ctx context.Context, channel string, message interface{}) (int64, error)

Publish publishes a message to a channel

func (*Client) RPop 

func (c *Client) RPop(ctx context.Context, key string) (string, error)

RPop removes and returns the last element of a list

func (*Client) RPush 

func (c *Client) RPush(ctx context.Context, key string, values ...interface{}) (int64, error)

RPush inserts values at the tail of a list

func (*Client) Raw 

func (c *Client) Raw() *redis.Client

Raw returns the underlying go-redis client for advanced operations

func (*Client) Set 

func (c *Client) Set(ctx context.Context, key string, value interface{}, expiration time.Duration) error

Set sets the value of a key

type Codec 

type Codec interface {
	Encode(v any) ([]byte, error)
	Decode(data []byte, v any) error
}

Codec handles message serialization/deserialization

type Envelope 

type Envelope struct {
	Type string          `json:"type"`
	Data json.RawMessage `json:"data"`
}

Envelope is the standard message format for routed messages

type FaultSet 

type FaultSet struct {
	// contains filtered or unexported fields
}

FaultSet manages a Redis set of fault codes with pub/sub notification.

func (*FaultSet) Add 

func (fs *FaultSet) Add(ctx context.Context, code int) error

Add adds a fault code to the set and publishes notification.

func (*FaultSet) All 

func (fs *FaultSet) All(ctx context.Context) ([]string, error)

All returns all fault codes in the set.

func (*FaultSet) Clear 

func (fs *FaultSet) Clear(ctx context.Context) error

Clear removes all fault codes and publishes notification.

func (*FaultSet) Has 

func (fs *FaultSet) Has(ctx context.Context, code int) (bool, error)

Has checks if a fault code is present.

func (*FaultSet) Remove 

func (fs *FaultSet) Remove(ctx context.Context, code int) error

Remove removes a fault code from the set and publishes notification.

func (*FaultSet) SetMany 

func (fs *FaultSet) SetMany(ctx context.Context, codes []int) error

SetMany atomically adds/removes multiple codes. positive codes are added, negative codes are removed.

type HashPublisher 

type HashPublisher struct {
	// contains filtered or unexported fields
}

HashPublisher provides atomic HSET + PUBLISH with change detection. This matches the LibreScoot pattern where: - State is stored in a Redis hash - Changes are notified via pub/sub with the field name as payload - Consumers fetch the value using HGET

func (*HashPublisher) Channel 

func (hp *HashPublisher) Channel() string

Channel returns the channel name.

func (*HashPublisher) Clear added in v0.4.0 

func (hp *HashPublisher) Clear(ctx context.Context) error

Clear deletes the entire hash and publishes "cleared" notification.

func (*HashPublisher) Delete added in v0.4.0 

func (hp *HashPublisher) Delete(ctx context.Context, field string) error

Delete removes a field from the hash and publishes notification.

func (*HashPublisher) Get 

func (hp *HashPublisher) Get(ctx context.Context, field string) (string, error)

Get retrieves a field value from the hash.

func (*HashPublisher) GetAll 

func (hp *HashPublisher) GetAll(ctx context.Context) (map[string]string, error)

GetAll retrieves all fields from the hash.

func (*HashPublisher) Hash 

func (hp *HashPublisher) Hash() string

Hash returns the hash name.

func (*HashPublisher) ReplaceAll added in v0.4.0 

func (hp *HashPublisher) ReplaceAll(ctx context.Context, fields map[string]any) error

ReplaceAll atomically deletes all fields and sets new ones. This solves the race condition in DEL + HMSET + PUBLISH patterns. If fields is nil or empty, just deletes and publishes "cleared".

func (*HashPublisher) Set 

func (hp *HashPublisher) Set(ctx context.Context, field string, value any, opts ...SetOption) error

Set atomically sets a hash field and publishes the field name. By default, always publishes. Use NoPublish() to skip publishing.

func (*HashPublisher) SetIfChanged 

func (hp *HashPublisher) SetIfChanged(ctx context.Context, field string, value any) (bool, error)

SetIfChanged atomically sets a hash field and publishes only if the value changed. Returns true if the value was changed and published.

func (*HashPublisher) SetMany 

func (hp *HashPublisher) SetMany(ctx context.Context, fields map[string]any, opts ...SetOption) error

SetMany atomically sets multiple fields and publishes each one. By default, publishes each field. Use NoPublish() to skip publishing.

func (*HashPublisher) SetManyIfChanged 

func (hp *HashPublisher) SetManyIfChanged(ctx context.Context, fields map[string]any) ([]string, error)

SetManyIfChanged atomically sets multiple fields, publishing only changed ones. Returns the list of fields that were actually changed.

func (*HashPublisher) SetManyPublishOne added in v0.5.0 

func (hp *HashPublisher) SetManyPublishOne(ctx context.Context, fields map[string]any, notification string) error

SetManyPublishOne sets multiple fields but publishes only a single notification. Useful for batch updates where you only want one notification at the end. The notification parameter is published as the message payload.

func (*HashPublisher) SetWithTimestamp 

func (hp *HashPublisher) SetWithTimestamp(ctx context.Context, field string, value any) error

SetWithTimestamp atomically sets a field, its timestamp, and publishes.

type HashWatcher 

type HashWatcher struct {
	// contains filtered or unexported fields
}

HashWatcher subscribes to a channel and fetches hash values on notification. This matches the LibreScoot consumer pattern.

func OnFieldTyped 

func OnFieldTyped[T any](hw *HashWatcher, field string, handler func(T) error) *HashWatcher

OnFieldTyped registers a typed handler for a specific field. The value is JSON-decoded into the type T.

func (*HashWatcher) Fetch 

func (hw *HashWatcher) Fetch(ctx context.Context, field string) (string, error)

Fetch retrieves a specific field from the hash.

func (*HashWatcher) FetchAll 

func (hw *HashWatcher) FetchAll(ctx context.Context) (map[string]string, error)

FetchAll retrieves all current values from the hash. Useful for initial state sync.

func (*HashWatcher) OnAny 

func (hw *HashWatcher) OnAny(handler func(field, value string) error) *HashWatcher

OnAny registers a catch-all handler for any field not explicitly handled.

func (*HashWatcher) OnField 

func (hw *HashWatcher) OnField(field string, handler func(value string) error) *HashWatcher

OnField registers a handler for a specific field. When a message is received with this field name, the handler is called with the fetched value from the hash.

func (*HashWatcher) SetDebounce 

func (hw *HashWatcher) SetDebounce(d time.Duration) *HashWatcher

SetDebounce sets a debounce duration for handler calls. When set, rapid updates to the same field are coalesced - only the last value is passed to the handler after the quiet period expires.

func (*HashWatcher) Start 

func (hw *HashWatcher) Start() error

Start begins listening for messages and fetching values.

func (*HashWatcher) StartWithSync 

func (hw *HashWatcher) StartWithSync(ctx context.Context) error

StartWithSync subscribes, fetches current state, calls handlers, then processes messages. This ensures no messages are missed between initial fetch and subscribe. Order: Subscribe (messages buffer) → HGETALL → call handlers → process buffered messages

func (*HashWatcher) Stop 

func (hw *HashWatcher) Stop() error

Stop stops the watcher.

type JSONCodec 

type JSONCodec struct{}

JSONCodec is the default codec using JSON

func (JSONCodec) Decode 

func (JSONCodec) Decode(data []byte, v any) error

func (JSONCodec) Encode 

func (JSONCodec) Encode(v any) ([]byte, error)

type Option 

type Option func(*options)

Option configures the client

func WithAddress 

func WithAddress(addr string) Option

WithAddress sets the Redis server address

func WithCodec 

func WithCodec(codec Codec) Option

WithCodec sets the message codec (default: JSONCodec)

func WithDialTimeout 

func WithDialTimeout(d time.Duration) Option

WithDialTimeout sets the connection timeout

func WithLogger 

func WithLogger(logger *slog.Logger) Option

WithLogger sets a custom logger

func WithMaxRetries 

func WithMaxRetries(n int) Option

WithMaxRetries sets the maximum number of retries for commands

func WithOnConnect 

func WithOnConnect(fn func()) Option

WithOnConnect sets a callback invoked when connection is established

func WithOnDisconnect 

func WithOnDisconnect(fn func(error)) Option

WithOnDisconnect sets a callback invoked when connection is lost

func WithPoolSize 

func WithPoolSize(n int) Option

WithPoolSize sets the connection pool size

func WithPort 

func WithPort(port int) Option

WithPort sets the Redis server port

func WithRetryInterval 

func WithRetryInterval(d time.Duration) Option

WithRetryInterval sets the connection monitoring interval

func WithURL added in v0.5.0 

func WithURL(rawURL string) Option

WithURL sets the Redis server address from a URL string. Supports formats: "redis://host:port", "host:port", or just "host". The redis:// scheme is optional. Port defaults to 6379 if not specified.

type QueueHandler 

type QueueHandler[T any] struct {
	// contains filtered or unexported fields
}

QueueHandler processes messages from a Redis list

func HandleRequests 

func HandleRequests[T any](c *Client, queue string, handler func(T) error) *QueueHandler[T]

HandleRequests creates a typed queue handler

func (*QueueHandler[T]) Stop 

func (qh *QueueHandler[T]) Stop()

Stop stops the queue handler

type Router 

type Router struct {
	// contains filtered or unexported fields
}

Router provides message-type based routing for a channel

func Handle 

func Handle[T any](r *Router, msgType string, handler func(T) error) *Router

Handle registers a typed handler for a message type

func (*Router) Start 

func (r *Router) Start() error

Start begins listening for messages

func (*Router) Stop 

func (r *Router) Stop() error

Stop stops the router

type SetOption added in v0.5.0 

type SetOption func(*setConfig)

SetOption configures Set/SetMany behavior

func NoPublish added in v0.5.0 

func NoPublish() SetOption

NoPublish prevents publishing after the set operation. Use when you want to update state without notifying subscribers.

type StreamConsumer 

type StreamConsumer struct {
	// contains filtered or unexported fields
}

StreamConsumer consumes messages from a Redis stream using XREAD.

func StreamHandle 

func StreamHandle[T any](sc *StreamConsumer, handler func(id string, msg T) error) *StreamConsumer

StreamHandle registers a typed handler for stream messages.

func (*StreamConsumer) Handle 

func (sc *StreamConsumer) Handle(handler func(id string, values map[string]string) error) *StreamConsumer

Handle sets the message handler function.

func (*StreamConsumer) Running 

func (sc *StreamConsumer) Running() bool

Running returns whether the consumer is currently running.

func (*StreamConsumer) Start 

func (sc *StreamConsumer) Start(ctx context.Context, startID string) error

Start begins consuming messages from the stream. startID can be:

  • "$" to read only new messages
  • "0" to read from the beginning
  • A specific message ID to read after that ID
  • ">" for consumer groups (only undelivered messages)

func (*StreamConsumer) Stop 

func (sc *StreamConsumer) Stop()

Stop stops the consumer (by cancelling the context passed to Start). Note: The consumer stops when the context passed to Start is cancelled.

type StreamConsumerOption 

type StreamConsumerOption func(*streamConsumerOptions)

StreamConsumerOption configures a StreamConsumer

func WithBlockTimeout 

func WithBlockTimeout(d time.Duration) StreamConsumerOption

WithBlockTimeout sets the XREAD block timeout

func WithConsumerGroup 

func WithConsumerGroup(group, consumer string) StreamConsumerOption

WithConsumerGroup configures consumer group mode (XREADGROUP)

type StreamPublisher 

type StreamPublisher struct {
	// contains filtered or unexported fields
}

StreamPublisher publishes messages to a Redis stream using XADD.

func (*StreamPublisher) Add 

func (sp *StreamPublisher) Add(ctx context.Context, values map[string]any) (string, error)

Add publishes a message to the stream with auto-generated ID. Returns the generated message ID.

func (*StreamPublisher) AddWithID 

func (sp *StreamPublisher) AddWithID(ctx context.Context, id string, values map[string]any) (string, error)

AddWithID publishes a message with a specific ID.

func (*StreamPublisher) Stream 

func (sp *StreamPublisher) Stream() string

Stream returns the stream name.

type StreamPublisherOption 

type StreamPublisherOption func(*streamPublisherOptions)

StreamPublisherOption configures a StreamPublisher

func WithMaxLen 

func WithMaxLen(n int64) StreamPublisherOption

WithMaxLen sets the maximum stream length (MAXLEN option)

type StringCodec 

type StringCodec struct{}

StringCodec passes strings through unchanged

func (StringCodec) Decode 

func (StringCodec) Decode(data []byte, v any) error

func (StringCodec) Encode 

func (StringCodec) Encode(v any) ([]byte, error)

type Subscription 

type Subscription[T any] struct {
	// contains filtered or unexported fields
}

Subscription is a typed channel subscription

func Subscribe 

func Subscribe[T any](c *Client, channel string, handler func(T) error) (*Subscription[T], error)

Subscribe creates or retrieves a typed subscription

func (*Subscription[T]) Unsubscribe 

func (s *Subscription[T]) Unsubscribe() error

Unsubscribe stops the subscription

type TxGroup 

type TxGroup struct {
	// contains filtered or unexported fields
}

TxGroup batches Redis commands for atomic execution

func (*TxGroup) Add 

func (g *TxGroup) Add(cmd string, args ...interface{}) *TxGroup

Add queues a Redis command

func (*TxGroup) Exec 

func (g *TxGroup) Exec(ctx context.Context) ([]interface{}, error)

Exec executes all queued commands and returns results

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.