adapter

package
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: 23 Imported by: 0

Documentation

Overview

Package adapter provides configuration options for the Redis-based Socket.IO adapter.

Package adapter defines types and interfaces for the Redis-based Socket.IO adapter implementation.

Package adapter provides a Redis-based adapter implementation for Socket.IO clustering.

Package adapter provides configuration options for the Redis sharded Pub/Sub adapter for Socket.IO. The sharded adapter leverages Redis 7.0's sharded Pub/Sub for improved scalability.

Package adapter defines the interface for the Redis sharded Pub/Sub adapter for Socket.IO. This adapter leverages Redis 7.0's sharded Pub/Sub for improved scalability in clustered environments.

Package adapter implements a Redis sharded Pub/Sub adapter for Socket.IO clustering.

This adapter uses Redis 7.0 sharded Pub/Sub, which distributes channels across Redis Cluster slots for improved horizontal scalability.

See: https://redis.io/docs/manual/pubsub/#sharded-pubsub

Package adapter provides configuration options for the Redis Streams-based Socket.IO adapter. The streams adapter uses Redis Streams for message persistence and session recovery.

Package adapter defines types and interfaces for the Redis Streams-based Socket.IO adapter. Redis Streams provide message persistence and enable session recovery across server restarts.

Package adapter implements a Redis Streams-based adapter for Socket.IO clustering. Redis Streams provide message persistence and enable session recovery across server restarts. Ephemeral messages (fetchSockets, serverSideEmit, broadcastWithAck) are sent via Redis PUB/SUB for compatibility with the Node.js @socket.io/redis-streams-adapter package.

Index

Constants

View Source 
const (
	DefaultStreamName       = "socket.io"
	DefaultStreamMaxLen     = 10_000
	DefaultStreamReadCount  = 100
	DefaultStreamCount      = 1
	DefaultChannelPrefix    = "socket.io"
	DefaultBlockTimeInMs    = 5_000
	DefaultSessionKeyPrefix = "sio:session:"
)

Default configuration values for RedisStreamsAdapterOptions.

View Source 
const (
	// DefaultRequestsTimeout is the default timeout for inter-node requests.
	DefaultRequestsTimeout = 5000 * time.Millisecond
)

Default configuration values for RedisAdapterOptions.

View Source 
const (
	DefaultShardedChannelPrefix = "socket.io"
)

Default configuration values for ShardedRedisAdapterOptions.

Variables

View Source 
var DefaultShardedSubscriptionMode = redis.DynamicSubscriptionMode

DefaultShardedSubscriptionMode is the default subscription mode for the sharded adapter.

Functions

This section is empty.

Types

type AckRequest 

type AckRequest = adapter.ClusterAckRequest

AckRequest is an alias for adapter.ClusterAckRequest, used for acknowledgement tracking.

type Packet 

type Packet = redis.RedisPacket

Packet is an alias for redis.RedisPacket, representing a broadcast packet sent via Redis.

type RawClusterMessage 

type RawClusterMessage map[string]any

RawClusterMessage represents a raw message from the Redis stream. It is a map of string keys to any values, matching the Redis XREAD output format.

func (RawClusterMessage) Data 

func (r RawClusterMessage) Data() string

Data returns the data field from the raw cluster message. Returns empty string if the field is missing or not a string.

func (RawClusterMessage) Nsp 

func (r RawClusterMessage) Nsp() string

Nsp returns the namespace from the raw cluster message. Returns empty string if the field is missing or not a string.

func (RawClusterMessage) Type 

func (r RawClusterMessage) Type() string

Type returns the message type from the raw cluster message. Returns empty string if the field is missing or not a string.

func (RawClusterMessage) Uid 

func (r RawClusterMessage) Uid() string

Uid returns the UID (unique identifier) from the raw cluster message. Returns empty string if the field is missing or not a string.

type RedisAdapter 

