redis

README

socket.io-go-redis

Description

A Redis adapter for Socket.IO server in Go, allowing to scale Socket.IO applications across multiple processes or servers.

Installation

go get github.com/zishang520/socket.io/adapters/redis/v3

Features

• Multiple servers support
• Real-time communication between processes
• Automatic reconnection
• Custom Redis configuration
• Event emission across servers

How to use

Basic usage example:

package main

import (
    "context"
    "fmt"
    "os"
    "os/signal"
    "syscall"

    rds "github.com/redis/go-redis/v9"
    "github.com/zishang520/socket.io/adapters/redis/v3"
    "github.com/zishang520/socket.io/adapters/redis/v3/adapter"
    "github.com/zishang520/socket.io/servers/socket/v3"
)

func main() {
    // Initialize Redis client
    redisClient := redis.NewRedisClient(context.TODO(), rds.NewClient(&rds.Options{
        Addr:     "127.0.0.1:6379",
        Username: "",
        Password: "",
        DB:       0,
    }))

    // Redis error handling
    redisClient.On("error", func(a ...any) {
        fmt.Println(a)
    })

    // Socket.IO server configuration
    config := socket.DefaultServerOptions()
    config.SetAdapter(&adapter.RedisAdapterBuilder{
        Redis: redisClient,
        Opts:  &adapter.RedisAdapterOptions{},
    })

    // Create and configure server
    httpServer := s.CreateServer(nil)
    io := socket.NewServer(httpServer, config)

    // Handle socket connections
    io.On("connection", func(clients ...any) {
        client := clients[0].(*socket.Socket)
        client.On("event", func(datas ...any) {
            // Handle your events here
        })
        client.On("disconnect", func(...any) {
            // Handle disconnect
        })
    })

    // Start server
    httpServer.Listen("127.0.0.1:9000", nil)

    // Graceful shutdown handling
    exit := make(chan struct{})
    SignalC := make(chan os.Signal)

    signal.Notify(SignalC, os.Interrupt, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
    go func() {
        for s := range SignalC {
            switch s {
            case os.Interrupt, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT:
                close(exit)
                return
            }
        }
    }()

    <-exit
    httpServer.Close(nil)
    os.Exit(0)
}

Configuration Options

The Redis adapter accepts the following options:

type RedisAdapterOptions struct {
    Prefix  string // Optional prefix for Redis keys
    // Add other available options here
}

Testing

Run the test suite with:

make test

Contributing

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add some amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

Support

If you encounter any issues or have questions, please file them in the issues section.

License

This project is licensed under the MIT License - see the LICENSE file for details.

Documentation

Overview

Package redis defines constants for Redis-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 redis provides Redis client wrapper for Socket.IO Redis adapter. This package offers a unified interface for Redis operations with event handling support.

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

Index

• Constants
• Variables
• func ShouldUseDynamicChannel(mode SubscriptionMode, room socket.Room) bool
• type Parser
• type RedisClient
  • func NewRedisClient(ctx context.Context, client redis.UniversalClient) *RedisClient
  • func NewRedisClientWithSub(ctx context.Context, client, subClient redis.UniversalClient) *RedisClient
  • func (r *RedisClient) Sub() redis.UniversalClient
• type RedisPacket
  • func (r *RedisPacket) MarshalJSON() ([]byte, error)
  • func (r *RedisPacket) UnmarshalJSON(data []byte) error
• type RedisRequest
• type RedisResponse
• type SubscriptionMode

Constants

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 Redis adapter inter-node communication. These constants define the different operations that can be performed across multiple Socket.IO server nodes using Redis as the message broker.

const PrivateRoomIdLength = 20

privateRoomIdLength is the length of a socket ID, used to determine if a room is private. Private rooms (socket IDs) have exactly this length.

const VERSION = version.VERSION

Variables

var ErrNilRedisPacket = errors.New("cannot unmarshal into nil RedisPacket")

ErrNilRedisPacket indicates an attempt to unmarshal into a nil RedisPacket.

Functions

func ShouldUseDynamicChannel

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

ShouldUseDynamicChannel determines if a dynamic channel should be used for the given room. This function is shared between the adapter and emitter packages to ensure consistent behavior.

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 Redis communication. Implementations must be thread-safe as they may be called from multiple goroutines.

type RedisClient

type RedisClient struct {
	types.EventEmitter

	// Client is the underlying Redis universal client used for write operations
	// (PUBLISH, XADD, SET, etc.) and metadata queries (PUBSUB NUMSUB).
	// It supports both standalone and cluster Redis deployments.
	Client redis.UniversalClient

	// SubClient is an optional separate Redis universal 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
	// Redis replicas for improved scalability.
	SubClient redis.UniversalClient

	// Context is the context used for Redis operations.
	// This context controls the lifecycle of Redis subscriptions and operations.
	Context context.Context
}

RedisClient wraps a Redis UniversalClient and provides context management and event emitting capabilities for the Socket.IO Redis 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.

The client supports error event emission, which allows higher-level components to handle Redis-related errors gracefully.

func NewRedisClient

func NewRedisClient(ctx context.Context, client redis.UniversalClient) *RedisClient

NewRedisClient creates a new RedisClient with the given context and Redis universal client. The same client is used for both read and write operations.

Parameters:

• ctx: The context that controls the lifecycle of Redis operations. When canceled, all subscriptions and pending operations will be terminated.
• client: A Redis UniversalClient instance that handles the actual Redis communication.

Returns:

• A pointer to the initialized RedisClient instance.

Example:

client := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
redisClient := NewRedisClient(context.Background(), client)

func NewRedisClientWithSub

func NewRedisClientWithSub(ctx context.Context, client, subClient redis.UniversalClient) *RedisClient

NewRedisClientWithSub creates a new RedisClient with separate clients for read/write separation.

Parameters:

• ctx: The context that controls the lifecycle of Redis operations.
• client: The Redis client for write operations (PUBLISH, XADD, SET, etc.) and metadata queries (PUBSUB NUMSUB).
• subClient: The Redis client for read/subscribe operations (SUBSCRIBE, XREAD, etc.). Both clients should connect to the same Redis deployment.

Example:

pubClient := redis.NewClient(&redis.Options{Addr: "master:6379"})
subClient := redis.NewClient(&redis.Options{Addr: "replica:6380"})
redisClient := NewRedisClientWithSub(context.Background(), pubClient, subClient)

func (*RedisClient) Sub

func (r *RedisClient) Sub() redis.UniversalClient

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

type RedisPacket

type RedisPacket 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
}

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

func (*RedisPacket) MarshalJSON

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

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

func (*RedisPacket) UnmarshalJSON

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

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

type RedisRequest

type RedisRequest 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"`
}

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

type RedisResponse

type RedisResponse 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"`
}

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

type SubscriptionMode

type SubscriptionMode string

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

const (
	// StaticSubscriptionMode uses 2 fixed channels per namespace.
	// This mode is simpler but may have higher message overhead for large deployments.
	StaticSubscriptionMode SubscriptionMode = "static"

	// DynamicSubscriptionMode uses 2 + 1 channel per public room per namespace.
	// This optimizes message routing for public rooms but excludes private rooms (socket IDs).
	DynamicSubscriptionMode SubscriptionMode = "dynamic"

	// DynamicPrivateSubscriptionMode creates separate channels for both public and private rooms.
	// This provides the finest granularity but uses the most Redis resources.
	DynamicPrivateSubscriptionMode SubscriptionMode = "dynamic-private"

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

Subscription mode constants for Redis adapter.