gws

README

gws

event-driven go websocket server

Highlight

• zero dependency, not use 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 environment
• WebSocket events are emitted synchronously, manage goroutines yourself

Benchmark

• machine: Ubuntu 20.04LTS VM (4C8T)
• client: tcpkali
• payload: 2.34KiB

Server	Connection	Send Speed (msg/s)	T (s)	Download / Upload Bandwidth (Mbps)
gws	200	4000	30	10916.094↓ 10951.587↑
gorilla	200	4000	30	4344.222↓ 4380.711↑
gws	2000	300	30	7941.090↓ 7951.117↑
gorilla	2000	300	30	4706.938↓ 4715.744↑
gws	5000	60	30	5891.151↓ 5908.599↑
gorilla	5000	60	30	-
gws	10000	10	60	1980.124↓ 1977.561↑
gorilla	10000	10	60	1972.556↓ 1979.981↑
gws	10000	20	60	3952.788↓ 3959.341↑
gorilla	10000	20	60	-

- means exception

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 (Autobahn Server)

package main

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

func main() {
	var upgrader = gws.NewUpgrader(func(c *gws.Upgrader) {
		c.CompressEnabled = true
		c.CheckTextEncoding = true
		c.MaxContentLength = 32 * 1024 * 1024
		c.EventHandler = new(WebSocket)
	})

	http.HandleFunc("/connect", func(writer http.ResponseWriter, request *http.Request) {
		socket, err := upgrader.Accept(writer, request)
		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)
	upgrader := gws.NewUpgrader(gws.WithEventHandler(handler))
	app.GET("/connect", func(ctx *gin.Context) {
		socket, err := upgrader.Accept(ctx.Writer, ctx.Request)
		if err != nil {
			return
		}
		upgrader.Listen(socket)
	})
	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

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 Conn
  • func (c *Conn) Close(code uint16, reason []byte)
  • func (c *Conn) Listen()
  • func (c *Conn) LocalAddr() net.Addr
  • func (c *Conn) NetConn() net.Conn
  • 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)
  • func (c *Conn) WriteString(s string)
• 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 Option
  • func WithCheckOrigin(f func(r *Request) bool) Option
  • func WithCheckTextEncoding(check bool) Option
  • func WithCompress(enabled bool, level int) Option
  • func WithEventHandler(eventHandler Event) Option
  • func WithInitialize() Option
  • func WithMaxContentLength(n int) Option
  • func WithResponseHeader(h http.Header) Option
• type Request
• type SessionStorage
• type Upgrader
  • func NewUpgrader(options ...Option) *Upgrader
  • func (c *Upgrader) Accept(w http.ResponseWriter, r *http.Request) (*Conn, error)

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 Conn

type Conn struct {

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

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) NetConn

func (c *Conn) NetConn() net.Conn

NetConn get tcp/tls/... conn

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

func (*Conn) WriteString

func (c *Conn) WriteString(s string)

WriteString write text 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 Option

type Option func(c *Upgrader)

func WithCheckOrigin

func WithCheckOrigin(f func(r *Request) bool) Option

WithCheckOrigin check request origin

func WithCheckTextEncoding

func WithCheckTextEncoding(check bool) Option

WithCheckTextEncoding set text encoding checking

func WithCompress

func WithCompress(enabled bool, level int) Option

WithCompress set deflate compress

func WithEventHandler

func WithEventHandler(eventHandler Event) Option

WithEventHandler set event handler

func WithInitialize

func WithInitialize() Option

WithInitialize initialize the upgrader configure

func WithMaxContentLength

func WithMaxContentLength(n int) Option

WithMaxContentLength set max content length

func WithResponseHeader

func WithResponseHeader(h http.Header) Option

WithResponseHeader set response header client may not support, use nil instead

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.

type Upgrader

type Upgrader struct {
	// websocket event handler
	EventHandler Event

	// 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
}

Upgrader websocket upgrader do not use &Upgrader unless, some options may not be initialized NewUpgrader is recommended

func NewUpgrader

func NewUpgrader(options ...Option) *Upgrader

func (*Upgrader) Accept

func (c *Upgrader) Accept(w http.ResponseWriter, r *http.Request) (*Conn, error)

Accept http protocol upgrade to websocket ctx done means server stopping