type RedisAdapter interface {
	socket.Adapter

	// SetRedis configures the Redis client for the adapter.
	SetRedis(*redis.RedisClient)

	// SetOpts configures adapter options.
	SetOpts(any)

	// Uid returns the unique server identifier for this adapter instance.
	Uid() adapter.ServerId

	// RequestsTimeout returns the configured timeout for inter-node requests.
	RequestsTimeout() time.Duration

	// PublishOnSpecificResponseChannel indicates if responses use node-specific channels.
	PublishOnSpecificResponseChannel() bool

	// Parser returns the parser used for message encoding/decoding.
	Parser() redis.Parser

	// AllRooms returns a function to retrieve all rooms across the cluster.
	AllRooms() func(func(*types.Set[socket.Room], error))
}

RedisAdapter defines the interface for a Redis-based Socket.IO adapter. It extends the base socket.Adapter with Redis-specific functionality.

func MakeRedisAdapter 

func MakeRedisAdapter() RedisAdapter

MakeRedisAdapter creates a new uninitialized redisAdapter with default options. Call Construct() to complete initialization.

func NewRedisAdapter 

func NewRedisAdapter(nsp socket.Namespace, redisClient *redis.RedisClient, opts any) RedisAdapter

NewRedisAdapter creates and initializes a new RedisAdapter for the given namespace. This is the primary constructor for creating Redis adapters.

type RedisAdapterBuilder 

type RedisAdapterBuilder struct {
	// Redis is the Redis client used by the adapter for Pub/Sub communication.
	Redis *redis.RedisClient
	// Opts contains configuration options for the adapter.
	Opts RedisAdapterOptionsInterface
}

RedisAdapterBuilder builds a RedisAdapter with the given Redis client and options. Use this builder to create adapters for Socket.IO namespaces.

func (*RedisAdapterBuilder) New 

func (rb *RedisAdapterBuilder) New(nsp socket.Namespace) socket.Adapter

New creates a new RedisAdapter for the given namespace. This method implements the socket.AdapterBuilder interface.

type RedisAdapterOptions 

type RedisAdapterOptions struct {
	emitter.EmitterOptions
	// contains filtered or unexported fields
}

RedisAdapterOptions holds configuration for the Redis adapter.

Fields:

  • requestsTimeout: Maximum time to wait for responses to inter-node requests. Default: 5000ms. After this timeout, the adapter stops waiting for responses.
  • publishOnSpecificResponseChannel: When true, responses are published to a channel specific to the requesting node, reducing unnecessary message processing. Default: false.

func DefaultRedisAdapterOptions 

func DefaultRedisAdapterOptions() *RedisAdapterOptions

DefaultRedisAdapterOptions returns a new RedisAdapterOptions with default values.

func (*RedisAdapterOptions) Assign 

func (s *RedisAdapterOptions) Assign(data RedisAdapterOptionsInterface) RedisAdapterOptionsInterface

Assign copies non-nil fields from another RedisAdapterOptionsInterface. This method is useful for merging user-provided options with defaults.

func (*RedisAdapterOptions) GetRawPublishOnSpecificResponseChannel 

func (s *RedisAdapterOptions) GetRawPublishOnSpecificResponseChannel() types.Optional[bool]

GetRawPublishOnSpecificResponseChannel returns the raw Optional value. Returns nil if not explicitly set.

func (*RedisAdapterOptions) GetRawRequestsTimeout 

func (s *RedisAdapterOptions) GetRawRequestsTimeout() types.Optional[time.Duration]

GetRawRequestsTimeout returns the raw Optional value for requestsTimeout. Returns nil if not explicitly set.

func (*RedisAdapterOptions) PublishOnSpecificResponseChannel 

func (s *RedisAdapterOptions) PublishOnSpecificResponseChannel() bool

PublishOnSpecificResponseChannel returns whether responses should be published to a node-specific channel. Returns false if not explicitly set.

func (*RedisAdapterOptions) RequestsTimeout 

func (s *RedisAdapterOptions) RequestsTimeout() time.Duration

RequestsTimeout returns the configured requests timeout. Returns 0 if not explicitly set; callers should use DefaultRequestsTimeout as fallback.

