gws

README

gws

a event driven websocket framework

Features

• event-driven api design
• use queues to process messages parallelly, with concurrency control
• middleware support
• self-implementing buffer, more memory-saving
• no dependency

Quick Start

chat room example:

server

package main

import (
	"context"
	"encoding/json"
	"github.com/lxzan/gws"
	"net/http"
	"sync"
)

type Handler struct {
	sessions sync.Map
}

func (h *Handler) deleteSession(socket *gws.Conn) {
	name, _ := socket.Storage.Get("name")
	h.sessions.Delete(name)
}

func (h *Handler) OnOpen(socket *gws.Conn) {
	name, _ := socket.Storage.Get("name")
	h.sessions.Store(name.(string), socket)

	go func(conn *gws.Conn) {
		for {
			<-conn.Context.Done()
			h.deleteSession(conn)
		}
	}(socket)
}

func (h *Handler) OnClose(socket *gws.Conn, code gws.Code, reason []byte) {}

type Request struct {
	To      string `json:"to"`
	Message string `json:"message"`
}

func (h *Handler) OnMessage(socket *gws.Conn, m *gws.Message) {
	var request Request
	json.Unmarshal(m.Bytes(), &request)
	defer m.Close()

	if receiver, ok := h.sessions.Load(request.To); ok {
		receiver.(*gws.Conn).Write(gws.OpcodeText, m.Bytes())
	}
}

func (h *Handler) OnError(socket *gws.Conn, err error) {}

func (h *Handler) OnPing(socket *gws.Conn, m []byte) {}

func (h *Handler) OnPong(socket *gws.Conn, m []byte) {}

func main() {
	var upgrader = gws.Upgrader{
		ServerOptions: &gws.ServerOptions{
			LogEnabled:      true,
			CompressEnabled: false,
		},
		CheckOrigin: func(r *gws.Request) bool {
			r.Storage.Put("name", r.URL.Query().Get("name"))
			return true
		},
	}

	var handler = &Handler{sessions: sync.Map{}}
	var ctx = context.Background()

	http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
		upgrader.Upgrade(ctx, w, r, handler)
	})

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

client(browser console)

let ws1 = new WebSocket('ws://127.0.0.1:3000/ws?name=caster');
let ws2 = new WebSocket('ws://127.0.0.1:3000/ws?name=lancer');
ws1.send('{"to": "lancer", "msg": "Hello! I am caster"}');

API

type EventHandler interface {
    OnOpen(socket *Conn)
    OnClose(socket *Conn, code Code, reason []byte)
    OnMessage(socket *Conn, m *Message)
    OnError(socket *Conn, err error)
    OnPing(socket *Conn, m []byte)
    OnPong(socket *Conn, m []byte)
}

Usage

Middleware

• use internal middleware

var upgrader = gws.Upgrader{}
upgrader.Use(gws.Recovery(func(exception interface{}) {
    fmt.Printf("%v", exception)
}))

• write a middleware

upgrader.Use(func (socket *gws.Conn, msg *gws.Message) {
    var t0 = time.Now().UnixNano()
    msg.Next(socket)
    var t1 = time.Now().UnixNano()
    fmt.Printf("cost=%dms\n", (t1-t0)/1000000)
})

Heartbeat

• Sever Side Heartbeat

func (h *Handler) OnOpen(socket *gws.Conn) {
    go func (ws *gws.Conn) {
        ticker := time.NewTicker(15 * time.Second)
        defer ticker.Stop()
    
        for {
            select {
            case <-ticker.C:
            	ws.WritePing(nil)
            case <-ws.Context.Done():
            	return
            }
        }

    }(socket)
}

func (h *Handler) OnPong(socket *gws.Conn, m []byte) {
    _ = socket.SetDeadline(30 * time.Second)
}

• Client Side Heartbeat

func (h *Handler) OnPing(socket *gws.Conn, m []byte) {
    socket.WritePong(nil)
    _ = socket.SetDeadline(30 * time.Second)
}

Documentation

Index

• Variables
• type Code
  • func (c Code) Bytes() []byte
  • func (c Code) Error() string
  • func (c Code) Uint16() uint16
• type Conn
  • func (c *Conn) Close(code Code, reason []byte) error
  • func (c *Conn) LocalAddr() net.Addr
  • func (c *Conn) RemoteAddr() net.Addr
  • func (c *Conn) SetDeadline(d time.Duration) error
  • func (c *Conn) Write(messageType Opcode, content []byte)
  • func (c *Conn) WritePing(payload []byte)
  • func (c *Conn) WritePong(payload []byte)
• type EventHandler
• type HandlerFunc
  • func RateLimiter(d time.Duration, n int64) HandlerFunc
  • func Recovery(exceptionHandler func(exception interface{})) HandlerFunc
• type Message
  • func NewMessage(compressed bool, messageType Opcode, data *internal.Buffer) *Message
  • func (c *Message) Abort(socket *Conn)
  • func (c *Message) Bytes() []byte
  • func (c *Message) Close()
  • func (c *Message) MessageType() Opcode
  • func (c *Message) Next(socket *Conn)
• type Opcode
• type Request
• type ServerOptions
• type Upgrader
  • func (c *Upgrader) Upgrade(ctx context.Context, w http.ResponseWriter, r *http.Request, ...) (*Conn, error)
  • func (c *Upgrader) Use(handlers ...HandlerFunc)

