websockets

README

ao-concepts websockets module

This module provides a websocket handler for usage with fiber. The handler is a abstraction layer that makes the handling of websocket messages easier.

Contributing

If you are interested in contributing to this project, feel free to open an issue to discus a new feature, enhancement or improvement. If you found a bug or security vulnerability in this package, please start a issue, or open a PR against master.

Installation

go get -u github.com/ao-concepts/websockets

Usage

cnt := NewServiceContainer() // the service container has to implement the `websockets.ServiceContainer` interface.
server, err := websockets.New(cnt)
if err != nil {
    log.ErrFatal(err)
}

// batch sent messages. This can be used to reduce load on clients.
// Batched message will be prefixed by `batch_`.
// The data will be stored as array of batched payloads below the `d` property of the actual sent message.
if err := server.UseBatch("event:name", time.Second); err != nil {
	log.ErrError(err)
}

server.Subscribe("event:name", func(msg *websockets.Message) {
    // here you can handle the message
    c := msg.Connection

    // access the payload
    fmt.Println(msg.Payload["value"])

    // you can set data to the current session
    c.Set("key", "value")
    c.Get("key")

    // you can respond directly on the connection that received the message.
    c.SendMessage(&websockets.Message{
        Event:   "event:name",
        Payload: websockets.Payload{
            "value": "data",
        },
    })

    // and you can respond to all connections that match a filter.
    c.SendMessage(&websockets.Message{
        Event:   "event:name",
        Payload: websockets.Payload{
            "value": "data",
        },
    }, func (c *websockets.Connection) bool {
        return c.Get("key") == true
    })
})

app := fiber.New()
app.Get("/ws", s.Handler)
app.Listen(":3000")

Used packages

This project uses some really great packages. Please make sure to check them out!

Package	Usage
github.com/ao-concepts/eventbus	Persistence helper
github.com/go-co-op/gocron	Batch messages
github.com/gofiber/fiber	HTTP router
github.com/stretchr/testify	Testing

Documentation

Index

• Constants
• type Batch
  • func NewBatch() *Batch
  • func (b *Batch) AddPayload(payload Payload)
  • func (b *Batch) GetDataAndRemoveAll() []Payload
• type Connection
  • func NewConnection(s *Server, wc WebsocketConn, ctx context.Context, cancel func()) *Connection
  • func (c *Connection) Get(key string) interface{}
  • func (c *Connection) Locals(key string) interface{}
  • func (c *Connection) Publish(msg *Message, filter Filter)
  • func (c *Connection) SendMessage(msg *Message)
  • func (c *Connection) SendMessageUnBatched(msg *Message)
  • func (c *Connection) Set(key string, value interface{})
• type Filter
• type Logger
• type Message
• type MessageWithConnection
• type OnConnectionClose
• type OnEndFunc
• type Payload
• type Server
  • func New(cnt ServiceContainer) *Server
  • func (s *Server) Connect(conn *Connection)
  • func (s *Server) CountConnections(filter Filter) int
  • func (s *Server) GetBatches() []string
  • func (s *Server) Handler(c *fiber.Ctx) error
  • func (s *Server) Publish(msg *Message, filter Filter)
  • func (s *Server) SetOnConnectionClose(fn OnConnectionClose)
  • func (s *Server) Shutdown()
  • func (s *Server) Subscribe(eventName string, handler func(msg *MessageWithConnection)) error
  • func (s *Server) UseBatch(event string, interval time.Duration) error
• type ServerConfig
• type ServiceContainer
• type WebsocketConn
• type WebsocketConnMock
  • func NewWebsocketConnMock() *WebsocketConnMock
  • func (c *WebsocketConnMock) Close() error
  • func (c *WebsocketConnMock) Locals(key string) interface{}
  • func (c *WebsocketConnMock) ReadJSON(v interface{}) error
  • func (c *WebsocketConnMock) WriteJSON(v interface{}) error
  • func (c *WebsocketConnMock) WriteMessage(messageType int, data []byte) error

Constants

const PingInterval = (PongTimeout * 9) / 10

const PongTimeout = 60 * time.Second

Types

type Batch

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

func NewBatch

func NewBatch() *Batch

NewBatch constructor

func (*Batch) AddPayload

func (b *Batch) AddPayload(payload Payload)

AddPayload add payload to the batch

func (*Batch) GetDataAndRemoveAll

func (b *Batch) GetDataAndRemoveAll() []Payload

GetDataAndRemoveAll get the current payload batch and remove all data from the batch

type Connection

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

Connection on a websocket

func NewConnection

func NewConnection(s *Server, wc WebsocketConn, ctx context.Context, cancel func()) *Connection

NewConnection constructor

func (*Connection) Get

func (c *Connection) Get(key string) interface{}