func (*RedisAdapterOptions) SetPublishOnSpecificResponseChannel 

func (s *RedisAdapterOptions) SetPublishOnSpecificResponseChannel(publishOnSpecificResponseChannel bool)

SetPublishOnSpecificResponseChannel sets whether responses should be published to a node-specific channel.

func (*RedisAdapterOptions) SetRequestsTimeout 

func (s *RedisAdapterOptions) SetRequestsTimeout(requestsTimeout time.Duration)

SetRequestsTimeout sets the timeout duration for inter-node requests.

type RedisAdapterOptionsInterface 

type RedisAdapterOptionsInterface interface {
	emitter.EmitterOptionsInterface

	SetRequestsTimeout(time.Duration)
	GetRawRequestsTimeout() types.Optional[time.Duration]
	RequestsTimeout() time.Duration

	SetPublishOnSpecificResponseChannel(bool)
	GetRawPublishOnSpecificResponseChannel() types.Optional[bool]
	PublishOnSpecificResponseChannel() bool
}

RedisAdapterOptionsInterface defines the interface for configuring RedisAdapterOptions. It extends EmitterOptionsInterface to include adapter-specific settings.

type RedisRequest 

type RedisRequest struct {
	// Type identifies the message/request type.
	Type adapter.MessageType

	// Resolve is the callback invoked when the request completes successfully.
	Resolve func(*types.Slice[any])

	// Timeout is the timer for request timeout handling.
	Timeout *atomic.Pointer[utils.Timer]

	// NumSub is the number of expected responses from other nodes.
	NumSub int64

	// MsgCount tracks the number of responses received.
	MsgCount *atomic.Int64

	// Rooms accumulates room information from responses.
	Rooms *types.Set[socket.Room]

	// Sockets accumulates socket information from responses.
	Sockets *types.Slice[*adapter.SocketResponse]

	// Responses accumulates generic response data.
	Responses *types.Slice[any]

	// Once ensures that the Resolve callback and cleanup are executed exactly once,
	// preventing double invocations from both normal completion and timeout paths.
	Once sync.Once
}

RedisRequest represents an internal request tracker with state management. It extends the base RedisRequest with fields for tracking request lifecycle.

type RedisStreamsAdapter 

type RedisStreamsAdapter interface {
	adapter.ClusterAdapter

	// SetRedis configures the Redis client for the adapter.
	SetRedis(*redis.RedisClient)

	// SetOpts sets the configuration options for the streams adapter.
	SetOpts(any)

	// Cleanup registers a cleanup callback to be called when the adapter is closed.
	Cleanup(func())

	// OnRawMessage processes a raw message from the Redis stream.
	OnRawMessage(RawClusterMessage, string) error
}

RedisStreamsAdapter defines the interface for a Redis Streams-based Socket.IO adapter. It extends ClusterAdapter with Redis Streams-specific functionality and PUB/SUB support for ephemeral messages, matching the Node.js implementation.

func MakeRedisStreamsAdapter 

func MakeRedisStreamsAdapter() RedisStreamsAdapter

MakeRedisStreamsAdapter creates a new uninitialized redisStreamsAdapter. Call Construct() to complete initialization before use.

func NewRedisStreamsAdapter 

func NewRedisStreamsAdapter(nsp socket.Namespace, client *redis.RedisClient, opts any) RedisStreamsAdapter

NewRedisStreamsAdapter creates and initializes a new Redis Streams adapter. This is the preferred way to create a streams adapter instance.

type RedisStreamsAdapterBuilder 

type RedisStreamsAdapterBuilder struct {
	// Redis is the Redis client used for stream operations.
	Redis *redis.RedisClient
	// Opts contains configuration options for the streams adapter.
	Opts RedisStreamsAdapterOptionsInterface
	// contains filtered or unexported fields
}

RedisStreamsAdapterBuilder creates Redis Streams adapters for Socket.IO namespaces. It manages the shared polling loops and PUB/SUB subscriptions across all namespace adapters.

