gws

package module
v1.6.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 30, 2023 License: MIT Imports: 21 Imported by: 64

README

gws

event-driven go websocket server & client

Mentioned in Awesome Go Build Status MIT licensed Go Version codecov Go Report Card

Feature
  • Event API
  • Broadcast
  • Dial via Proxy
  • IO Multiplexing
  • Concurrent Write
  • Passes WebSocket autobahn-testsuite
Attention
  • The errors returned by the gws.Conn export methods are ignored, and are handled internally
  • Transferring large files with gws tends to block the connection
Install
go get -v github.com/lxzan/gws@latest
Event
type Event interface {
	OnOpen(socket *Conn)
	OnClose(socket *Conn, err error)
	OnPing(socket *Conn, payload []byte)
	OnPong(socket *Conn, payload []byte)
	OnMessage(socket *Conn, message *Message)
}
Quick Start
package main

import "github.com/lxzan/gws"

func main() {
	gws.NewServer(new(gws.BuiltinEventHandler), nil).Run(":6666")
}
Best Practice
package main

import (
	"github.com/lxzan/gws"
	"time"
)

const PingInterval = 10 * time.Second

func main() {
	options := &gws.ServerOption{ReadAsyncEnabled: true, ReadAsyncGoLimit: 4, CompressEnabled: true}
	gws.NewServer(new(Handler), options).Run(":6666")
}

type Handler struct{}

func (c *Handler) OnOpen(socket *gws.Conn) { _ = socket.SetDeadline(time.Now().Add(2 * PingInterval)) }

func (c *Handler) OnClose(socket *gws.Conn, err error) {}

func (c *Handler) OnPing(socket *gws.Conn, payload []byte) {
	_ = socket.SetDeadline(time.Now().Add(2 * PingInterval))
	_ = socket.WritePong(nil)
}

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

func (c *Handler) OnMessage(socket *gws.Conn, message *gws.Message) {
	defer message.Close()
}
Usage
Upgrade from HTTP
package main

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

func main() {
	upgrader := gws.NewUpgrader(new(gws.BuiltinEventHandler), &gws.ServerOption{
		Authorize: func(r *http.Request, session gws.SessionStorage) bool {
			session.Store("username", r.URL.Query().Get("username"))
			return true
		},
	})

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

	if err := http.ListenAndServe(":6666", nil); err != nil {
		log.Fatalf("%v", err)
	}
}
Unix Domain Socket
  • server
package main

import (
	"log"
	"net"
	"github.com/lxzan/gws"
)

func main() {
	listener, err := net.Listen("unix", "/tmp/gws.sock")
	if err != nil {
		log.Println(err.Error())
		return
	}
	var app = gws.NewServer(new(gws.BuiltinEventHandler), nil)
	if err := app.RunListener(listener); err != nil {
		log.Println(err.Error())
	}
}
  • client
package main

import (
	"log"
	"net"
	"github.com/lxzan/gws"
)

func main() {
	conn, err := net.Dial("unix", "/tmp/gws.sock")
	if err != nil {
		log.Println(err.Error())
		return
	}

	option := gws.ClientOption{}
	socket, _, err := gws.NewClientFromConn(new(gws.BuiltinEventHandler), &option, conn)
	if err != nil {
		log.Println(err.Error())
		return
	}
	socket.ReadLoop()
}
Client Proxy
package main

import (
	"crypto/tls"
	"github.com/lxzan/gws"
	"golang.org/x/net/proxy"
	"log"
)

func main() {
	socket, _, err := gws.NewClient(new(gws.BuiltinEventHandler), &gws.ClientOption{
		Addr:      "wss://example.com/connect",
		TlsConfig: &tls.Config{InsecureSkipVerify: true},
		NewDialer: func() (gws.Dialer, error) {
			return proxy.SOCKS5("tcp", "127.0.0.1:1080", nil, nil)
		},
	})
	if err != nil {
		log.Println(err.Error())
		return
	}
	socket.ReadLoop()
}
Broadcast
func Broadcast(conns []*gws.Conn, opcode gws.Opcode, payload []byte) {
	var b = gws.NewBroadcaster(opcode, payload)
	defer b.Release()
	for _, item := range conns {
		_ = b.Broadcast(item)
	}
}
Autobahn Test
cd examples/autobahn
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
  • GOMAXPROCS = 4
  • Connection = 1000
  • Compress Disabled

