valkey

package module
v3.0.6 Latest Latest
Warning

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

 Go to latest Published: Jun 20, 2026 License: MIT Imports: 13 Imported by: 0

README

@socket.io/valkey-adapter (Go)

A Valkey adapter for Socket.IO in Go. This module provides Socket.IO clustering via Valkey Pub/Sub, Valkey Sharded Pub/Sub (Valkey 7+), and Valkey Streams — mirroring the functionality of the adapters/redis module but using the valkey-go client.

Adapter Types

Type Description
ValkeyAdapterBuilder Classic Pub/Sub — suitable for standalone and replicated Valkey
ShardedValkeyAdapterBuilder Sharded Pub/Sub (SSUBSCRIBE/SPUBLISH) — for Valkey Cluster
ValkeyStreamsAdapterBuilder Valkey Streams — persistent messages + session recovery

Installation

go get github.com/technance-foundation/socket.io/adapters/valkey/v3

Usage

Classic Pub/Sub Adapter
import (
    "context"

    vk "github.com/valkey-io/valkey-go"
    io "github.com/technance-foundation/socket.io/servers/socket/v3"
    vkadapter "github.com/technance-foundation/socket.io/adapters/valkey/v3/adapter"
    valkey "github.com/technance-foundation/socket.io/adapters/valkey/v3"
)

client, err := vk.NewClient(vk.ClientOption{
    InitAddress: []string{"localhost:6379"},
})
if err != nil {
    log.Fatal(err)
}

valkeyClient := valkey.NewValkeyClient(context.Background(), client)

server := io.NewServer(nil, nil)
server.SetAdapter(&vkadapter.ValkeyAdapterBuilder{Valkey: valkeyClient})
Sharded Pub/Sub Adapter (Valkey Cluster)
server.SetAdapter(&vkadapter.ShardedValkeyAdapterBuilder{Valkey: valkeyClient})
Streams Adapter
server.SetAdapter(&vkadapter.ValkeyStreamsAdapterBuilder{Valkey: valkeyClient})
Reusing an Existing Client

Pass a pre-created vk.Client to NewValkeyClient. This allows you to share an existing connection pool instead of creating a second one:

// existing client created elsewhere in your application
existingClient := myAppValkeyClient

valkeyClient := valkey.NewValkeyClient(ctx, existingClient)
server.SetAdapter(&vkadapter.ValkeyAdapterBuilder{Valkey: valkeyClient})
Emitter

Use the emitter to broadcast events from a process that does not run a Socket.IO server:

import (
    "context"

    vk "github.com/valkey-io/valkey-go"
    valkey "github.com/technance-foundation/socket.io/adapters/valkey/v3"
    "github.com/technance-foundation/socket.io/adapters/valkey/v3/emitter"
)

client, _ := vk.NewClient(vk.ClientOption{InitAddress: []string{"localhost:6379"}})
valkeyClient := valkey.NewValkeyClient(context.Background(), client)

e := emitter.NewEmitter(valkeyClient, nil)
e.To("room1").Emit("hello", "world")

Configuration

ValkeyAdapterOptions
Option Type Default Description
Key string "socket.io" Channel prefix
RequestsTimeout time.Duration 5000ms Inter-node request timeout
PublishOnSpecificResponseChannel bool false Route responses to per-node channels
Parser valkey.Parser MsgPack Encoder/decoder for messages
ShardedValkeyAdapterOptions
Option Type Default Description
ChannelPrefix string "socket.io" Channel prefix
SubscriptionMode SubscriptionMode DynamicSubscriptionMode Channel strategy
ValkeyStreamsAdapterOptions
Option Type Default Description
StreamName string "socket.io" Stream name
MaxLen int64 10000 Approximate stream max length
ReadCount int64 100 Messages per XREAD call
SessionKeyPrefix string "sio:session:" Session key prefix

Subscription Modes

Mode Description
StaticSubscriptionMode 2 fixed channels per namespace
DynamicSubscriptionMode 2 + 1 channel per public room (default)
DynamicPrivateSubscriptionMode Separate channel per room (including private)

