gws

README

gws

event-driven websocket server

Highlight

• zero dependency, no channel but event driven
• zero extra goroutine to manage connection
• zero error to read/write operation, errors have been handled appropriately
• built-in concurrent_map implementation
• fully passes the WebSocket autobahn-testsuite

Attention

• It's designed for api server, do not write big message
• It's recommended not to enable data compression in the intranet
• WebSocket Events are emitted synchronously, manage goroutines yourself

Core Interface

type Event interface {
OnOpen(socket *Conn)
OnError(socket *Conn, err error)
OnClose(socket *Conn, code uint16, reason []byte)
OnPing(socket *Conn, payload []byte)
OnPong(socket *Conn, payload []byte)
OnMessage(socket *Conn, message *Message)
}

Install

go get -v github.com/lxzan/gws@latest

Quick Start

package main

import (
	"fmt"
	"github.com/lxzan/gws"
	"net/http"
)

func main() {
	var config = &gws.Config{
		CompressEnabled:   true,
		CheckTextEncoding: true,
		MaxContentLength:  32 * 1024 * 1024,
	}
	var handler = new(WebSocket)
	http.HandleFunc("/connect", func(writer http.ResponseWriter, request *http.Request) {
		socket, err := gws.Accept(writer, request, handler, config)
		if err != nil {
			return
		}
		socket.Listen()
	})

	_ = http.ListenAndServe(":3000", nil)
}

type WebSocket struct{}

func (c *WebSocket) OnClose(socket *gws.Conn, code uint16, reason []byte) {
	fmt.Printf("onclose: code=%d, payload=%s\n", code, string(reason))
}

func (c *WebSocket) OnError(socket *gws.Conn, err error) {
	fmt.Printf("onerror: err=%s\n", err.Error())
}

func (c *WebSocket) OnOpen(socket *gws.Conn) {
	println("connected")
}

func (c *WebSocket) OnPing(socket *gws.Conn, payload []byte) {
	fmt.Printf("onping: payload=%s\n", string(payload))
	socket.WritePong(payload)
}

func (c *WebSocket) OnPong(socket *gws.Conn, payload []byte) {}

func (c *WebSocket) OnMessage(socket *gws.Conn, message *gws.Message) {
	socket.WriteMessage(message.Typ(), message.Bytes())
	message.Close()
}

TLS

package main

import (
	"github.com/gin-gonic/gin"
	"github.com/lxzan/gws"
)

func main() {
	app := gin.New()
	handler := new(WebSocket)
	app.GET("/connect", func(ctx *gin.Context) {
		socket, err := gws.Accept(ctx.Writer, ctx.Request, handler, nil)
		if err != nil {
			return
		}
		socket.Listen()
	})
	cert := "server.crt"
	key := "server.key"
	if err := app.RunTLS(":8443", cert, key); err != nil {
		panic(err)
	}
}

Test

cd examples/testsuite
mkdir reports
docker run -it --rm \
    -v ${PWD}/config:/config \
    -v ${PWD}/reports:/reports \
    crossbario/autobahn-testsuite \
    wstest -m fuzzingclient -s /config/fuzzingclient.json

Benchmark

• machine: MacBook Pro M1
• body.json size: 2.34KiB
• max cost: cpu=440% memory=90MiB
• command: tcpkali -c 200 -r 20000 -T 20s -f body.json --ws 127.0.0.1:3000/connect

lxzan/gws

Destination: [127.0.0.1]:3000
Interface lo0 address [127.0.0.1]:0
Using interface lo0 to connect to [127.0.0.1]:3000
Ramped up to 200 connections.
Total data sent:     28864.1 MiB (30266158036 bytes)
Total data received: 28812.3 MiB (30211911226 bytes)
Bandwidth per channel: 120.909⇅ Mbps (15113.7 kBps)
Aggregate bandwidth: 12080.082↓, 12101.773↑ Mbps
Packet rate estimate: 1098496.2↓, 1105141.9↑ (8↓, 14↑ TCP MSS/op)
Test duration: 20.0078 s.