performance

$ go test -benchmem -run=^$ -bench ^(BenchmarkConn_WriteMessage|BenchmarkConn_ReadMessage)$ github.com/lxzan/gws

goos: darwin
goarch: arm64
pkg: github.com/lxzan/gws
BenchmarkConn_WriteMessage/compress_disabled-8         	 4494459	       239.2 ns/op	       0 B/op	       0 allocs/op
BenchmarkConn_WriteMessage/compress_enabled-8          	  107365	     10726 ns/op	     509 B/op	       0 allocs/op
BenchmarkConn_ReadMessage/compress_disabled-8          	 3037701	       395.6 ns/op	     120 B/op	       3 allocs/op
BenchmarkConn_ReadMessage/compress_enabled-8           	  175388	      6355 ns/op	    7803 B/op	       7 allocs/op
PASS
ok  	github.com/lxzan/gws	5.813s
Communication

微信二维码在讨论区不定时更新

WeChat      QQ
Acknowledgments

The following project had particular influence on gws's design.

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Broadcaster added in v1.6.2 

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

func NewBroadcaster added in v1.6.2 

func NewBroadcaster(opcode Opcode, payload []byte) *Broadcaster

NewBroadcaster 相比WriteAsync, Broadcaster只会压缩一次消息, 可以节省大量CPU开销. Compared to WriteAsync, Broadcaster compresses the message only once, saving a lot of CPU overhead.

func (*Broadcaster) Broadcast added in v1.6.2 

func (c *Broadcaster) Broadcast(socket *Conn) error

Broadcast 广播 向单个客户端发送广播消息. 注意: 不要并行调用Broadcast方法 Send a broadcast message to a single client. Note: Do not call the Broadcast method in parallel.

func (*Broadcaster) Release added in v1.6.2 

func (c *Broadcaster) Release()

Release 在完成所有Broadcast之后调用Release方法释放资源. Call the Release method after all the Broadcasts have been completed to release the resources.

type BuiltinEventHandler added in v1.3.0 

type BuiltinEventHandler struct{}

func (BuiltinEventHandler) OnClose added in v1.3.0 

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

func (BuiltinEventHandler) OnMessage added in v1.3.0 

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

func (BuiltinEventHandler) OnOpen added in v1.3.0 

func (b BuiltinEventHandler) OnOpen(socket *Conn)

func (BuiltinEventHandler) OnPing added in v1.3.0 

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

func (BuiltinEventHandler) OnPong added in v1.3.0 

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

type ClientOption added in v1.4.2 

type ClientOption struct {
	// 写缓冲区的大小, v1.4.5版本此参数被废弃
	// Deprecated: Size of the write buffer, v1.4.5 version of this parameter is deprecated
	WriteBufferSize int

	ReadAsyncEnabled    bool
	ReadAsyncGoLimit    int
	ReadMaxPayloadSize  int
	ReadBufferSize      int
	WriteMaxPayloadSize int
	CompressEnabled     bool
	CompressLevel       int
	CompressThreshold   int
	CheckUtf8Enabled    bool

	// 连接地址, 例如 wss://example.com/connect
	// server address, eg: wss://example.com/connect
	Addr string

	// 额外的请求头
	// extra request header
	RequestHeader http.Header

	// 握手超时时间
	HandshakeTimeout time.Duration

	// TLS设置
	TlsConfig *tls.Config

	// 拨号器
	// 默认是返回net.Dialer实例, 也可以用于设置代理.
	// The default is to return the net.Dialer instance
	// Can also be used to set a proxy, for example
	// NewDialer: func() (proxy.Dialer, error) {
	//		return proxy.SOCKS5("tcp", "127.0.0.1:1080", nil, nil)
	// },
	NewDialer func() (Dialer, error)

	// 创建session存储空间
	// 用于自定义SessionStorage实现
	// For custom SessionStorage implementations
	NewSessionStorage func() SessionStorage
}

type CloseError added in v1.6.0 

type CloseError struct {
	Code   uint16
	Reason []byte
}

func (*CloseError) Error added in v1.6.0 

func (c *CloseError) Error() string

type Comparable added in v1.5.1 

type Comparable interface {
	string | int | int64 | int32 | uint | uint64 | uint32
}

type ConcurrentMap added in v1.2.5 