License

MIT

Documentation

Overview

Package valkey defines constants for Valkey-based message types used in the Socket.IO adapter. These message types are used for inter-node communication in a clustered Socket.IO environment.

Package valkey provides a Valkey client wrapper for the Socket.IO Valkey adapter. It bridges valkey-go's callback-based Pub/Sub API to the channel-based patterns used by the adapter and emitter implementations.

Package valkey provides Valkey-based adapter types and interfaces for Socket.IO clustering. These types define the message structures used for inter-node communication via Valkey.

Index

Constants

View Source 
const (
	// SOCKETS requests a list of socket IDs from other nodes.
	SOCKETS adapter.MessageType = iota

	// ALL_ROOMS requests a list of all rooms from other nodes.
	ALL_ROOMS

	// REMOTE_JOIN instructs other nodes to join a socket to specified rooms.
	REMOTE_JOIN

	// REMOTE_LEAVE instructs other nodes to remove a socket from specified rooms.
	REMOTE_LEAVE

	// REMOTE_DISCONNECT instructs other nodes to disconnect a specific socket.
	REMOTE_DISCONNECT

	// REMOTE_FETCH requests detailed socket information from other nodes.
	REMOTE_FETCH

	// SERVER_SIDE_EMIT broadcasts a server-side event to other nodes.
	SERVER_SIDE_EMIT

	// BROADCAST sends a packet to clients across all nodes.
	BROADCAST

	// BROADCAST_CLIENT_COUNT reports the number of clients that will receive a broadcast.
	BROADCAST_CLIENT_COUNT

	// BROADCAST_ACK sends acknowledgement responses for broadcast operations.
	BROADCAST_ACK
)

Message types for Socket.IO Valkey adapter inter-node communication.

View Source 
const PrivateRoomIdLength = 20

PrivateRoomIdLength is the length of a socket ID, used to determine if a room is private.

View Source 
const VERSION = version.VERSION

Variables

View Source 
var ErrNilValkeyPacket = errors.New("cannot unmarshal into nil ValkeyPacket")

ErrNilValkeyPacket indicates an attempt to unmarshal into a nil ValkeyPacket.

View Source 
var ErrValkeyPubSubClosed = errors.New("valkey: pubsub closed")

ErrValkeyPubSubClosed is returned when receiving from a closed ValkeyPubSub.

Functions

func ShouldUseDynamicChannel 

func ShouldUseDynamicChannel(mode SubscriptionMode, room socket.Room) bool

ShouldUseDynamicChannel determines if a dynamic channel should be used for the given room.

Types

type Parser 

type Parser interface {
	// Encode serializes the given value into a byte slice.
	Encode(any) ([]byte, error)

	// Decode deserializes the byte slice into the given value.
	Decode([]byte, any) error
}

Parser defines the interface for encoding and decoding data for Valkey communication. Implementations must be thread-safe as they may be called from multiple goroutines.

type SubscriptionMode 

type SubscriptionMode string

SubscriptionMode determines how Valkey Pub/Sub channels are managed. This type is shared between the adapter and emitter packages. 

const (
	// StaticSubscriptionMode uses 2 fixed channels per namespace.
	StaticSubscriptionMode SubscriptionMode = "static"

	// DynamicSubscriptionMode uses 2 + 1 channel per public room per namespace.
	DynamicSubscriptionMode SubscriptionMode = "dynamic"

	// DynamicPrivateSubscriptionMode creates separate channels for both public and private rooms.
	DynamicPrivateSubscriptionMode SubscriptionMode = "dynamic-private"

	// DefaultSubscriptionMode is the default subscription mode.
	DefaultSubscriptionMode = DynamicSubscriptionMode
)

Subscription mode constants for Valkey adapter.

type ValkeyClient 