gorilla/websocket

Destination: [127.0.0.1]:3000
Interface lo0 address [127.0.0.1]:0
Using interface lo0 to connect to [127.0.0.1]:3000
Ramped up to 200 connections.
Total data sent:     15043.6 MiB (15774402996 bytes)
Total data received: 14990.0 MiB (15718181647 bytes)
Bandwidth per channel: 62.972⇅ Mbps (7871.5 kBps)
Aggregate bandwidth: 6285.944↓, 6308.428↑ Mbps
Packet rate estimate: 589201.4↓, 579598.1↑ (5↓, 12↑ TCP MSS/op)
Test duration: 20.0042 s.

Documentation

Index

• type ConcurrentMap
  • func NewConcurrentMap(segments uint64) *ConcurrentMap
  • func (c *ConcurrentMap) Delete(key interface{})
  • func (c *ConcurrentMap) Len() int
  • func (c *ConcurrentMap) Load(key interface{}) (value interface{}, exist bool)
  • func (c *ConcurrentMap) Range(f func(key interface{}, value interface{}) bool)
  • func (c *ConcurrentMap) Store(key interface{}, value interface{})
• type Config
• type Conn
  • func Accept(w http.ResponseWriter, r *http.Request, eventHandler Event, config *Config) (*Conn, error)
  • func (c *Conn) Close(code uint16, reason []byte)
  • func (c *Conn) Listen()
  • func (c *Conn) LocalAddr() net.Addr
  • func (c *Conn) RemoteAddr() net.Addr
  • func (c *Conn) SetDeadline(t time.Time)
  • func (c *Conn) SetReadDeadline(t time.Time)
  • func (c *Conn) SetWriteDeadline(t time.Time)
  • func (c *Conn) WriteMessage(opcode Opcode, payload []byte)
  • func (c *Conn) WritePing(payload []byte)
  • func (c *Conn) WritePong(payload []byte)
• type Event
• type Map
  • func NewMap() *Map
  • func (c *Map) Delete(key interface{})
  • func (c *Map) Len() int
  • func (c *Map) Load(key interface{}) (value interface{}, exist bool)
  • func (c *Map) Range(f func(key, value interface{}) bool)
  • func (c *Map) Store(key interface{}, value interface{})
• type Message
  • func (c *Message) Bytes() []byte
  • func (c *Message) Close()
  • func (c *Message) Read(p []byte) (n int, err error)
  • func (c *Message) Typ() Opcode
• type Opcode
  • func (c Opcode) IsDataFrame() bool
• type Request
• type SessionStorage

Types

type ConcurrentMap

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

ConcurrentMap used to store websocket connections in the IM server 用来存储IM等服务的连接

func NewConcurrentMap

func NewConcurrentMap(segments uint64) *ConcurrentMap

func (*ConcurrentMap) Delete

func (c *ConcurrentMap) Delete(key interface{})

func (*ConcurrentMap) Len

func (c *ConcurrentMap) Len() int

func (*ConcurrentMap) Load

func (c *ConcurrentMap) Load(key interface{}) (value interface{}, exist bool)

func (*ConcurrentMap) Range

func (c *ConcurrentMap) Range(f func(key interface{}, value interface{}) bool)

Range calls f sequentially for each key and value present in the map. If f returns false, range stops the iteration.

func (*ConcurrentMap) Store

func (c *ConcurrentMap) Store(key interface{}, value interface{})

type Config

type Config struct {
	// whether to compress data
	CompressEnabled bool

	// compress level eg: flate.BestSpeed
	CompressLevel int

	// max message size
	MaxContentLength int

	// whether to check utf8 encoding, disabled for better performance
	CheckTextEncoding bool

	// https://www.rfc-editor.org/rfc/rfc6455.html#section-1.3
	// attention: client may not support custom response header, use nil instead
	ResponseHeader http.Header

	// client authentication
	CheckOrigin func(r *Request) bool
}

type Conn

