socketio

README

SocketIO v4 and EngineIO v4 server implementation for Go

Note: implementation is not feature-complete. For example only default namespace is supported.

This repo provides server implementation for SocketIO v4 and EngineIO v4.

Implementation provided is used in PleaseTalk project.

Note: Only websocket is supported as transport.

const socket = io("https://example.com", { transports: ["websocket"] });

Usage

Basic example

package main

import (
    "net/http"

    "github.com/ffenix113/go-socketio"
    "github.com/ffenix113/go-socketio/engineio"
    "github.com/gobwas/ws"
    "github.com/gobwas/ws/wsutil"
)

var (
    pingInterval = 60 * time.Second
    pingTimeout = 5 * time.Second
    // codec is specifically extracted to show that nil is valid.
    // In this case JSON codec will be used.
    codec socketio.DataCodec = nil
)

func main() {
    sIO := socketio.NewEngine(pingInterval, pingTimeout, wsutil.ReadClientText, wsutil.WriteServerText, codec)

    // This method will be executed when new client connects.
    // The `data` argument is raw `auth` option value as specified [here](https://socket.io/docs/v4/client-options/#auth)
    sIO.OnConnect = func(s *socketio.Socket, _ string, data []byte) (any, error) {
        // Do some validations / JWT parsing for example
        // Then if you want to attach some userID to this socket do
        s.UserID = ""

        return nil, nil
    }

    // Clean-up can be done in `OnDisconnect` callback.
    // At this point socket is already closed, so don't send anything to it.
    sIO.OnDisconnect = func(s *socketio.Socket, event string, data []byte) (any, error) {
		return nil, nil
	}

    // Events can be attached with `sIO.On(..., ...)`.
    // `data` argument contains raw JSON message as sent by the client.
    //
    // Returned value will be passed on to client only if client does [emitWithAck](https://socket.io/docs/v4/client-api/#socketemitwithackeventname-args).
    // If client just does `emit` - response is omitted.
    sIO.On("hello", func(s *socketio.Socket, _ string, data []byte) (any, error) {
        var req struct {
            Name string `json:"name"`
        }

        if err := json.Unmarshal(data, &req); err != nil {
            return nil, err
        }

        // Anything that can be marshaled to JSON can be returned.
        return map[string]string{
            "msg": fmt.Sprintf("hello user %q with id %q", req.Name, s.UserID)
        }, nil
    })

    // Attach socketio handler to any http server
    socketHandler := WebsocketHandler(sIO)
    http.DefaultServeMux.Handle("/socket.io/", socketHandler)
    http.ListenAndServe("0.0.0.0:3000", http.DefaultServeMux)
}

func WebsocketHandler(e *socketio.Engine) http.HandlerFunc {
	return func(rw http.ResponseWriter, req *http.Request) {
		query := req.URL.Query()
		if query.Get("EIO") != engineio.Version {
			http.Error(rw, "unsupported version", http.StatusBadRequest)
			return
		}
        // Polling is indeed not supported currently.
		if query.Get("transport") == "polling" {
			http.Error(rw, "polling is not supported", http.StatusBadRequest)
			return
		}
        // This is optional, but if client provides `sid` it might expect
        // continuation of the session, which we will not do.
		if sid := query.Get("sid"); sid != "" {
			http.Error(rw, "sid should be empty", http.StatusBadRequest)
			return
		}

		conn, _, _, err := ws.UpgradeHTTP(req, rw)
		if err != nil {
			panic(err)
		}

		e.AddClient(conn)
	}
}

Documentation

Index

• Constants
• func Marshal(val any) []byte
• func NewJSONCodec() jsonCodec
• type AdapterReceiver
• type AdapterSender
  • func NewRedisAdapter(r redis.UniversalClient, recvr AdapterReceiver, codec DataCodec, ...) AdapterSender
• type DataCodec
• type Engine
  • func NewEngine(pingInterval, pingTimeout time.Duration, read engineio.ReadBytes, ...) *Engine
  • func (e *Engine) AddClient(conn net.Conn) *Socket
  • func (e *Engine) Broadcast(_ context.Context, event string, data any) error
  • func (e *Engine) EmitForUser(_ context.Context, userID, event string, data any) error
  • func (e *Engine) GetCodec() DataCodec
  • func (e *Engine) NewSocket(cl *engineio.Socket) *Socket
  • func (e *Engine) On(event string, handler SocketEventHandler)
  • func (e *Engine) ReceivedNew(ctx context.Context, userID, event string, data json.RawMessage)
  • func (e *Engine) RemoveSocket(socket *Socket) *Socket
  • func (e *Engine) RemoveSocketUnsafe(socket *Socket) *Socket
• type ErrorData
• type Packet
  • func UnmarshalPacket(data []byte) (p Packet, err error)
  • func (p Packet) MarshalBinary() (data []byte, err error)
  • func (p *Packet) UnmarshalBinary(data []byte) error
• type PacketType
• type PushData
• type RedisAdapter
  • func (a RedisAdapter) Broadcast(ctx context.Context, event string, data any) error
  • func (a RedisAdapter) EmitForUser(ctx context.Context, userID string, event string, data any) error
  • func (a RedisAdapter) Listen(ctx context.Context) error