func (*RedisStreamsAdapterBuilder) New 

func (sb *RedisStreamsAdapterBuilder) New(nsp socket.Namespace) socket.Adapter

New creates a new Redis Streams adapter for the given namespace. This method implements the socket.AdapterBuilder interface.

type RedisStreamsAdapterOptions 

type RedisStreamsAdapterOptions struct {
	adapter.ClusterAdapterOptions
	// contains filtered or unexported fields
}

RedisStreamsAdapterOptions holds configuration for the Redis Streams adapter.

func DefaultRedisStreamsAdapterOptions 

func DefaultRedisStreamsAdapterOptions() *RedisStreamsAdapterOptions

DefaultRedisStreamsAdapterOptions returns a new RedisStreamsAdapterOptions with default values.

func (*RedisStreamsAdapterOptions) Assign 

func (s *RedisStreamsAdapterOptions) Assign(data RedisStreamsAdapterOptionsInterface) RedisStreamsAdapterOptionsInterface

Assign copies non-nil fields from another RedisStreamsAdapterOptionsInterface. This method is useful for merging user-provided options with defaults.

func (*RedisStreamsAdapterOptions) BlockTimeInMs 

func (s *RedisStreamsAdapterOptions) BlockTimeInMs() int64

func (*RedisStreamsAdapterOptions) ChannelPrefix 

func (s *RedisStreamsAdapterOptions) ChannelPrefix() string

func (*RedisStreamsAdapterOptions) GetRawBlockTimeInMs 

func (s *RedisStreamsAdapterOptions) GetRawBlockTimeInMs() types.Optional[int64]

func (*RedisStreamsAdapterOptions) GetRawChannelPrefix 

func (s *RedisStreamsAdapterOptions) GetRawChannelPrefix() types.Optional[string]

func (*RedisStreamsAdapterOptions) GetRawMaxLen 

func (s *RedisStreamsAdapterOptions) GetRawMaxLen() types.Optional[int64]

func (*RedisStreamsAdapterOptions) GetRawOnlyPlaintext 

func (s *RedisStreamsAdapterOptions) GetRawOnlyPlaintext() types.Optional[bool]

func (*RedisStreamsAdapterOptions) GetRawReadCount 

func (s *RedisStreamsAdapterOptions) GetRawReadCount() types.Optional[int64]

func (*RedisStreamsAdapterOptions) GetRawSessionKeyPrefix 

func (s *RedisStreamsAdapterOptions) GetRawSessionKeyPrefix() types.Optional[string]

func (*RedisStreamsAdapterOptions) GetRawStreamCount 

func (s *RedisStreamsAdapterOptions) GetRawStreamCount() types.Optional[int]

func (*RedisStreamsAdapterOptions) GetRawStreamName 

func (s *RedisStreamsAdapterOptions) GetRawStreamName() types.Optional[string]

func (*RedisStreamsAdapterOptions) GetRawUseShardedPubSub 

func (s *RedisStreamsAdapterOptions) GetRawUseShardedPubSub() types.Optional[bool]

func (*RedisStreamsAdapterOptions) MaxLen 

func (s *RedisStreamsAdapterOptions) MaxLen() int64

func (*RedisStreamsAdapterOptions) OnlyPlaintext 

func (s *RedisStreamsAdapterOptions) OnlyPlaintext() bool

func (*RedisStreamsAdapterOptions) ReadCount 

func (s *RedisStreamsAdapterOptions) ReadCount() int64

func (*RedisStreamsAdapterOptions) SessionKeyPrefix 

func (s *RedisStreamsAdapterOptions) SessionKeyPrefix() string

func (*RedisStreamsAdapterOptions) SetBlockTimeInMs 

func (s *RedisStreamsAdapterOptions) SetBlockTimeInMs(blockTimeInMs int64)

func (*RedisStreamsAdapterOptions) SetChannelPrefix 

func (s *RedisStreamsAdapterOptions) SetChannelPrefix(channelPrefix string)

