feat: 新版 server 包 HTTP 基础实现

go.mod

@@ -36,6 +36,9 @@ require (
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.14.0 // indirect
+	github.com/gobwas/httphead v0.1.0 // indirect
+	github.com/gobwas/pool v0.2.1 // indirect
+	github.com/gobwas/ws v1.3.2 // indirect
 	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/gopherjs/gopherjs v1.17.2 // indirect

go.sum

@@ -45,6 +45,12 @@ github.com/go-playground/validator/v10 v10.14.0 h1:vgvQWe3XCz3gIeFDm/HnTIbj6UGmg
 github.com/go-playground/validator/v10 v10.14.0/go.mod h1:9iXMNT7sEkjXb0I+enO7QXmzG6QCsPWY4zveKFVRSyU=
 github.com/go-resty/resty/v2 v2.11.0 h1:i7jMfNOJYMp69lq7qozJP+bjgzfAzeOhuGlyDrqxT/8=
 github.com/go-resty/resty/v2 v2.11.0/go.mod h1:iiP/OpA0CkcL3IGt1O0+/SIItFUbkkyw5BGXiVdTu+A=
+github.com/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU=
+github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM=
+github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og=
+github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw=
+github.com/gobwas/ws v1.3.2 h1:zlnbNHxumkRvfPWgfXu8RBwyNR1x8wh9cf5PTOCqs9Q=
+github.com/gobwas/ws v1.3.2/go.mod h1:hRKAFb8wOxFROYNsT1bqfWnhX+b5MFeJM9r2ZSwg/KY=
 github.com/goccy/go-json v0.9.7/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
 github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
 github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=

server/v2/conn.go

@@ -0,0 +1,59 @@
+package server
+
+import (
+	"context"
+	"github.com/kercylan98/minotaur/utils/log"
+	"net"
+	"unsafe"
+)
+
+type Conn interface {
+	net.Conn
+}
+
+type conn struct {
+	net.Conn
+	cs     *connections
+	ctx    context.Context
+	cancel context.CancelFunc
+	idx    int
+}
+
+func (c *conn) init(ctx context.Context, cs *connections, conn net.Conn, idx int) *conn {
+	c.Conn = conn
+	c.cs = cs
+	c.ctx, c.cancel = context.WithCancel(ctx)
+	c.idx = idx
+	return c
+}
+
+func (c *conn) awaitRead() {
+	defer func() { _ = c.Close() }()
+
+	const bufferSize = 4096
+	buf := make([]byte, bufferSize) // 避免频繁的内存分配，初始化一个固定大小的缓冲区
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		default:
+			ptr := unsafe.Pointer(&buf[0])
+			n, err := c.Read((*[bufferSize]byte)(ptr)[:])
+			if err != nil {
+				log.Error("READ", err)
+				return
+			}
+
+			if n > 0 {
+				if _, err := c.Write(buf[:n]); err != nil {
+					log.Error("Write", err)
+				}
+			}
+		}
+	}
+}
+
+func (c *conn) Close() (err error) {
+	c.cs.Event() <- c
+	return
+}

server/v2/connections.go

