redis_ipc

Aug 4, 2026

Changes in this version

+ var ErrClientClosed = fmt.Errorf("redis-ipc: client is closed")

May 8, 2026

Changes in this version

type Client

+ func (c *Client) NewFaultReporter(group string, opts ...FaultReporterOption) *FaultReporter

+ type FaultReporter struct

+ func (fr *FaultReporter) All() ([]int, error)

+ func (fr *FaultReporter) Clear(code int) error

+ func (fr *FaultReporter) Group() string

+ func (fr *FaultReporter) Has(code int) (bool, error)

+ func (fr *FaultReporter) Raise(code int, description string) error

+ type FaultReporterOption func(*faultReporterOptions)

+ func WithFaultChannel(ch string) FaultReporterOption

+ func WithFaultMaxLen(n int64) FaultReporterOption

+ func WithFaultPayload(p string) FaultReporterOption

+ func WithFaultSetKey(key string) FaultReporterOption

+ func WithFaultStream(s string) FaultReporterOption

May 5, 2026

Changes in this version

+ var ErrUnknownMethod = errors.New("unknown method")

+ func CallMethod[Req, Resp any](c *Client, channel, method string, req Req, timeout time.Duration) (Resp, error)

+ func RegisterCall[Req, Resp any](s *CallServer, method string, handler func(Req) (Resp, error))

+ type CallServer struct

+ func NewCallServer(c *Client, channel string, opts ...CallServerOption) *CallServer

+ func (s *CallServer) Start()

+ func (s *CallServer) Stop()

+ type CallServerOption func(*callServerOpts)

+ func WithCallServerConcurrency(n int) CallServerOption

May 5, 2026

Changes in this version

+ var ErrCallTimeout = errors.New("redis-ipc: call timed out")

+ func Call[Req, Resp any](c *Client, channel string, req Req, timeout time.Duration) (Resp, error)

+ func IsCallError(err error) bool

+ type CallError struct

+ Msg string

+ func (e *CallError) Error() string

+ type CallHandler struct

+ func HandleCalls[Req, Resp any](c *Client, channel string, handler func(Req) (Resp, error), ...) *CallHandler[Req, Resp]

+ func (h *CallHandler[Req, Resp]) Stop()

+ type CallHandlerOption func(*callHandlerOpts)

+ func WithCallConcurrency(n int) CallHandlerOption

Mar 12, 2026

Changes in this version

+ var ErrNil = redis.Nil

Mar 12, 2026

Changes in this version

type Client

+ func (c *Client) NewTx() *Tx

+ func (c *Client) NewTxWithContext(ctx context.Context) *Tx

+ type Tx struct

+ func (tx *Tx) Exec() error

+ func (tx *Tx) FaultAdd(fs *FaultSet, code int)

+ func (tx *Tx) FaultRemove(fs *FaultSet, code int)

+ func (tx *Tx) FaultUpdate(fs *FaultSet, adds, removes []int)

+ func (tx *Tx) HashSetManyIfChanged(hp *HashPublisher, fields map[string]any) []string

+ func (tx *Tx) StreamAdd(sp *StreamPublisher, values map[string]any)

Jan 28, 2026

Changes in this version

type SetOption

+ func Sync() SetOption

Dec 25, 2025

Changes in this version

type HashWatcher

+ func (hw *HashWatcher) OnEvent(event string, handler func() error) *HashWatcher

Dec 20, 2025

Changes in this version

type Client

+ func (c *Client) Context() context.Context

+ func (c *Client) WithContext(ctx context.Context) *ContextClient

+ type ContextClient struct

+ func (cc *ContextClient) Context() context.Context

Dec 20, 2025

Changes in this version

type Client

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

Dec 20, 2025

Changes in this version

type Client

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

type HashPublisher

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

type Option

+ func WithURL(rawURL string) Option

+ type SetOption func(*setConfig)

+ func NoPublish() SetOption

Dec 20, 2025

Changes in this version

type HashPublisher

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

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

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

Dec 20, 2025

Changes in this version

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

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

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

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

+ type Client struct

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

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

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

+ func (c *Client) Close() error

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

+ func (c *Client) Codec() Codec

+ func (c *Client) Connected() bool

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

+ func (c *Client) NewTxGroup() *TxGroup

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

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

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

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

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

+ type Codec interface

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

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

+ type Envelope struct

+ Data json.RawMessage

+ Type string

+ type FaultSet struct

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

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

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

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

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

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

+ type HashPublisher struct

+ func (hp *HashPublisher) Channel() string

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

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

+ func (hp *HashPublisher) Hash() string

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

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

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

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

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

+ type HashWatcher struct

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

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

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

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

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

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

+ func (hw *HashWatcher) Start() error

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

+ func (hw *HashWatcher) Stop() error

+ type JSONCodec struct

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

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

+ type Option func(*options)

+ func WithAddress(addr string) Option

+ func WithCodec(codec Codec) Option

+ func WithDialTimeout(d time.Duration) Option

+ func WithLogger(logger *slog.Logger) Option

+ func WithMaxRetries(n int) Option

+ func WithOnConnect(fn func()) Option

+ func WithOnDisconnect(fn func(error)) Option

+ func WithPoolSize(n int) Option

+ func WithPort(port int) Option

+ func WithRetryInterval(d time.Duration) Option

+ type QueueHandler struct

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

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

+ type Router struct

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

+ func (r *Router) Start() error

+ func (r *Router) Stop() error

+ type StreamConsumer struct

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

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

+ func (sc *StreamConsumer) Running() bool

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

+ func (sc *StreamConsumer) Stop()

+ type StreamConsumerOption func(*streamConsumerOptions)

+ func WithBlockTimeout(d time.Duration) StreamConsumerOption

+ func WithConsumerGroup(group, consumer string) StreamConsumerOption

+ type StreamPublisher struct

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

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

+ func (sp *StreamPublisher) Stream() string

+ type StreamPublisherOption func(*streamPublisherOptions)

+ func WithMaxLen(n int64) StreamPublisherOption

+ type StringCodec struct

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

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

+ type Subscription struct

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

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

+ type TxGroup struct

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

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