func (*RedisStreamsAdapterOptions) SetMaxLen 

func (s *RedisStreamsAdapterOptions) SetMaxLen(maxLen int64)

func (*RedisStreamsAdapterOptions) SetOnlyPlaintext 

func (s *RedisStreamsAdapterOptions) SetOnlyPlaintext(onlyPlaintext bool)

func (*RedisStreamsAdapterOptions) SetReadCount 

func (s *RedisStreamsAdapterOptions) SetReadCount(readCount int64)

func (*RedisStreamsAdapterOptions) SetSessionKeyPrefix 

func (s *RedisStreamsAdapterOptions) SetSessionKeyPrefix(sessionKeyPrefix string)

func (*RedisStreamsAdapterOptions) SetStreamCount 

func (s *RedisStreamsAdapterOptions) SetStreamCount(streamCount int)

func (*RedisStreamsAdapterOptions) SetStreamName 

func (s *RedisStreamsAdapterOptions) SetStreamName(streamName string)

func (*RedisStreamsAdapterOptions) SetUseShardedPubSub 

func (s *RedisStreamsAdapterOptions) SetUseShardedPubSub(useShardedPubSub bool)

func (*RedisStreamsAdapterOptions) StreamCount 

func (s *RedisStreamsAdapterOptions) StreamCount() int

func (*RedisStreamsAdapterOptions) StreamName 

func (s *RedisStreamsAdapterOptions) StreamName() string

func (*RedisStreamsAdapterOptions) UseShardedPubSub 

func (s *RedisStreamsAdapterOptions) UseShardedPubSub() bool

type RedisStreamsAdapterOptionsInterface 

type RedisStreamsAdapterOptionsInterface interface {
	adapter.ClusterAdapterOptionsInterface

	SetStreamName(string)
	GetRawStreamName() types.Optional[string]
	StreamName() string

	SetStreamCount(int)
	GetRawStreamCount() types.Optional[int]
	StreamCount() int

	SetChannelPrefix(string)
	GetRawChannelPrefix() types.Optional[string]
	ChannelPrefix() string

	SetUseShardedPubSub(bool)
	GetRawUseShardedPubSub() types.Optional[bool]
	UseShardedPubSub() bool

	SetMaxLen(int64)
	GetRawMaxLen() types.Optional[int64]
	MaxLen() int64

	SetReadCount(int64)
	GetRawReadCount() types.Optional[int64]
	ReadCount() int64

	SetBlockTimeInMs(int64)
	GetRawBlockTimeInMs() types.Optional[int64]
	BlockTimeInMs() int64

	SetSessionKeyPrefix(string)
	GetRawSessionKeyPrefix() types.Optional[string]
	SessionKeyPrefix() string

	SetOnlyPlaintext(bool)
	GetRawOnlyPlaintext() types.Optional[bool]
	OnlyPlaintext() bool
}

RedisStreamsAdapterOptionsInterface defines the interface for configuring RedisStreamsAdapterOptions. It extends ClusterAdapterOptionsInterface with streams-specific settings.

type Request 

type Request = redis.RedisRequest

Request is an alias for redis.RedisRequest, representing an inter-node request.

type Response 

type Response = redis.RedisResponse

Response is an alias for redis.RedisResponse, representing an inter-node response.

type ShardedRedisAdapter 

type ShardedRedisAdapter interface {
	adapter.ClusterAdapter

	// SetRedis configures the Redis client for the adapter.
	SetRedis(*redis.RedisClient)

	// SetOpts configures adapter options.
	SetOpts(any)
}

ShardedRedisAdapter defines the interface for a sharded Redis-based Socket.IO adapter. It extends ClusterAdapter with Redis-specific configuration methods.

func MakeShardedRedisAdapter 

func MakeShardedRedisAdapter() ShardedRedisAdapter

MakeShardedRedisAdapter creates a new uninitialized shardedRedisAdapter. Call Construct to complete initialization.

func NewShardedRedisAdapter 