type ValkeyClient struct {
	types.EventEmitter

	// Client is the underlying Valkey client used for write operations
	// (PUBLISH, XADD, SET, etc.) and metadata queries (PUBSUB NUMSUB).
	Client vk.Client

	// SubClient is an optional separate Valkey client used for read/subscribe
	// operations (SUBSCRIBE, PSUBSCRIBE, SSUBSCRIBE, XREAD, XRANGE, etc.).
	// When nil, Client is used for all operations.
	//
	// Using a separate client for subscriptions prevents blocking read operations
	// from starving the write connection pool, and allows routing reads to
	// Valkey replicas for improved scalability.
	SubClient vk.Client

	// Context is the context used for Valkey operations.
	Context context.Context
}

ValkeyClient wraps a valkey-go client and provides context management and event emitting capabilities for the Socket.IO Valkey adapter.

The client supports read/write separation: Client is used for write operations (PUBLISH, XADD, SET, etc.) and SubClient is used for read/subscribe operations (SUBSCRIBE, XREAD, XRANGE, etc.). If SubClient is nil, Client is used for both.

func NewValkeyClient 

func NewValkeyClient(ctx context.Context, client vk.Client) *ValkeyClient

NewValkeyClient creates a new ValkeyClient with the given context and valkey-go client. The same client is used for both read and write operations.

Parameters:

  • ctx: The context that controls the lifecycle of Valkey operations. When canceled, all subscriptions and pending operations will be terminated.
  • client: A valkey-go Client instance.

Example: 

client, _ := valkey.NewClient(valkey.ClientOption{InitAddress: []string{"localhost:6379"}})
valkeyClient := NewValkeyClient(context.Background(), client)

func NewValkeyClientWithSub 

func NewValkeyClientWithSub(ctx context.Context, client, subClient vk.Client) *ValkeyClient

NewValkeyClientWithSub creates a new ValkeyClient with separate clients for read/write separation.

Parameters:

  • ctx: The context that controls the lifecycle of Valkey operations.
  • client: The Valkey client for write operations (PUBLISH, XADD, SET, etc.) and metadata queries (PUBSUB NUMSUB).
  • subClient: The Valkey client for read/subscribe operations (SUBSCRIBE, XREAD, etc.).

Example: 

pubClient, _ := valkey.NewClient(valkey.ClientOption{InitAddress: []string{"master:6379"}})
subClient, _ := valkey.NewClient(valkey.ClientOption{InitAddress: []string{"replica:6380"}})
valkeyClient := NewValkeyClientWithSub(context.Background(), pubClient, subClient)

func (*ValkeyClient) GetDel 

func (c *ValkeyClient) GetDel(ctx context.Context, key string) (string, error)

GetDel atomically gets and deletes a key. Returns ("", nil) if the key does not exist. Uses the read client (SubClient if set) for read/write separation.

func (*ValkeyClient) PSubscribe 

func (c *ValkeyClient) PSubscribe(ctx context.Context, patterns ...string) *ValkeyPubSub

PSubscribe creates a pattern-subscription on one or more Valkey patterns. The returned ValkeyPubSub delivers messages with Pattern and Channel set.

func (*ValkeyClient) PubSubNumSub 

func (c *ValkeyClient) PubSubNumSub(ctx context.Context, channels ...string) (map[string]int64, error)

PubSubNumSub returns the number of subscribers for each channel using PUBSUB NUMSUB.

func (*ValkeyClient) PubSubShardNumSub 

func (c *ValkeyClient) PubSubShardNumSub(ctx context.Context, channels ...string) (map[string]int64, error)

PubSubShardNumSub returns the subscriber count for sharded channels using PUBSUB SHARDNUMSUB.

func (*ValkeyClient) Publish 

func (c *ValkeyClient) Publish(ctx context.Context, channel string, message []byte) error

Publish publishes a message to a Valkey channel.

func (*ValkeyClient) SPublish 

func (c *ValkeyClient) SPublish(ctx context.Context, channel string, message []byte) error

SPublish publishes a message to a sharded Valkey channel (SPUBLISH).

func (*ValkeyClient) SSubscribe 

