gws

README

gws

event-driven go websocket server

• gws
  • Highlight
  • Core Interface
  • Install
  • Examples
  • Quick Start (Autobahn Server)
  • TLS
  • Autobahn Test
  • Benchmark

Highlight

• No dependency
• No additional resident concurrent goroutine
• Asynchronous non-blocking read and write support
• High IOPS and low latency
• Fully passes the WebSocket autobahn-testsuite

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

Examples

• chat room
• echo

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
		}
		go 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)
	}
}

Autobahn 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: Ubuntu 20.04LTS VM (4C8T)

• High IOPS

tcpkali -c 1000 --connect-rate 500 -r 1000 -T 300s -f assets/1K.txt --ws 127.0.0.1:${port}/connect

• Low Latency

tcpkali -c 1000 --connect-rate 500 -r 100 -T 300s -f assets/1K.txt --ws 127.0.0.1:${port}/connect

• Low CPU Usage

PID  USER      PR   NI VIRT    RES     SHR  S %CPU    %MEM    TIME+ COMMAND
4557 caster    20   0  720228  38524   7340 R 255.0   1.0  48:44.97 gorilla-linux-a
4552 caster    20   0  720612  53080   7212 S 171.0   1.3  32:00.80 gws-linux-amd64

Documentation

Index

• type BuiltinEventHandler
  • func (b BuiltinEventHandler) OnClose(socket *Conn, code uint16, reason []byte)
  • func (b BuiltinEventHandler) OnError(socket *Conn, err error)
  • func (b BuiltinEventHandler) OnMessage(socket *Conn, message *Message)
  • func (b BuiltinEventHandler) OnOpen(socket *Conn)
  • func (b BuiltinEventHandler) OnPing(socket *Conn, payload []byte)
  • func (b BuiltinEventHandler) OnPong(socket *Conn, payload []byte)
• 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) 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) error
  • func (c *Conn) SetReadDeadline(t time.Time) error
  • func (c *Conn) SetWriteDeadline(t time.Time) error
  • func (c *Conn) WriteAsync(opcode Opcode, payload []byte) error
  • func (c *Conn) WriteClose(code uint16, reason []byte)
  • func (c *Conn) WriteMessage(opcode Opcode, payload []byte) error
  • func (c *Conn) WritePing(payload []byte) error
  • func (c *Conn) WritePong(payload []byte) error
  • func (c *Conn) WriteString(s string) error
• type Event
• type Message
  • func (c *Message) Close()
  • func (c *Message) Read(p []byte) (n int, err error)
• type Opcode
  • func (c Opcode) IsDataFrame() bool
• type Option
  • func WithAsyncReadEnabled() Option
  • func WithAsyncReadGoLimit(limit int) Option
  • func WithAsyncWriteCap(capacity int) Option
  • func WithCheckOrigin(f func(r *Request) bool) Option
  • func WithCheckTextEncoding() Option
  • func WithCompress(level int, threshold int) Option
  • func WithEventHandler(eventHandler Event) 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)
  • func (c *Upgrader) Initialize()

Types

type BuiltinEventHandler

type BuiltinEventHandler struct{}

func (BuiltinEventHandler) OnClose

func (b BuiltinEventHandler) OnClose(socket *Conn, code uint16, reason []byte)

func (BuiltinEventHandler) OnError

func (b BuiltinEventHandler) OnError(socket *Conn, err error)

func (BuiltinEventHandler) OnMessage

func (b BuiltinEventHandler) OnMessage(socket *Conn, message *Message)

func (BuiltinEventHandler) OnOpen

func (b BuiltinEventHandler) OnOpen(socket *Conn)

func (BuiltinEventHandler) OnPing

func (b BuiltinEventHandler) OnPing(socket *Conn, payload []byte)

func (BuiltinEventHandler) OnPong

func (b BuiltinEventHandler) OnPong(socket *Conn, payload []byte)

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

SetDeadline sets deadline

func (*Conn) SetReadDeadline

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

SetReadDeadline sets read deadline

func (*Conn) SetWriteDeadline

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

SetWriteDeadline sets write deadline

func (*Conn) WriteAsync

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

WriteAsync 异步写入消息, 适合广播等需要非阻塞的场景 asynchronous write messages, suitable for non-blocking scenarios such as broadcasting