func NewShardedRedisAdapter(nsp socket.Namespace, redisClient *redis.RedisClient, opts any) ShardedRedisAdapter

NewShardedRedisAdapter creates and fully initializes a sharded Redis adapter.

type ShardedRedisAdapterBuilder 

type ShardedRedisAdapterBuilder struct {
	// Redis is the Redis client used for sharded Pub/Sub communication.
	Redis *redis.RedisClient
	// Opts contains configuration options for the adapter.
	Opts ShardedRedisAdapterOptionsInterface
}

ShardedRedisAdapterBuilder creates sharded Redis adapters for Socket.IO namespaces.

func (*ShardedRedisAdapterBuilder) New 

func (sb *ShardedRedisAdapterBuilder) New(nsp socket.Namespace) socket.Adapter

New creates a new sharded Redis adapter for the given namespace. It implements the socket.AdapterBuilder interface.

type ShardedRedisAdapterOptions 

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

ShardedRedisAdapterOptions holds configuration for the sharded Redis adapter.

Fields:

  • channelPrefix: The prefix for Redis Pub/Sub channels. Default: "socket.io".
  • subscriptionMode: Determines the channel management strategy. Default: DynamicSubscriptionMode.

func DefaultShardedRedisAdapterOptions 

func DefaultShardedRedisAdapterOptions() *ShardedRedisAdapterOptions

DefaultShardedRedisAdapterOptions returns a new ShardedRedisAdapterOptions with default values.

func (*ShardedRedisAdapterOptions) Assign 

func (s *ShardedRedisAdapterOptions) Assign(data ShardedRedisAdapterOptionsInterface) ShardedRedisAdapterOptionsInterface

Assign copies non-nil fields from another ShardedRedisAdapterOptionsInterface. This method is useful for merging user-provided options with defaults.

func (*ShardedRedisAdapterOptions) ChannelPrefix 

func (s *ShardedRedisAdapterOptions) ChannelPrefix() string

ChannelPrefix returns the configured channel prefix. Returns empty string if not set; callers should use DefaultShardedChannelPrefix as fallback.

func (*ShardedRedisAdapterOptions) GetRawChannelPrefix 

func (s *ShardedRedisAdapterOptions) GetRawChannelPrefix() types.Optional[string]

GetRawChannelPrefix returns the raw Optional value for channelPrefix.

func (*ShardedRedisAdapterOptions) GetRawSubscriptionMode 

func (s *ShardedRedisAdapterOptions) GetRawSubscriptionMode() types.Optional[redis.SubscriptionMode]

GetRawSubscriptionMode returns the raw Optional value for subscriptionMode.

func (*ShardedRedisAdapterOptions) SetChannelPrefix 

func (s *ShardedRedisAdapterOptions) SetChannelPrefix(channelPrefix string)

SetChannelPrefix sets the channel prefix for Redis Pub/Sub.

func (*ShardedRedisAdapterOptions) SetSubscriptionMode 

func (s *ShardedRedisAdapterOptions) SetSubscriptionMode(subscriptionMode redis.SubscriptionMode)

SetSubscriptionMode sets the subscription mode for channel management.

func (*ShardedRedisAdapterOptions) SubscriptionMode 

func (s *ShardedRedisAdapterOptions) SubscriptionMode() redis.SubscriptionMode

SubscriptionMode returns the configured subscription mode. Returns empty string if not set; callers should use DefaultShardedSubscriptionMode as fallback.

type ShardedRedisAdapterOptionsInterface 

type ShardedRedisAdapterOptionsInterface interface {
	SetChannelPrefix(string)
	GetRawChannelPrefix() types.Optional[string]
	ChannelPrefix() string

	SetSubscriptionMode(redis.SubscriptionMode)
	GetRawSubscriptionMode() types.Optional[redis.SubscriptionMode]
	SubscriptionMode() redis.SubscriptionMode
}

ShardedRedisAdapterOptionsInterface defines the interface for configuring ShardedRedisAdapterOptions.

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.