func (c *ValkeyClient) SSubscribe(ctx context.Context, channels ...string) *ValkeyPubSub

SSubscribe creates a sharded Pub/Sub subscription on one or more channels (SSUBSCRIBE). The returned ValkeyPubSub delivers messages via ReceiveMessage.

func (*ValkeyClient) Set 

func (c *ValkeyClient) Set(ctx context.Context, key, value string, expiry time.Duration) error

Set stores a string value at key with an expiry duration.

func (*ValkeyClient) Sub 

func (c *ValkeyClient) Sub() vk.Client

Sub returns the Valkey client to use for read/subscribe operations. If SubClient is set, it is returned; otherwise Client is used as the fallback.

func (*ValkeyClient) Subscribe 

func (c *ValkeyClient) Subscribe(ctx context.Context, channels ...string) *ValkeyPubSub

Subscribe creates a channel-subscription on one or more Valkey channels. The returned ValkeyPubSub delivers messages via ReceiveMessage.

func (*ValkeyClient) XAdd 

func (c *ValkeyClient) XAdd(ctx context.Context, stream string, maxLen int64, values map[string]any) (string, error)

XAdd appends a message to a Valkey stream and returns the generated entry ID. maxLen is used with the approximate trimming operator (~) for performance.

func (*ValkeyClient) XRange 

func (c *ValkeyClient) XRange(ctx context.Context, stream, start, stop string) ([]vk.XRangeEntry, error)

XRange reads a range of entries from a Valkey stream. Uses the read client (SubClient if set) for read/write separation.

func (*ValkeyClient) XRangeN 

func (c *ValkeyClient) XRangeN(ctx context.Context, stream, start, stop string, count int64) ([]vk.XRangeEntry, error)

XRangeN reads a limited range of entries from a Valkey stream. Uses the read client (SubClient if set) for read/write separation.

func (*ValkeyClient) XRead 

func (c *ValkeyClient) XRead(ctx context.Context, streams []string, id string, count int64, block time.Duration) ([]vk.XRangeEntry, error)

XRead reads messages from one or more Valkey streams, blocking up to the given duration. Returns entries from the first stream that has data, or nil if the timeout is reached. Uses the read client (SubClient if set) for read/write separation.

type ValkeyMessage 

type ValkeyMessage struct {
	// Pattern is non-empty for pattern-subscribed messages (PSUBSCRIBE).
	Pattern string
	// Channel is the channel the message was published on.
	Channel string
	// Payload is the raw message payload.
	Payload string
}

ValkeyMessage represents a single Pub/Sub message received from Valkey.

type ValkeyPacket 

type ValkeyPacket struct {

	// Uid identifies the source server that sent this packet.
	Uid adapter.ServerId `json:"-"`

	// Packet is the Socket.IO packet to be broadcast.
	Packet *parser.Packet `json:"-"`

	// Opts contains the broadcast options including target rooms and exclusions.
	Opts *adapter.PacketOptions `json:"-"`
	// contains filtered or unexported fields
}

ValkeyPacket represents a packet to be broadcast via Valkey. It contains the server UID, the Socket.IO packet, and broadcast options.

func (*ValkeyPacket) MarshalJSON 

func (r *ValkeyPacket) MarshalJSON() ([]byte, error)

MarshalJSON implements the json.Marshaler interface for ValkeyPacket. It serializes the ValkeyPacket as a JSON array in the format [Uid, Packet, Opts].

func (*ValkeyPacket) UnmarshalJSON 

func (r *ValkeyPacket) UnmarshalJSON(data []byte) error

UnmarshalJSON implements the json.Unmarshaler interface for ValkeyPacket. It deserializes a JSON array [Uid, Packet?, Opts?] back into the ValkeyPacket struct. The Uid field is required; Packet and Opts are optional.

type ValkeyPubSub 

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

ValkeyPubSub wraps a valkey-go dedicated client Pub/Sub subscription into a channel-based interface that mirrors the go-redis *PubSub API.