func (*Conn) WriteClose

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

WriteClose proactively close the connection code: https://developer.mozilla.org/zh-CN/docs/Web/API/CloseEvent#status_codes 通过emitError发送关闭帧, 将连接状态置为关闭, 用于服务端主动断开连接 没有特殊原因的话, 建议code=0, reason=nil

func (*Conn) WriteMessage

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

WriteMessage writes message 发送消息

func (*Conn) WritePing

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

WritePing write ping frame

func (*Conn) WritePong

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

WritePong write pong frame

func (*Conn) WriteString

func (c *Conn) WriteString(s string) error

WriteString write text frame

type Event

type Event interface {
	// 建立连接事件
	OnOpen(socket *Conn)

	// 错误事件
	// IO错误, 协议错误, 压缩解压错误...
	OnError(socket *Conn, err error)

	// 关闭事件
	// 另一端发送了关闭帧
	OnClose(socket *Conn, code uint16, reason []byte)

	// 心跳探测事件
	OnPing(socket *Conn, payload []byte)

	// 心跳响应事件
	OnPong(socket *Conn, payload []byte)

	// 消息事件
	// 如果开启了AsyncReadEnabled, 可以在一个连接里面并行处理多个请求
	OnMessage(socket *Conn, message *Message)
}

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

type Message

type Message struct {
	Opcode Opcode        // 帧状态码
	Data   *bytes.Buffer // 数据缓冲
}

func (*Message) Close

func (c *Message) Close()

Close recycle buffer

func (*Message) Read

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

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 WithAsyncReadEnabled

func WithAsyncReadEnabled() Option

WithAsyncReadEnabled 开启异步读功能, 并行地调用onmessage, 并发度会受到AsyncReadGoLimit的限制. enable asynchronous read, call onmessage concurrently, concurrency is limited by AsyncReadGoLimit.

func WithAsyncReadGoLimit

func WithAsyncReadGoLimit(limit int) Option

WithAsyncReadGoLimit 并行处理消息的最大协程数量限制 limit on the maximum number of concurrently processed messages

func WithAsyncWriteCap

func WithAsyncWriteCap(capacity int) Option

WithAsyncWriteCap 异步非阻塞写入的容量限制, 超过限制的消息会被丢弃 capacity limit for asynchronous non-blocking writes, messages exceeding the limit will be discarded

func WithCheckOrigin

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

WithCheckOrigin 检查请求来源, 进行鉴权. check request origin

func WithCheckTextEncoding

func WithCheckTextEncoding() Option

WithCheckTextEncoding 检查文本utf8编码, 关闭性能会更好. set text encoding checking

func WithCompress

func WithCompress(level int, threshold int) Option

WithCompress 设置数据压缩. 是否压缩, 压缩级别和阈值, 低于阈值的数据不会被压缩. set data compression. set the compression level and the threshold value, below which the data will not be compressed.

func WithEventHandler

func WithEventHandler(eventHandler Event) Option

WithEventHandler 设置事件处理器 set event handler

func WithMaxContentLength

func WithMaxContentLength(n int) Option

WithMaxContentLength 设置消息最大长度(字节) set max content length (byte).

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 string) (value interface{}, exist bool)
	Delete(key string)
	Store(key string, value interface{})
	Range(f func(key string, value interface{}) bool)
}

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

type Upgrader

type Upgrader struct {
	// websocket event handler
	EventHandler Event

	// whether to enable asynchronous reading. if on, onmessage will be called concurrently.
	AsyncReadEnabled bool

	// goroutine limits on concurrent read
	AsyncReadGoLimit int

	// capacity of async write queue
	// if the capacity is full, the message will be discarded
	AsyncWriteCap int

	// whether to compress data
	CompressEnabled bool

	// compress level eg: flate.BestSpeed
	CompressLevel int

	// if contentLength < compressionThreshold, it won't be compressed.
	CompressionThreshold int

	// max message size
	MaxContentLength int

	// whether to check utf8 encoding when read messages, 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

func NewUpgrader

func NewUpgrader(options ...Option) *Upgrader

func (*Upgrader) Accept

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

Accept http upgrade to websocket protocol

func (*Upgrader) Initialize

func (c *Upgrader) Initialize()

Initialize the upgrader configure 如果没有使用NewUpgrader, 需要调用此方法初始化配置