format imports

README.md

@@ -17,7 +17,7 @@
 
 GWS (Go WebSocket) is a very simple, fast, reliable and feature-rich WebSocket implementation written in Go. It is
 designed to be used in highly-concurrent environments, and it is suitable for
-building `API`, `PROXY`, `GAME`, `Live Video`, `MESSAGE`, etc. It supports both server and client side with a simple API
+building `API`, `Proxy`, `Game`, `Live Video`, `Message`, etc. It supports both server and client side with a simple API
 which mean you can easily write a server or client by yourself.
 
 GWS developed base on Event-Driven model. every connection has a goroutine to handle the event, and the event is able
@@ -26,19 +26,17 @@ to be processed in a non-blocking way.
 ### Why GWS
 
 - <font size=3>Simplicity and Ease of Use</font>
-    - **User-Friendly API**: Straightforward and easy-to-understand API, making server and client setup hassle-free.
+    - **User-Friendly**: Simple and clear `WebSocket` Event API design makes server-client interaction easy.
     - **Code Efficiency**: Minimizes the amount of code needed to implement complex WebSocket solutions.
 
 - <font size=3>High-Performance</font>
-    - **Zero Allocs IO**: Built-in multi-level memory pool to minimize dynamic memory allocation during reads and
-      writes.
-    - **Optimized for Speed**: Designed for rapid data transmission and reception, ideal for time-sensitive
+    - **High IOPS Low Latency**: Designed for rapid data transmission and reception, ideal for time-sensitive
       applications.
+    - **Low Memory Usage**: Highly optimized memory multiplexing system to minimize memory usage and reduce your cost of ownership.
 
 - <font size=3>Reliability and Stability</font>
-    - **Event-Driven Architecture**: Ensures stable performance even in highly concurrent environments.
     - **Robust Error Handling**: Advanced mechanisms to manage and mitigate errors, ensuring continuous operation.
-
+    - **Well-Developed Test Cases**: Passed all `Autobahn` test cases, fully compliant with `RFC 6455`. 99% unit test coverage, covering almost all conditional branches.
 ### Benchmark
 
 #### IOPS (Echo Server)