Variables

var (
	ERR_CheckOrigin        = errors.New("check origin error")
	ERR_WebSocketHandshake = errors.New("websocket handshake error")
)

Types

type Code

type Code uint16

const (
	// 正常关闭; 无论为何目的而创建, 该链接都已成功完成任务.
	CloseNormalClosure Code = 1000

	// 终端离开, 可能因为服务端错误, 也可能因为浏览器正从打开连接的页面跳转离开.
	CloseGoingAway Code = 1001

	// 由于协议错误而中断连接.
	CloseProtocolError Code = 1002

	// 由于接收到不允许的数据类型而断开连接 (如仅接收文本数据的终端接收到了二进制数据).
	CloseUnsupported Code = 1003

	// 保留. 表示没有收到预期的状态码.
	CloseNoStatusReceived Code = 1005

	// 保留. 用于期望收到状态码时连接非正常关闭 (也就是说, 没有发送关闭帧).
	CloseAbnormalClosure Code = 1006

	// 由于收到了格式不符的数据而断开连接 (如文本消息中包含了非 UTF-8 数据).
	CloseUnsupportedData Code = 1007

	// 由于收到不符合约定的数据而断开连接. 这是一个通用状态码, 用于不适合使用 1003 和 1009 状态码的场景.
	ClosePolicyViolation Code = 1008

	// 由于收到过大的数据帧而断开连接.
	CloseMessageTooLarge Code = 1009

	// 客户端期望服务器商定一个或多个拓展, 但服务器没有处理, 因此客户端断开连接.
	CloseMissingExtension Code = 1010

	// 客户端由于遇到没有预料的情况阻止其完成请求, 因此服务端断开连接.
	CloseInternalServerErr Code = 1011

	// 服务器由于重启而断开连接. [Ref]
	CloseServiceRestart Code = 1012

	// 服务器由于临时原因断开连接, 如服务器过载因此断开一部分客户端连接. [Ref]
	CloseTryAgainLater Code = 1013

	// 保留. 表示连接由于无法完成 TLS 握手而关闭 (例如无法验证服务器证书).
	CloseTLSHandshake Code = 1015
)

func (Code) Bytes

func (c Code) Bytes() []byte

func (Code) Error

func (c Code) Error() string

func (Code) Uint16

func (c Code) Uint16() uint16

type Conn

type Conn struct {
	// context
	Context context.Context
	// store session information
	Storage *internal.Map
	// contains filtered or unexported fields
}

func (*Conn) Close

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

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(d time.Duration) error

set connection deadline

func (*Conn) Write

func (c *Conn) Write(messageType Opcode, content []byte)

发送消息 send a message

func (*Conn) WritePing

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

send ping frame

func (*Conn) WritePong

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

send pong frame

type EventHandler

type EventHandler interface {
	OnOpen(socket *Conn)
	OnClose(socket *Conn, code Code, reason []byte)
	OnMessage(socket *Conn, m *Message)
	OnError(socket *Conn, err error)
	OnPing(socket *Conn, m []byte)
	OnPong(socket *Conn, m []byte)
}

type HandlerFunc

type HandlerFunc func(socket *Conn, msg *Message)

func RateLimiter

func RateLimiter(d time.Duration, n int64) HandlerFunc

global rate limiter if d=1min and n=100, max speed is 100/min

func Recovery

func Recovery(exceptionHandler func(exception interface{})) HandlerFunc

type Message

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

func NewMessage

func NewMessage(compressed bool, messageType Opcode, data *internal.Buffer) *Message

func (*Message) Abort

func (c *Message) Abort(socket *Conn)

abort the next handlerFuncs, but previous handlerFuncs will be executed

func (*Message) Bytes

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

func (*Message) Close

func (c *Message) Close()

func (*Message) MessageType

func (c *Message) MessageType() Opcode

func (*Message) Next

func (c *Message) Next(socket *Conn)

call next handler function

type Opcode

type Opcode uint8

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

type Request

type Request struct {
	*http.Request               // http request
	Storage       *internal.Map // store user session
}

type ServerOptions

type ServerOptions struct {
	// whether to show error log, dv=true
	LogEnabled bool

	// whether to compress data, dv = false
	CompressEnabled bool

	// compress level eg: flate.BestSpeed
	CompressLevel int

	// websocket  handshake timeout, dv=3s
	HandshakeTimeout time.Duration

	// max message length, dv=1024*1024 (1MB)
	MaxContentLength int

	// number of concurrently processed messages allowed by the connection, dv=4
	// Concurrency=pow(2, n), eg: 4, 8, 16...
	Concurrency uint8

	// read frame timeout, dv=5s
	ReadTimeout time.Duration

	// write frame timeout, dv=5s
	WriteTimeout time.Duration
}

dv means default value

type Upgrader

type Upgrader struct {
	*ServerOptions             // config
	Header         http.Header // set response header

	CheckOrigin func(r *Request) bool // filter user request
	// contains filtered or unexported fields
}

func (*Upgrader) Upgrade

func (c *Upgrader) Upgrade(ctx context.Context, w http.ResponseWriter, r *http.Request, handler EventHandler) (*Conn, error)

http protocol upgrade to websocket

func (*Upgrader) Use

func (c *Upgrader) Use(handlers ...HandlerFunc)

use middleware