gws

README

gws

minimal websocket server

Highlight

• No dependency
• 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
• You need to manage your own message handling coroutine

Quick Start

package main

import (
	"context"
	"errors"
	"github.com/lxzan/gws"
	"net/http"
	"os"
	"os/signal"
	"strconv"
	"syscall"
	"time"
)

func main() {
	var upgrader = gws.Upgrader{}
	var handler = new(WebSocketHandler)
	ctx, cancel := context.WithCancel(context.Background())

	http.HandleFunc("/connect", func(writer http.ResponseWriter, request *http.Request) {
		socket, err := upgrader.Upgrade(ctx, writer, request)
		if err != nil {
			return
		}

		ticker := time.NewTicker(15 * time.Second)
		defer func() {
			socket.Close()
			ticker.Stop()
		}()

		handler.OnOpen(socket)
		for {
			select {
			case <-ctx.Done():
				handler.OnError(socket, gws.CloseServiceRestart)
				return
			case <-ticker.C:
				socket.WriteMessage(gws.OpcodePing, nil)
			case msg := <-socket.ReadMessage():
				if err := msg.Err(); err != nil {
					handler.OnError(socket, err)
					return
				}

				switch msg.Typ() {
				case gws.OpcodeText, gws.OpcodeBinary:
					handler.OnMessage(socket, msg)
				case gws.OpcodePing:
					handler.OnPing(socket, msg.Bytes())
				case gws.OpcodePong:
					handler.OnPong(socket, msg.Bytes())
				default:
					handler.OnError(socket, errors.New("unexpected opcode: "+strconv.Itoa(int(msg.Typ()))))
					return
				}
			}
		}
	})

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

	quit := make(chan os.Signal)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	<-quit
	cancel()
	time.Sleep(100 * time.Millisecond)
}

type WebSocketHandler struct{}

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

func (c *WebSocketHandler) OnMessage(socket *gws.Conn, m *gws.Message) {
	defer m.Close()
	println(string(m.Bytes()))
}

func (c *WebSocketHandler) OnError(socket *gws.Conn, err error) {
	println(err.Error())
}

func (c *WebSocketHandler) OnPing(socket *gws.Conn, m []byte) {
	println("onping")
}

func (c *WebSocketHandler) OnPong(socket *gws.Conn, m []byte) {
	println("onpong")
}

Documentation

Index

• Constants
• Variables
• type CloseCode
  • func (c CloseCode) Bytes() []byte
  • func (c CloseCode) Error() string
  • func (c CloseCode) Uint16() uint16
• type Conn
  • func (c *Conn) Close() error
  • func (c *Conn) FlushWriter()
  • func (c *Conn) LocalAddr() net.Addr
  • func (c *Conn) ReadMessage() <-chan *Message
  • func (c *Conn) RemoteAddr() net.Addr
  • func (c *Conn) SetDeadline(d time.Duration) error
  • func (c *Conn) WriteBatch(messageType Opcode, content []byte)
  • func (c *Conn) WriteClose(code CloseCode, reason []byte)
  • func (c *Conn) WriteMessage(messageType Opcode, content []byte)
  • func (c *Conn) WritePing(payload []byte)
  • func (c *Conn) WritePong(payload []byte)
• type EventHandler
• type Message
  • func (c *Message) Bytes() []byte
  • func (c *Message) Close() error
  • func (c *Message) Err() error
  • func (c *Message) Read(p []byte) (n int, err error)
  • func (c *Message) Typ() Opcode
• type Opcode
• type Request
• type Upgrader
  • func (c *Upgrader) Upgrade(ctx context.Context, w http.ResponseWriter, r *http.Request) (*Conn, error)

Constants

const (
	DefaultMessageChannelBufferSize = 16
	DefaultHandshakeTimeout         = 5 * time.Second
	DefaultReadTimeout              = 30 * time.Second
	DefaultWriteTimeout             = 30 * time.Second
	DefaultCompressLevel            = flate.BestSpeed
	DefaultMaxContentLength         = 1 * 1024 * 1024 // 1MiB
)

Variables

var (
	ErrCheckOrigin = errors.New("check origin error")
	ErrHandshake   = errors.New("handshake error")
)

Types

type CloseCode

type CloseCode uint16

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

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

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

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

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

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

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

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

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

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

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

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

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

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

func (CloseCode) Bytes

func (c CloseCode) Bytes() []byte

func (CloseCode) Error

func (c CloseCode) Error() string

func (CloseCode) Uint16

func (c CloseCode) Uint16() uint16

type Conn

type Conn struct {

	// store session information
	Storage *internal.Map
	// contains filtered or unexported fields
}

func (*Conn) Close

func (c *Conn) Close() error

func (*Conn) FlushWriter

func (c *Conn) FlushWriter()

FlushWriter 刷新写入缓冲区 flush write buffer

func (*Conn) LocalAddr

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

func (*Conn) ReadMessage

func (c *Conn) ReadMessage() <-chan *Message

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

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

WriteBatch 批量写入消息，最后一次写入后需要调用FlushWriter

func (*Conn) WriteClose

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

WriteClose send close frame 发送关闭帧

func (*Conn) WriteMessage

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

WriteMessage send message 发送消息

func (*Conn) WritePing

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

func (*Conn) WritePong

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

type EventHandler

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

type Message

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

func (*Message) Bytes

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

func (*Message) Close

func (c *Message) Close() error

func (*Message) Err

func (c *Message) Err() error

func (*Message) Read

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

func (*Message) Typ

func (c *Message) Typ() Opcode

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 Upgrader

type Upgrader struct {
	// 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 size, dv=1024*1024 (1MiB)
	MaxContentLength int

	// message channel buffer size, dv=16
	MessageChannelBufferSize int

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

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

	// filter user request
	CheckOrigin func(r *Request) bool
}

func (*Upgrader) Upgrade

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

http protocol upgrade to websocket