WebsocketHub

About

WebsocketHub is a standalone server / Go library that supports registering channels with specific message type for specific roles

Usage as standalone server

install the binary:

go install github.com/SallimanR/websockethub@latest

run the server:

websockethub -config websockethub.json

Example config:

{
  "port": 8080,
  "roles": ["tow_driver", "tow_subscriber"],
  "channels": [
    { "name": "GPS_REALTIME", "roles": ["tow_driver", "tow_subscriber"] }
  ]
}

Usage as library

install the library:

go get github.com/SallimanR/websockethub

example with gin router, allowed origins for CORS protection and auth middleware

import (
    "github.com/SallimanR/websockethub/websockethub"
    "github.com/gin-gonic/gin"
)

func registerWSRoutes(router *gin.Engine, origins []string, authMiddleware gin.HandlerFunc) {
    wsOptions := websockethub.WebsocketServerOptions{
        Roles:          []string{"tow_driver", "tow_subscriber"},
        AllowedOrigins: origins,
    }
    wsServer := websockethub.NewWebsocketServer(wsOptions)
	wsGroup := router.Group("/ws")
	wsGroup.Use(authMiddleware)
	wsGroup.GET("/:role", wsServer.WebsocketUpgradeHandler)
}

register channel with message type

example with gps realtime channel

import (
    "time"

    "github.com/SallimanR/websockethub/websockethub"
    wsPB "github.com/SallimanR/websockethub/websockethub/proto"
    "google.golang.org/protobuf/proto"
)

type MovingDriver struct {
    DriverID   int64
    Latitude   float32
    Longitude  float32
    TravelTime time.Time
    PathMeters int32
}

type MovingDriverWithPoints struct {
    MovingDriver
    Points [][2]float32
}

type GPSRealtimeChannel struct {
    *websockethub.PubSubChannel[MovingDriverWithPoints]
}

func NewGPSRealtimeChannel(wsServer *websockethub.WebsocketServer, roles []string) (*GPSRealtimeChannel, error) {
    ch := &GPSRealtimeChannel{
        PubSubChannel: websockethub.NewPubSubChannel[MovingDriverWithPoints](),
    }
    err := wsServer.RegisterChannel(roles, wsPB.Channel_GPS_REALTIME, ch)
    if err != nil {
        return nil, err
    }
    return ch, nil
}

func (c *GPSRealtimeChannel) Publish(publisherID int64, msg []byte) error {
    var data wsPB.GPSUpdate
    if err := proto.Unmarshal(msg, &data); err != nil {
        return err
    }
    gpsData := MovingDriverWithPoints{
        MovingDriver: MovingDriver{
            DriverID:   publisherID,
            Latitude:   data.Coordinates[0].Lat,
            Longitude:  data.Coordinates[0].Lng,
            TravelTime: time.Now(),
            PathMeters: int32(len(data.Coordinates)),
        },
    }
    c.Messages.Store(publisherID, gpsData)
    return nil
}

func (c *GPSRealtimeChannel) GetMessages(publisherIDs []int64) ([]byte, error) {
    var batch wsPB.MessageBatch
    for _, id := range publisherIDs {
        item, ok := c.Messages.Load(id)
        if !ok {
            continue
        }
        data, err := proto.Marshal(&item)
        if err != nil {
            return nil, err
        }
        batch.Data = append(batch.Data, data)
    }
    return proto.Marshal(&batch)
}

Developing

Development mode:

run server with hot code reload:

air

Production:

run in the project root:

go run .

Or build the binary:

go build -o ./bin/websockethub .

Running tests:

go test ./...

generate protobuf

generate_protobuf.sh

Architecture

Non-blocking I/O

• Message reads and writes to websocket connection are fully non-blocking
• No goroutine per connection: NBIO uses epoll(Linux) / kqueue(BSD based OSs, MacOS) which means that memory usage is way lower per connection and Go's scheduler is not loaded so in theory it scales up to 1 Million+ connections

Roles

• WebsocketHub have a set of roles under which channels can be registered

Channels

WebsocketHub is based on channels, roles and RBAC:

• each channels have its own message type
• each channel is interfaced by ChannelActions interface
• each channel is registered to set of roles

Messages:

• WebsocketHub uses protobuf for messages
• request/response messages
• each message have a type of Publish, Subscribe, Unsubscribe

Message scheduler

• WebsocketHub have a message scheduler that schedules each message's broadcast for each user for a constant period of ticks
• it uses a version of time wheel data structure

Why Websocket?

Websocket is supported in all major browsers Websocket vs REST:

1. no need for http server and load balancer
2. bidirectional
3. supports binary format for messages
4. stateful => ability to cache subscriptions on server side