type ConcurrentMap[K Comparable, V any] struct {
	// contains filtered or unexported fields
}

func NewConcurrentMap added in v1.2.5 

func NewConcurrentMap[K Comparable, V any](segments uint64) *ConcurrentMap[K, V]

func (*ConcurrentMap[K, V]) Delete added in v1.2.5 

func (c *ConcurrentMap[K, V]) Delete(key K)

func (*ConcurrentMap[K, V]) Len added in v1.2.5 

func (c *ConcurrentMap[K, V]) Len() int

func (*ConcurrentMap[K, V]) Load added in v1.2.5 

func (c *ConcurrentMap[K, V]) Load(key K) (value V, exist bool)

func (*ConcurrentMap[K, V]) Range added in v1.2.5 

func (c *ConcurrentMap[K, V]) Range(f func(key K, value V) bool)

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

func (*ConcurrentMap[K, V]) Store added in v1.2.5 

func (c *ConcurrentMap[K, V]) Store(key K, value V)

type Config added in v1.2.0 

type Config struct {

	// 是否开启异步读, 开启的话会并行调用OnMessage
	// Whether to enable asynchronous reading, if enabled OnMessage will be called in parallel
	ReadAsyncEnabled bool

	// 异步读的最大并行协程数量
	// Maximum number of parallel concurrent processes for asynchronous reads
	ReadAsyncGoLimit int

	// 最大读取的消息内容长度
	// Maximum read message content length
	ReadMaxPayloadSize int

	// 读缓冲区的大小
	// Size of the read buffer
	ReadBufferSize int

	// 最大写入的消息内容长度
	// Maximum length of written message content
	WriteMaxPayloadSize int

	// 写缓冲区的大小, v1.4.5版本此参数被废弃
	// Deprecated: Size of the write buffer, v1.4.5 version of this parameter is deprecated
	WriteBufferSize int

	// 是否开启数据压缩
	// Whether to turn on data compression
	CompressEnabled bool

	// 压缩级别
	// Compress level
	CompressLevel int

	// 压缩阈值, 低于阈值的消息不会被压缩
	// Compression threshold, messages below the threshold will not be compressed
	CompressThreshold int

	// CompressorNum 压缩器数量
	// 数值越大竞争的概率越小, 但是会耗费大量内存, 注意取舍
	// Number of compressors
	// The higher the value the lower the probability of competition, but it will consume a lot of memory, so be careful about the trade-off
	CompressorNum int

	// 是否检查文本utf8编码, 关闭性能会好点
	// Whether to check the text utf8 encoding, turn off the performance will be better
	CheckUtf8Enabled bool
	// contains filtered or unexported fields
}

type Conn 

type Conn struct {
	// store session information
	SessionStorage SessionStorage
	// contains filtered or unexported fields
}

func NewClient added in v1.4.2 

func NewClient(handler Event, option *ClientOption) (*Conn, *http.Response, error)

NewClient

func NewClientFromConn added in v1.5.1 

func NewClientFromConn(handler Event, option *ClientOption, conn net.Conn) (*Conn, *http.Response, error)

NewClientFromConn

func (*Conn) LocalAddr added in v1.0.1 

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

func (*Conn) NetConn added in v1.2.10 

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

NetConn get tcp/tls/... conn

func (*Conn) ReadLoop added in v1.4.8 

func (c *Conn) ReadLoop()

ReadLoop start a read message loop 启动一个读消息的死循环

func (*Conn) RemoteAddr added in v1.0.1 

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

func (*Conn) SetDeadline 

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

SetDeadline sets deadline

func (*Conn) SetNoDelay added in v1.6.2 

func (c *Conn) SetNoDelay(noDelay bool) error