Unlike the previous implementation which used Client.Receive (owning the connection and only supporting context cancellation), this uses Client.Dedicate + SetPubSubHooks, giving a persistent dedicated connection where SUBSCRIBE/UNSUBSCRIBE commands can be issued independently.

func (*ValkeyPubSub) Close 

func (p *ValkeyPubSub) Close() error

Close releases the dedicated connection and stops message delivery.

func (*ValkeyPubSub) PUnsubscribe 

func (p *ValkeyPubSub) PUnsubscribe(ctx context.Context, patterns ...string) error

PUnsubscribe issues a PUNSUBSCRIBE command for the given patterns on the dedicated connection, without tearing down the subscription stream.

func (*ValkeyPubSub) ReceiveMessage 

func (p *ValkeyPubSub) ReceiveMessage(ctx context.Context) (*ValkeyMessage, error)

ReceiveMessage blocks until a message is available or the context is done.

func (*ValkeyPubSub) SUnsubscribe 

func (p *ValkeyPubSub) SUnsubscribe(ctx context.Context, channels ...string) error

SUnsubscribe issues a SUNSUBSCRIBE command for the given channels on the dedicated connection, without tearing down the subscription stream.

func (*ValkeyPubSub) Unsubscribe 

func (p *ValkeyPubSub) Unsubscribe(ctx context.Context, channels ...string) error

Unsubscribe issues an UNSUBSCRIBE command for the given channels on the dedicated connection, without tearing down the subscription stream.

type ValkeyRequest 

type ValkeyRequest struct {
	Type      adapter.MessageType    `json:"type,omitempty" msgpack:"type,omitempty"`
	RequestId string                 `json:"requestId,omitempty" msgpack:"requestId,omitempty"`
	Rooms     []socket.Room          `json:"rooms,omitempty" msgpack:"rooms,omitempty"`
	Opts      *adapter.PacketOptions `json:"opts,omitempty" msgpack:"opts,omitempty"`
	Sid       socket.SocketId        `json:"sid,omitempty" msgpack:"sid,omitempty"`
	Room      socket.Room            `json:"room,omitempty" msgpack:"room,omitempty"`
	Close     bool                   `json:"close,omitempty" msgpack:"close,omitempty"`
	Uid       adapter.ServerId       `json:"uid,omitempty" msgpack:"uid,omitempty"`
	Data      []any                  `json:"data,omitempty" msgpack:"data,omitempty"`
	Packet    *parser.Packet         `json:"packet,omitempty" msgpack:"packet,omitempty"`
}

ValkeyRequest represents a request message sent between servers via Valkey. It is used for various inter-node operations such as remote joins, leaves, and fetches.

type ValkeyResponse 

type ValkeyResponse struct {
	Type        adapter.MessageType       `json:"type,omitempty" msgpack:"type,omitempty"`
	RequestId   string                    `json:"requestId,omitempty" msgpack:"requestId,omitempty"`
	Rooms       []socket.Room             `json:"rooms,omitempty" msgpack:"rooms,omitempty"`
	Sockets     []*adapter.SocketResponse `json:"sockets,omitempty" msgpack:"sockets,omitempty"`
	Data        []any                     `json:"data,omitempty" msgpack:"data,omitempty"`
	ClientCount uint64                    `json:"clientcount,omitempty" msgpack:"clientcount,omitempty"`
	Packet      []any                     `json:"packet,omitempty" msgpack:"packet,omitempty"`
}

ValkeyResponse represents a response message sent between servers via Valkey. It contains the response data for various inter-node requests.

Source Files

View all Source files

Directories

Path Synopsis
Package adapter provides configuration options for the Valkey sharded Pub/Sub adapter for Socket.IO.
 Package adapter provides configuration options for the Valkey sharded Pub/Sub adapter for Socket.IO.
Package emitter provides broadcast capabilities for Socket.IO via Valkey pub/sub.
 Package emitter provides broadcast capabilities for Socket.IO via Valkey pub/sub.

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.