Get the value of a key on the connection. Returns nil if the key does not exist

func (*Connection) Locals

func (c *Connection) Locals(key string) interface{}

Locals gets a local by key from the underlying connection.

func (*Connection) Publish

func (c *Connection) Publish(msg *Message, filter Filter)

Publish a message to all matching connections

func (*Connection) SendMessage

func (c *Connection) SendMessage(msg *Message)

SendMessage via this connection. Message can be sent batched using `UseBatch`.

func (*Connection) SendMessageUnBatched

func (c *Connection) SendMessageUnBatched(msg *Message)

SendMessageUnBatched send a message via this connection. Massage will not be batched.

func (*Connection) Set

func (c *Connection) Set(key string, value interface{})

Set a key on a connection to a specific value. Overwrites old value.

type Filter

type Filter func(c *Connection) bool

Filter function that checks if a connection matches some criteria

type Logger

type Logger interface {
	Fatal(s string, args ...interface{})
	ErrError(err error)
	Error(s string, args ...interface{})
	ErrInfo(err error)
	Info(s string, args ...interface{})
	Debug(s string, args ...interface{})
}

Logger interface

type Message

type Message struct {
	Event   string  `json:"e"`
	Payload Payload `json:"p"`
}

Message that is received or sent via a websocket.

type MessageWithConnection

type MessageWithConnection struct {
	Message
	Connection *Connection `json:"-"`
}

type OnConnectionClose

type OnConnectionClose func(c *Connection)

OnConnectionClose is executed when a connection is closed

type OnEndFunc

type OnEndFunc = func(c *Connection)

OnEndFunc function that is executed on connection ends

type Payload

type Payload map[string]interface{}

Payload send by a websocket connection

type Server

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

Server for websockets

func New

func New(cnt ServiceContainer) *Server

New server constructor

func (*Server) Connect

func (s *Server) Connect(conn *Connection)

Connect a websocket to the server

func (*Server) CountConnections

func (s *Server) CountConnections(filter Filter) int

CountConnections returns the number of currently active connections (a filter can be used to restrict the count)

func (*Server) GetBatches

func (s *Server) GetBatches() []string

GetBatches return the current list of batches registered for the server.

func (*Server) Handler

func (s *Server) Handler(c *fiber.Ctx) error

Handler that reads from and writes to a websocket connection

func (*Server) Publish

func (s *Server) Publish(msg *Message, filter Filter)

Publish data to all matching connections

func (*Server) SetOnConnectionClose

func (s *Server) SetOnConnectionClose(fn OnConnectionClose)

SetOnConnectionClose sets the function that is executed when a connection is closed

func (*Server) Shutdown

func (s *Server) Shutdown()

Shutdown gracefully stops the server

func (*Server) Subscribe

func (s *Server) Subscribe(eventName string, handler func(msg *MessageWithConnection)) error

Subscribe to a websocket event

func (*Server) UseBatch

func (s *Server) UseBatch(event string, interval time.Duration) error

UseBatch registers a batch interval sender. Sends events in a batched manner when they are sent by `Connection.SendMessage`. Has to be called before the first connection is established. Interval in seconds.

type ServerConfig

type ServerConfig struct {
	ReadBufferSize  int
	WriteBufferSize int
}

ServerConfig configuration of the server

type ServiceContainer

type ServiceContainer interface {
	GetLogger() Logger
	GetWebsocketsConfig() *ServerConfig
}

type WebsocketConn

type WebsocketConn interface {
	ReadJSON(v interface{}) error
	WriteJSON(v interface{}) error
	WriteMessage(messageType int, data []byte) error
	Close() error
	Locals(key string) interface{}
}

WebsocketConn websocket connection interface

type WebsocketConnMock

type WebsocketConnMock struct {
	Read      chan interface{}
	Write     chan interface{}
	LocalData map[string]interface{}
	DoError   bool
}

WebsocketConnMock mock struct for websocket connections You can set `DoError` to let any method that can return an error return errors

func NewWebsocketConnMock

func NewWebsocketConnMock() *WebsocketConnMock

NewWebsocketConnMock constructor

func (*WebsocketConnMock) Close

func (c *WebsocketConnMock) Close() error

Close the connection

func (*WebsocketConnMock) Locals

func (c *WebsocketConnMock) Locals(key string) interface{}

Locals access local data

func (*WebsocketConnMock) ReadJSON

func (c *WebsocketConnMock) ReadJSON(v interface{}) error

ReadJSON receives data from the Write chan

func (*WebsocketConnMock) WriteJSON

func (c *WebsocketConnMock) WriteJSON(v interface{}) error

WriteJSON writes data to the Read chan

func (*WebsocketConnMock) WriteMessage

func (c *WebsocketConnMock) WriteMessage(messageType int, data []byte) error

WriteMessage to the connection