SetNoDelay controls whether the operating system should delay packet transmission in hopes of sending fewer packets (Nagle's algorithm). The default is true (no delay), meaning that data is sent as soon as possible after a Write.

func (*Conn) SetReadDeadline added in v1.1.2 

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

SetReadDeadline sets read deadline

func (*Conn) SetWriteDeadline added in v1.1.2 

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

SetWriteDeadline sets write deadline

func (*Conn) WriteAsync added in v1.3.0 

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

WriteAsync 异步非阻塞地写入消息 Write messages asynchronously and non-blockingly

func (*Conn) WriteClose 

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

WriteClose code: https://developer.mozilla.org/zh-CN/docs/Web/API/CloseEvent#status_codes 通过emitError发送关闭帧, 将连接状态置为关闭, 用于服务端主动断开连接 没有特殊原因的话, 建议code=0, reason=nil Send a close frame via emitError to set the connection state to closed, for server-initiated disconnection If there is no special reason, we suggest code=0, reason=nil

func (*Conn) WriteMessage added in v1.1.0 

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

WriteMessage 发送消息

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 added in v1.2.10 

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

WriteString write text frame

type Dialer added in v1.6.0 

type Dialer interface {
	Dial(network, addr string) (c net.Conn, err error)
}

type Event added in v1.1.2 

type Event interface {
	// 建立连接事件
	// WebSocket connection was successfully established
	OnOpen(socket *Conn)

	// 关闭事件
	// 接收到了网络连接另一端发送的关闭帧, 或者IO过程中出现错误主动断开连接
	// 如果是前者, err可以断言为*CloseError
	// Received a close frame from the other end of the network connection, or disconnected voluntarily due to an error in the IO process
	// In the former case, err can be asserted as *CloseError
	OnClose(socket *Conn, err error)

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

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

	// 消息事件
	// 如果开启了ReadAsyncEnabled, 会并行地调用OnMessage; 没有做recover处理.
	// If ReadAsyncEnabled is enabled, OnMessage is called in parallel. No recover is done.
	OnMessage(socket *Conn, message *Message)
}

WebSocket Event

type Message 

type Message struct {

	// 操作码
	Opcode Opcode

	// 消息内容
	Data *bytes.Buffer
	// contains filtered or unexported fields
}

func (*Message) Bytes 

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

func (*Message) Close 

func (c *Message) Close() error

Close recycle buffer

func (*Message) Read added in v1.1.0 

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
)

type Server added in v1.4.7 

type Server struct {

	// OnError 接收握手过程中产生的错误回调
	// Receive error callbacks generated during the handshake
	OnError func(conn net.Conn, err error)

	// OnRequest
	OnRequest func(socket *Conn, request *http.Request)
	// contains filtered or unexported fields
}

func NewServer added in v1.4.7 

func NewServer(eventHandler Event, option *ServerOption) *Server

NewServer 创建websocket服务器 create a websocket server

func (*Server) Run added in v1.4.7 

func (c *Server) Run(addr string) error

Run runs ws server addr: Address of the listener

func (*Server) RunListener added in v1.4.9 

func (c *Server) RunListener(listener net.Listener) error

func (*Server) RunTLS added in v1.4.7 

func (c *Server) RunTLS(addr string, certFile, keyFile string) error

RunTLS runs wss server addr: Address of the listener config: tls config

type ServerOption added in v1.4.0 

type ServerOption struct {

	// 写缓冲区的大小, v1.4.5版本此参数被废弃
	// Deprecated: Size of the write buffer, v1.4.5 version of this parameter is deprecated
	WriteBufferSize int

	ReadAsyncEnabled    bool
	ReadAsyncGoLimit    int
	ReadMaxPayloadSize  int
	ReadBufferSize      int
	WriteMaxPayloadSize int
	CompressEnabled     bool
	CompressLevel       int
	CompressThreshold   int
	CompressorNum       int
	CheckUtf8Enabled    bool

	// 握手超时时间
	HandshakeTimeout time.Duration

	// WebSocket子协议, 一般不需要设置
	// WebSocket subprotocol, usually no need to set
	Subprotocols []string

	// 连接握手时添加的额外的响应头, 如果客户端不支持就不要传
	// 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

	// 鉴权
	// Authentication of requests for connection establishment
	Authorize func(r *http.Request, session SessionStorage) bool

	// 创建session存储空间
	// 用于自定义SessionStorage实现
	// For custom SessionStorage implementations
	NewSessionStorage func() SessionStorage
	// contains filtered or unexported fields
}

type SessionStorage added in v1.2.3 

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

type Upgrader 

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

func NewUpgrader added in v1.2.11 

func NewUpgrader(eventHandler Event, option *ServerOption) *Upgrader

func (*Upgrader) Upgrade 

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

Upgrade http upgrade to websocket protocol

Source Files

View all Source files

Directories

Path Synopsis
autobahn
client command
reporter command
server command
examples
chatroom command
client command
echo command
push command
wss command

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.