Merge pull request #13 from lxzan/dev

v1.5.0

.github/workflows/go.yml

@@ -19,6 +19,6 @@ jobs:
  - name: Set up Go
  uses: actions/setup-go@v3
  with:
- go-version: 1.16
+ go-version: 1.18
  - name: Test
  run: go test -v ./...

README.md

@@ -12,7 +12,7 @@
 
 [4]: LICENSE
 
-[5]: https://img.shields.io/badge/go-%3E%3D1.16-30dff3?style=flat-square&logo=go
+[5]: https://img.shields.io/badge/go-%3E%3D1.18-30dff3?style=flat-square&logo=go
 
 [6]: https://github.com/lxzan/gws
 
@@ -34,14 +34,15 @@
   - [Upgrade from HTTP](#upgrade-from-http)
   - [Unix Domain Socket](#unix-domain-socket)
   - [Broadcast](#broadcast)
+  - [Write JSON](#write-json)
  - [Autobahn Test](#autobahn-test)
  - [Benchmark](#benchmark)
  - [Communication](#communication)
  - [Acknowledgments](#acknowledgments)
 
 ### Highlight
 
-- No dependency
+- Single dependency
 - IO multiplexing support, concurrent message processing and asynchronous non-blocking message writing
 - High IOPS and low latency, low CPU usage
 - Support fast parsing WebSocket protocol directly from TCP, faster handshake, 30% lower memory usage
@@ -207,6 +208,11 @@ func Broadcast(conns []*gws.Conn, opcode gws.Opcode, payload []byte) {
 }
 ```
 
+#### Write JSON
+```go
+socket.WriteAny(gws.JsonCodec, gws.OpcodeText, data)
+```
+
 ### Autobahn Test
 
 ```bash

compress.go

@@ -2,25 +2,55 @@ package gws
 
 import (
  "bytes"
- "compress/flate"
  "encoding/binary"
+ "github.com/klauspost/compress/flate"
  "github.com/lxzan/gws/internal"
  "io"
  "math"
+ "sync"
+ "sync/atomic"
 )
 
+const numCompressor = 32
+
+type compressors struct {
+ sync.RWMutex
+ serial uint64
+ compressors [12][numCompressor]*compressor
+}
+
+func (c *compressors) Select(level int) *compressor {
+ var i = level + 2
+ var j = atomic.AddUint64(&c.serial, 1) & (numCompressor - 1)
+ c.RLock()
+ var cps = c.compressors[i][j]
+ c.RUnlock()
+
+ if cps == nil {
+ c.Lock()
+ cps = newCompressor(level)
+ c.compressors[i][j] = cps
+ c.Unlock()
+ }
+ return cps
+}
+
 func newCompressor(level int) *compressor {
  fw, _ := flate.NewWriter(nil, level)
- return &compressor{fw: fw}
+ return &compressor{fw: fw, mu: &sync.Mutex{}}
 }
 
 // 压缩器
 type compressor struct {
+ mu *sync.Mutex
  fw *flate.Writer
 }
 
 // Compress 压缩
 func (c *compressor) Compress(content []byte, buf *bytes.Buffer) error {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
  c.fw.Reset(buf)
  if err := internal.WriteN(c.fw, content, len(content)); err != nil {
  return err
@@ -37,20 +67,64 @@ func (c *compressor) Compress(content []byte, buf *bytes.Buffer) error {
  return nil
 }
 
-func newDecompressor() *decompressor { return &decompressor{fr: flate.NewReader(nil)} }
+func (c *compressor) CompressAny(codec Codec, v interface{}, buf *bytes.Buffer) error {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ c.fw.Reset(buf)
+ if err := codec.NewEncoder(c.fw).Encode(v); err != nil {
+ return err
+ }
+ if err := c.fw.Flush(); err != nil {
+ return err
+ }
+ if n := buf.Len(); n >= 4 {
+ compressedContent := buf.Bytes()
+ if tail := compressedContent[n-4:]; binary.BigEndian.Uint32(tail) == math.MaxUint16 {
+ buf.Truncate(n - 4)
+ }
+ }
+ return nil
+}
+
+type decompressors struct {
+ serial uint64
+ decompressors [numCompressor]*decompressor
+}
+
+func (c *decompressors) init() *decompressors {
+ for i, _ := range c.decompressors {
+ c.decompressors[i] = newDecompressor()
+ }
+ return c
+}
+
+func (c *decompressors) Select() *decompressor {
+ var index = atomic.AddUint64(&c.serial, 1) & (numCompressor - 1)
+ return c.decompressors[index]
+}
+
+func newDecompressor() *decompressor {
+ return &decompressor{fr: flate.NewReader(nil), mu: &sync.Mutex{}}
+}
 
 type decompressor struct {
- fr io.ReadCloser
+ mu *sync.Mutex
+ fr io.ReadCloser
+ buffer [internal.Lv3]byte
 }
 
 // Decompress 解压
 func (c *decompressor) Decompress(payload *bytes.Buffer) (*bytes.Buffer, error) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
  _, _ = payload.Write(internal.FlateTail)
  resetter := c.fr.(flate.Resetter)
  _ = resetter.Reset(payload, nil) // must return a null pointer
 
  var buf = _bpool.Get(3 * payload.Len())
- _, err := io.Copy(buf, c.fr)
+ _, err := io.CopyBuffer(buf, c.fr, c.buffer[0:])
  _bpool.Put(payload)
  return buf, err
 }

conn.go

@@ -26,14 +26,10 @@ type Conn struct {
  config *Config
  // read buffer
  rbuf *bufio.Reader
- // flate decompressor
- decompressor *decompressor
  // continuation frame
  continuationFrame continuationFrame
  // frame header for read
  fh frameHeader
- // flate compressor
- compressor *compressor
  // WebSocket Event Handler
  handler Event
 
@@ -62,12 +58,6 @@ func serveWebSocket(isServer bool, config *Config, session SessionStorage, netCo
  readQueue: workerQueue{maxConcurrency: int32(config.ReadAsyncGoLimit), capacity: config.ReadAsyncCap},
  writeQueue: workerQueue{maxConcurrency: 1, capacity: config.WriteAsyncCap},
  }
-
- if c.compressEnabled {
- c.compressor = newCompressor(config.CompressLevel)
- c.decompressor = newDecompressor()
- }
-
  return c
 }
 

examples/chatroom/main.go

@@ -46,17 +46,17 @@ func main() {
  _, _ = writer.Write(html)
  })
 
- if err := http.ListenAndServe(":3000", nil); err != nil {
+ if err := http.ListenAndServe(":8000", nil); err != nil {
  log.Fatalf("%+v", err)
  }
 }
 
 func NewWebSocket() *WebSocket {
- return &WebSocket{sessions: gws.NewConcurrentMap(16)}
+ return &WebSocket{sessions: gws.NewConcurrentMap[string, *gws.Conn](16)}
 }
 
 type WebSocket struct {
- sessions *gws.ConcurrentMap // 使用内置的ConcurrentMap存储连接, 可以减少锁冲突
+ sessions *gws.ConcurrentMap[string, *gws.Conn] // 使用内置的ConcurrentMap存储连接, 可以减少锁冲突
 }
 
 func (c *WebSocket) getName(socket *gws.Conn) string {
@@ -69,21 +69,11 @@ func (c *WebSocket) getKey(socket *gws.Conn) string {
  return name.(string)
 }
 
-// 根据用户名获取WebSocket连接
-func (c *WebSocket) GetSocket(name string) (*gws.Conn, bool) {
- if v0, ok0 := c.sessions.Load(name); ok0 {
- if v1, ok1 := v0.(*gws.Conn); ok1 {
- return v1, true
- }
- }
- return nil, false
-}
-
 // RemoveSocket 移除WebSocket连接
 func (c *WebSocket) RemoveSocket(socket *gws.Conn) {
  name := c.getName(socket)
  key := c.getKey(socket)
- if mSocket, ok := c.GetSocket(name); ok {
+ if mSocket, ok := c.sessions.Load(name); ok {
  if mKey := c.getKey(mSocket); mKey == key {
  c.sessions.Delete(name)
  }
@@ -92,8 +82,7 @@ func (c *WebSocket) RemoveSocket(socket *gws.Conn) {
 
 func (c *WebSocket) OnOpen(socket *gws.Conn) {
  name := c.getName(socket)
- if v, ok := c.sessions.Load(name); ok {
- var conn = v.(*gws.Conn)
+ if conn, ok := c.sessions.Load(name); ok {
  conn.WriteClose(1000, []byte("connection replaced"))
  }
  socket.SetDeadline(time.Now().Add(3 * PingInterval))
@@ -134,7 +123,7 @@ func (c *WebSocket) OnMessage(socket *gws.Conn, message *gws.Message) {
 
  var input = &Input{}
  _ = json.Unmarshal(message.Data.Bytes(), input)
- if v, ok := c.sessions.Load(input.To); ok {
- v.(*gws.Conn).WriteMessage(gws.OpcodeText, message.Data.Bytes())
+ if conn, ok := c.sessions.Load(input.To); ok {
+ conn.WriteMessage(gws.OpcodeText, message.Data.Bytes())
  }
 }

go.mod

@@ -1,5 +1,14 @@
 module github.com/lxzan/gws
 
-go 1.16
+go 1.18
 
-require github.com/stretchr/testify v1.8.1
+require (
+ github.com/klauspost/compress v1.16.5
+ github.com/stretchr/testify v1.8.1
+)
+
+require (
+ github.com/davecgh/go-spew v1.1.1 // indirect
+ github.com/pmezard/go-difflib v1.0.0 // indirect
+ gopkg.in/yaml.v3 v3.0.1 // indirect
+)

go.sum

@@ -1,6 +1,8 @@
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/klauspost/compress v1.16.5 h1:IFV2oUNUzZaz+XyusxpLzpzS8Pt5rh0Z16For/djlyI=
+github.com/klauspost/compress v1.16.5/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=

init.go

@@ -0,0 +1,12 @@
+package gws
+
+import "github.com/lxzan/gws/internal"
+
+var (
+ _bpool = internal.NewBufferPool()
+ _cps = new(compressors)
+ _dps = new(decompressors).init()
+ _padding = frameHeader{}
+
+ JsonCodec = new(jsonCodec)
+)

reader.go

@@ -7,8 +7,6 @@ import (
  "github.com/lxzan/gws/internal"
 )
 
-var _bpool = internal.NewBufferPool()
-
 func (c *Conn) checkMask(enabled bool) error {
  // RFC6455: All frames sent from client to server have this bit set to 1.
  if (c.isServer && !enabled) || (!c.isServer && enabled) {
@@ -149,7 +147,7 @@ func (c *Conn) readMessage() error {
 
 func (c *Conn) emitMessage(msg *Message, compressed bool) error {
  if compressed {
- data, err := c.decompressor.Decompress(msg.Data)
+ data, err := _dps.Select().Decompress(msg.Data)
  if err != nil {
  return internal.NewError(internal.CloseInternalServerErr, err)
  }