@@ -81,6 +79,7 @@ PASS
 	- [KCP](#kcp)
 	- [Proxy](#proxy)
 	- [Broadcast](#broadcast)
+	- [Pub / Sub](#pub--sub)
 - [Autobahn Test](#autobahn-test)
 - [Communication](#communication)
 - [Acknowledgments](#acknowledgments)
@@ -110,11 +109,11 @@ go get -v github.com/lxzan/gws@latest
 
 ```go
 type Event interface {
-    OnOpen(socket *Conn)                        // the connection is established
+    OnOpen(socket *Conn)                        // connection is established
     OnClose(socket *Conn, err error)            // received a close frame or I/O error occurs
-    OnPing(socket *Conn, payload []byte)        // receive a ping frame
-    OnPong(socket *Conn, payload []byte)        // receive a pong frame
-    OnMessage(socket *Conn, message *Message)   // receive a text/binary frame
+    OnPing(socket *Conn, payload []byte)        // received a ping frame
+    OnPong(socket *Conn, payload []byte)        // received a pong frame
+    OnMessage(socket *Conn, message *Message)   // received a text/binary frame
 }
 ```
 
@@ -287,6 +286,37 @@ func Broadcast(conns []*gws.Conn, opcode gws.Opcode, payload []byte) {
 }
 ```
 
+#### Pub / Sub
+
+```go
+package main
+
+import (
+	"github.com/lxzan/event_emitter"
+	"github.com/lxzan/gws"
+)
+
+type Socket struct{ *gws.Conn }
+
+// GetSubscriberID gets the subscription ID, which needs to be unique.
+func (c *Socket) GetSubscriberID() int64 {
+	userId, _ := c.Session().Load("userId")
+	return userId.(int64)
+}
+
+func Sub(em *event_emitter.EventEmitter[*Socket], topic string, socket *Socket) {
+	em.Subscribe(socket, topic, func(subscriber *Socket, msg any) {
+		_ = msg.(*gws.Broadcaster).Broadcast(subscriber.Conn)
+	})
+}
+
+func Pub(em *event_emitter.EventEmitter[*Socket], topic string, op gws.Opcode, msg []byte) {
+	var broadcaster = gws.NewBroadcaster(op, msg)
+	defer broadcaster.Close()
+	em.Publish(topic, broadcaster)
+}
+```
+
 ### Autobahn Test
 
 ```bash

README_CN.md

@@ -19,16 +19,16 @@ GWS（Go WebSocket）是一个用 Go 编写的非常简单、快速、可靠且
 ### 为什么选择 GWS
 
 - <font size=3>简单易用</font>
-    - **用户友好的 API 设计**: 简单易懂的应用程序接口，让服务器和客户端的设置变得轻松简单。
-    - **编码效率**: 最大限度地减少实施复杂的 WebSocket 解决方案所需的代码量。
+    - **用户友好**: 简单明了的 `WebSocket` 事件接口设计，让服务器和客户端的交互变得轻松简单.
+    - **编码效率**: 最大限度地减少实施复杂的解决方案所需的代码量.
 
-- <font size=3>性能良好</font>
-    - **零动态内存分配 I/O**: 内置多级内存池，可最大限度地减少读写过程中的动态内存分配。
-    - **性能优化**: 专为快速传输和接收数据而设计，是时间敏感型应用的理想之选。
+- <font size=3>性能出众</font>
+    - **高吞吐低延迟**: 专为快速传输和接收数据而设计，是时间敏感型应用的理想之选.
+    - **低内存占用**: 高度优化的内存复用系统, 最大限度降低内存占用，降低您的使用成本.
 
 - <font size=3>稳定可靠</font>
-    - **事件驱动式架构**: 即使在高度并发的环境中，也能确保稳定的性能。
-    - **健壮的错误处理**: 管理和减少错误的先进机制，确保持续运行。
+    - **健壮的错误处理**: 管理和减少错误的先进机制，确保持续运行.
+    - **完善的测试用例**: 通过了所有 `Autobahn` 测试用例, 完全符合 `RFC 6455` 标准. 单元测试覆盖率达到99%, 几乎覆盖所有条件分支.
 
 ### 基准测试
 
@@ -72,6 +72,7 @@ PASS
 	- [KCP](#kcp)
 	- [代理](#代理)
 	- [广播](#广播)
+	- [发布订阅](#发布订阅)
 - [Autobahn 测试](#autobahn-测试)
 - [交流](#交流)
 - [致谢](#致谢)
@@ -102,11 +103,11 @@ go get -v github.com/lxzan/gws@latest
 
 ```go
 type Event interface {
-    OnOpen(socket *Conn)                        // the connection is established
+    OnOpen(socket *Conn)                        // connection is established
     OnClose(socket *Conn, err error)            // received a close frame or I/O error occurs
-    OnPing(socket *Conn, payload []byte)        // receive a ping frame
-    OnPong(socket *Conn, payload []byte)        // receive a pong frame
-    OnMessage(socket *Conn, message *Message)   // receive a text/binary frame
+    OnPing(socket *Conn, payload []byte)        // received a ping frame
+    OnPong(socket *Conn, payload []byte)        // received a pong frame
+    OnMessage(socket *Conn, message *Message)   // received a text/binary frame
 }
 ```
 
@@ -279,6 +280,37 @@ func Broadcast(conns []*gws.Conn, opcode gws.Opcode, payload []byte) {
 }
 ```
 
+#### 发布订阅
+
+```go
+package main
+
+import (
+	"github.com/lxzan/event_emitter"
+	"github.com/lxzan/gws"
+)
+
+type Socket struct{ *gws.Conn }
+
+// GetSubscriberID 获取订阅ID, 需要保证唯一
+func (c *Socket) GetSubscriberID() int64 {
+	userId, _ := c.Session().Load("userId")
+	return userId.(int64)
+}
+
+func Sub(em *event_emitter.EventEmitter[*Socket], topic string, socket *Socket) {
+	em.Subscribe(socket, topic, func(subscriber *Socket, msg any) {
+		_ = msg.(*gws.Broadcaster).Broadcast(subscriber.Conn)
+	})
+}
+
+func Pub(em *event_emitter.EventEmitter[*Socket], topic string, op gws.Opcode, msg []byte) {
+	var broadcaster = gws.NewBroadcaster(op, msg)
+	defer broadcaster.Close()
+	em.Publish(topic, broadcaster)
+}
+```
+
 ### Autobahn 测试
 
 ```bash

benchmark_test.go

@@ -6,11 +6,12 @@ import (
 	"compress/flate"
 	_ "embed"
 	"encoding/binary"
-	klauspost "github.com/klauspost/compress/flate"
-	"github.com/lxzan/gws/internal"
 	"io"
 	"net"
 	"testing"
+
+	klauspost "github.com/klauspost/compress/flate"
+	"github.com/lxzan/gws/internal"
 )
 
 //go:embed assets/github.json

client_test.go

@@ -3,13 +3,14 @@ package gws
 import (
 	"crypto/tls"
 	"errors"
-	"github.com/lxzan/gws/internal"
-	"github.com/stretchr/testify/assert"
 	"io"
 	"net"
 	"net/http"
 	"testing"
 	"time"
+
+	"github.com/lxzan/gws/internal"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestNewClient(t *testing.T) {

compress.go

@@ -3,12 +3,13 @@ package gws
 import (
 	"bytes"
 	"encoding/binary"
-	"github.com/klauspost/compress/flate"
-	"github.com/lxzan/gws/internal"
 	"io"
 	"math"
 	"sync"
 	"sync/atomic"
+
+	"github.com/klauspost/compress/flate"
+	"github.com/lxzan/gws/internal"
 )
 
 // FlateTail Add four bytes as specified in RFC

conn.go

@@ -91,24 +91,25 @@ func (c *Conn) emitError(err error) {
 		return
 	}
 
-	var responseCode = internal.CloseNormalClosure
-	var responseErr error = internal.CloseNormalClosure
-	switch v := err.(type) {
-	case internal.StatusCode:
-		responseCode = v
-	case *internal.Error:
-		responseCode = v.Code
-		responseErr = v.Err
-	default:
-		responseErr = err
-	}
-
-	var content = responseCode.Bytes()
-	content = append(content, err.Error()...)
-	if len(content) > internal.ThresholdV1 {
-		content = content[:internal.ThresholdV1]
-	}
 	if atomic.CompareAndSwapUint32(&c.closed, 0, 1) {
+		var responseCode = internal.CloseNormalClosure
+		var responseErr error = internal.CloseNormalClosure
+		switch v := err.(type) {
+		case internal.StatusCode:
+			responseCode = v
+		case *internal.Error:
+			responseCode = v.Code
+			responseErr = v.Err
+		default:
+			responseErr = err
+		}
+
+		var content = responseCode.Bytes()
+		content = append(content, err.Error()...)
+		if len(content) > internal.ThresholdV1 {
+			content = content[:internal.ThresholdV1]
+		}
+
 		c.close(content, responseErr)
 	}
 }

examples/chatroom/main.go

@@ -3,10 +3,11 @@ package main
 import (
 	_ "embed"
 	"encoding/json"
-	"github.com/lxzan/gws"
 	"log"
 	"net/http"
 	"time"
+
+	"github.com/lxzan/gws"
 )
 
 const (

examples/echo/main.go

@@ -1,9 +1,10 @@
 package main
 
 import (
-	"github.com/lxzan/gws"
 	"log"
 	"net/http"
+
+	"github.com/lxzan/gws"
 )
 
 func main() {

examples/push/main.go

@@ -35,13 +35,13 @@ type Handler struct {
 	gws.BuiltinEventHandler
 }
 
-func (c *Handler) MustGet(ss gws.SessionStorage, key string) any {
-	v, _ := ss.Load(key)
+func (c *Handler) getSession(socket *gws.Conn, key string) any {
+	v, _ := socket.Session().Load(key)
 	return v
 }
 
 func (c *Handler) Send(socket *gws.Conn, payload []byte) {
-	var channel = c.MustGet(socket.Session(), "channel").(chan []byte)
+	var channel = c.getSession(socket, "channel").(chan []byte)
 	select {
 	case channel <- payload:
 	default:
@@ -50,7 +50,7 @@ func (c *Handler) Send(socket *gws.Conn, payload []byte) {
 }
 
 func (c *Handler) OnClose(socket *gws.Conn, err error) {
-	var closer = c.MustGet(socket.Session(), "closer").(chan struct{})
+	var closer = c.getSession(socket, "closer").(chan struct{})
 	closer <- struct{}{}
 }
 

init.go

@@ -3,7 +3,7 @@ package gws
 import "github.com/lxzan/gws/internal"
 
 var (
-	myPadding     = frameHeader{}            // 帧头填充物
+	framePadding  = frameHeader{}            // 帧头填充物
 	binaryPool    = internal.NewBufferPool() // 缓冲池
 	defaultLogger = new(stdLogger)           // 默认日志工具
 )

internal/pool_test.go

@@ -2,8 +2,9 @@ package internal
 
 import (
 	"bytes"
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestBufferPool(t *testing.T) {

internal/utils_test.go

@@ -4,14 +4,15 @@ import (
 	"bytes"
 	"encoding/binary"
 	"encoding/hex"
-	"github.com/stretchr/testify/assert"
 	"hash/fnv"
 	"io"
 	"net/http"
 	"reflect"
 	"strings"
 	"testing"
 	"unsafe"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestStringToBytes(t *testing.T) {

option.go

@@ -2,12 +2,13 @@ package gws
 
 import (
 	"bufio"
-	"compress/flate"
 	"crypto/tls"
-	"github.com/lxzan/gws/internal"
 	"net"
 	"net/http"
 	"time"
+
+	"github.com/klauspost/compress/flate"
+	"github.com/lxzan/gws/internal"
 )
 
 const (

reader.go

@@ -3,8 +3,9 @@ package gws
 import (
 	"bytes"
 	"fmt"
-	"github.com/lxzan/gws/internal"
 	"unsafe"
+
+	"github.com/lxzan/gws/internal"
 )
 
 func (c *Conn) checkMask(enabled bool) error {