@@ -0,0 +1,111 @@
+package server
+
+import (
+	"context"
+	"github.com/kercylan98/minotaur/utils/log"
+	"net"
+	"time"
+)
+
+// connections 结构体用于管理连接
+type connections struct {
+	ctx   context.Context // 上下文对象，用于取消连接管理器
+	ch    chan any        // 事件通道，用于接收连接管理器的操作事件
+	items []*conn         // 连接列表，存储所有打开的连接
+	gap   []int           // 连接空隙，记录已关闭的连接索引，用于重用索引
+}
+
+// 初始化连接管理器
+func (cs *connections) init(ctx context.Context) *connections {
+	cs.ctx = ctx
+	cs.ch = make(chan any, 1024)
+	cs.items = make([]*conn, 0, 128)
+	go cs.awaitRun()
+	return cs
+}
+
+// 清理连接列表中的空隙
+func (cs *connections) clearGap() {
+	cs.gap = cs.gap[:0]
+	var gap = make([]int, 0, len(cs.items))
+	for i, c := range cs.items {
+		if c == nil {
+			continue
+		}
+		c.idx = i
+		gap = append(gap, i)
+	}
+
+	cs.gap = gap
+}
+
+// 打开新连接
+func (cs *connections) open(c net.Conn) error {
+	// 如果存在连接空隙，则重用连接空隙中的索引，否则分配新的索引
+	var idx int
+	var reuse bool
+	if len(cs.gap) > 0 {
+		idx = cs.gap[0]
+		cs.gap = cs.gap[1:]
+		reuse = true
+	} else {
+		idx = len(cs.items)
+	}
+	conn := new(conn).init(cs.ctx, cs, c, idx)
+	if reuse {
+		cs.items[idx] = conn
+	} else {
+		cs.items = append(cs.items, conn)
+	}
+	go conn.awaitRead()
+	return nil
+}
+
+// 关闭连接
+func (cs *connections) close(c *conn) error {
+	if c == nil {
+		return nil
+	}
+	defer c.cancel()
+	// 如果连接索引是连接列表的最后一个索引，则直接删除连接对象，否则将连接对象置空，并将索引添加到连接空隙中
+	if c.idx == len(cs.items)-1 {
+		cs.items = cs.items[:c.idx]
+	} else {
+		cs.items[c.idx] = nil
+		cs.gap = append(cs.gap, c.idx)
+	}
+	return c.Conn.Close()
+}
+
+// 等待连接管理器的事件并处理
+func (cs *connections) awaitRun() {
+	clearGapTicker := time.NewTicker(time.Second * 30)
+	defer clearGapTicker.Stop()
+
+	for {
+		select {
+		case <-cs.ctx.Done():
+			return
+		case <-clearGapTicker.C:
+			cs.clearGap()
+		case a := <-cs.ch:
+			var err error
+
+			switch v := a.(type) {
+			case *conn:
+				err = cs.close(v)
+			case net.Conn:
+				err = cs.open(v)
+			}
+
+			if err != nil {
+				log.Error("connections.awaitRun", log.Any("err", err))
+			}
+		}
+	}
+}
+
+// Event 获取连接管理器的事件通道
+func (cs *connections) Event() chan<- any {
+	return cs.ch
+}

server/v2/core.go

@@ -0,0 +1,9 @@
+package server
+
+type Core interface {
+	connectionManager
+}
+
+type connectionManager interface {
+	Event() chan<- any
+}

server/v2/network.go

@@ -0,0 +1,11 @@
+package server
+
+import "context"
+
+type Network interface {
+	OnSetup(ctx context.Context, core Core) error
+
+	OnRun(ctx context.Context) error
+
+	OnShutdown() error
+}

server/v2/network/http.go

@@ -0,0 +1,53 @@
+package network
+
+import (
+	"context"
+	"github.com/kercylan98/minotaur/server/v2"
+	"github.com/pkg/errors"
+	"net"
+	"net/http"
+	"time"
+)
+
+func Http(addr string) server.Network {
+	return HttpWithHandler(addr, &HttpServe{ServeMux: http.NewServeMux()})
+}
+
+func HttpWithHandler[H http.Handler](addr string, handler H) server.Network {
+	c := &httpCore[H]{
+		addr:    addr,
+		handler: handler,
+		srv: &http.Server{
+			Addr:                         addr,
+			Handler:                      handler,
+			DisableGeneralOptionsHandler: false,
+		},
+	}
+	return c
+}
+
+type httpCore[H http.Handler] struct {
+	addr    string
+	handler H
+	srv     *http.Server
+}
+
+func (h *httpCore[H]) OnSetup(ctx context.Context, core server.Core) (err error) {
+	h.srv.BaseContext = func(listener net.Listener) context.Context {
+		return ctx
+	}
+	return
+}
+
+func (h *httpCore[H]) OnRun(ctx context.Context) (err error) {
+	if err = h.srv.ListenAndServe(); errors.Is(err, http.ErrServerClosed) {
+		err = nil
+	}
+	return
+}
+
+func (h *httpCore[H]) OnShutdown() error {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	return h.srv.Shutdown(ctx)
+}

server/v2/network/http_serve.go

@@ -0,0 +1,7 @@
+package network
+
+import "net/http"
+
+type HttpServe struct {
+	*http.ServeMux
+}