• type Socket
  • func (s *Socket) Close()
  • func (s *Socket) Emit(event string, data any)
  • func (s *Socket) Server() *Engine
• type SocketEventHandler

Constants

const (
	// CatchAllEvent is special event, that will be selected
	// if no handlers are registered for received event type.
	CatchAllEvent    = "*"
	DefaultNamespace = "/"
)

const Version = 5

Functions

func Marshal

func Marshal(val any) []byte

func NewJSONCodec

func NewJSONCodec() jsonCodec

Types

type AdapterReceiver

type AdapterReceiver interface {
	// ReceivedNew will be called when new event is received.
	//
	// userID may be empty if this is a broadcast event.
	//
	// Currently all notifications will be processed here,
	// without difference between user or broadcast notifications or
	// if user in received event is present on this server.
	ReceivedNew(ctx context.Context, userID, event string, data json.RawMessage)
}

type AdapterSender

type AdapterSender interface {
	Broadcast(ctx context.Context, event string, data any) error
	EmitForUser(ctx context.Context, userID, event string, data any) error
}

func NewRedisAdapter

func NewRedisAdapter(r redis.UniversalClient, recvr AdapterReceiver, codec DataCodec, eventsChannel string) AdapterSender

type DataCodec

type DataCodec interface {
	Marashal(data any) ([]byte, error)
	UnmarshalTo(data []byte, to any) error
}

type Engine

type Engine struct {
	OnConnect    SocketEventHandler
	OnDisconnect SocketEventHandler
	// contains filtered or unexported fields
}

func NewEngine

func NewEngine(pingInterval, pingTimeout time.Duration, read engineio.ReadBytes, write engineio.WriteBytes, codec DataCodec) *Engine

func (*Engine) AddClient

func (e *Engine) AddClient(conn net.Conn) *Socket

AddClient creates and stores Socket.

TODO: Better init handshake. Wait for client to send connect & auth before moving forward.

func (*Engine) Broadcast

func (e *Engine) Broadcast(_ context.Context, event string, data any) error

func (*Engine) EmitForUser

func (e *Engine) EmitForUser(_ context.Context, userID, event string, data any) error

EmitForUser emits provided event to all sockets for this user connected to this engine.

It is only useful if `UserID` field is set on socket.

func (*Engine) GetCodec

func (e *Engine) GetCodec() DataCodec

func (*Engine) NewSocket

func (e *Engine) NewSocket(cl *engineio.Socket) *Socket

func (*Engine) On

func (e *Engine) On(event string, handler SocketEventHandler)

On adds event listener to specified event.

To add catch-all listener use `On(socketio.CatchAllEvent, ...)`.

func (*Engine) ReceivedNew

func (e *Engine) ReceivedNew(ctx context.Context, userID, event string, data json.RawMessage)

ReceivedNew is used for adapter only.

This method should not be used anywhere in the code apart from adapter implementation.

func (*Engine) RemoveSocket

func (e *Engine) RemoveSocket(socket *Socket) *Socket

func (*Engine) RemoveSocketUnsafe

func (e *Engine) RemoveSocketUnsafe(socket *Socket) *Socket

type ErrorData

type ErrorData struct {
	Error string `json:"error"`
}

type Packet

type Packet struct {
	Type      PacketType
	Namespace string
	AckID     *int
	Data      json.RawMessage
}

func UnmarshalPacket

func UnmarshalPacket(data []byte) (p Packet, err error)

func (Packet) MarshalBinary

func (p Packet) MarshalBinary() (data []byte, err error)

func (*Packet) UnmarshalBinary

func (p *Packet) UnmarshalBinary(data []byte) error

type PacketType

type PacketType string

const (
	PacketTypeConnect      PacketType = "0"
	PacketTypeDisconnect   PacketType = "1"
	PacketTypeEvent        PacketType = "2"
	PacketTypeAck          PacketType = "3"
	PacketTypeConnectError PacketType = "4"
	PacketTypeBinaryEvent  PacketType = "5"
	PacketTypeBinaryAck    PacketType = "6"
)

type PushData

type PushData struct {
	UserID string
	Event  string
	Data   json.RawMessage
}

type RedisAdapter

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

func (RedisAdapter) Broadcast

func (a RedisAdapter) Broadcast(ctx context.Context, event string, data any) error

func (RedisAdapter) EmitForUser

func (a RedisAdapter) EmitForUser(ctx context.Context, userID string, event string, data any) error

func (RedisAdapter) Listen

func (a RedisAdapter) Listen(ctx context.Context) error

type Socket

type Socket struct {
	// UserID is used only for `engine.EmitForUser` method.
	// If that method is not used - this field can be empty.
	UserID string
	// contains filtered or unexported fields
}

func (*Socket) Close

func (s *Socket) Close()

func (*Socket) Emit

func (s *Socket) Emit(event string, data any)

func (*Socket) Server

func (s *Socket) Server() *Engine

type SocketEventHandler

type SocketEventHandler func(s *Socket, event string, data []byte) (any, error)