type Conn struct {

	// Concurrent Variable
	// store session information
	SessionStorage SessionStorage
	// contains filtered or unexported fields
}

func Accept

func Accept(w http.ResponseWriter, r *http.Request, eventHandler Event, config *Config) (*Conn, error)

Accept http protocol upgrade to websocket ctx done means server stopping

func (*Conn) Close

func (c *Conn) Close(code uint16, reason []byte)

Close proactively close the connection code: https://developer.mozilla.org/zh-CN/docs/Web/API/CloseEvent#status_codes 主动关闭连接, 发送关闭帧, 并将连接状态置为关闭

func (*Conn) Listen

func (c *Conn) Listen()

Listen listening to websocket messages through a dead loop 通过死循环监听websocket消息

func (*Conn) LocalAddr

func (c *Conn) LocalAddr() net.Addr

func (*Conn) RemoteAddr

func (c *Conn) RemoteAddr() net.Addr

func (*Conn) SetDeadline

func (c *Conn) SetDeadline(t time.Time)

SetDeadline sets deadline

func (*Conn) SetReadDeadline

func (c *Conn) SetReadDeadline(t time.Time)

SetReadDeadline sets read deadline

func (*Conn) SetWriteDeadline

func (c *Conn) SetWriteDeadline(t time.Time)

SetWriteDeadline sets write deadline

func (*Conn) WriteMessage

func (c *Conn) WriteMessage(opcode Opcode, payload []byte)

WriteMessage write text/binary message text message must be utf8 encoding 发送文本/二进制消息, 文本消息必须是utf8编码

func (*Conn) WritePing

func (c *Conn) WritePing(payload []byte)

WritePing write ping frame

func (*Conn) WritePong

func (c *Conn) WritePong(payload []byte)

WritePong write pong frame

type Event

type Event interface {
	OnOpen(socket *Conn)
	OnError(socket *Conn, err error)
	OnClose(socket *Conn, code uint16, reason []byte)
	OnPing(socket *Conn, payload []byte)
	OnPong(socket *Conn, payload []byte)
	OnMessage(socket *Conn, message *Message)
}

WebSocket Event one of onclose and onerror will be called once during the connection's lifetime. 在连接的生命周期中，onclose和onerror中的一个有且只有一次被调用

type Map

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

func NewMap

func NewMap() *Map

func (*Map) Delete

func (c *Map) Delete(key interface{})

Delete deletes the value for a key.

func (*Map) Len

func (c *Map) Len() int

func (*Map) Load

func (c *Map) Load(key interface{}) (value interface{}, exist bool)

func (*Map) Range

func (c *Map) Range(f func(key, value interface{}) bool)

Range calls f sequentially for each key and value present in the map. If f returns false, range stops the iteration.

func (*Map) Store

func (c *Map) Store(key interface{}, value interface{})

Store sets the value for a key.

type Message

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

func (*Message) Bytes

func (c *Message) Bytes() []byte

Bytes get message content

func (*Message) Close

func (c *Message) Close()

Close recycle buffer

func (*Message) Read

func (c *Message) Read(p []byte) (n int, err error)

func (*Message) Typ

func (c *Message) Typ() Opcode

Typ get message type

type Opcode

type Opcode uint8

const (
	OpcodeContinuation    Opcode = 0x0
	OpcodeText            Opcode = 0x1
	OpcodeBinary          Opcode = 0x2
	OpcodeCloseConnection Opcode = 0x8
	OpcodePing            Opcode = 0x9
	OpcodePong            Opcode = 0xA
)

func (Opcode) IsDataFrame

func (c Opcode) IsDataFrame() bool

type Request

type Request struct {
	*http.Request                 // http request
	SessionStorage SessionStorage // store user session
}

type SessionStorage

type SessionStorage interface {
	Load(key interface{}) (value interface{}, exist bool)
	Delete(key interface{})
	Store(key interface{}, value interface{})
	Range(f func(key, value interface{}) bool)
}

SessionStorage because sync.Map is not easy to debug, so I implemented my own map. if you don't like it, you can also use sync.Map instead.