feat: 新版 server 包 HTTP 基础实现

server/v2/event_handler.go

@@ -0,0 +1,69 @@
+package server
+
+import (
+	"fmt"
+	"github.com/panjf2000/ants/v2"
+	"github.com/panjf2000/gnet/v2"
+	"time"
+)
+
+func newEventHandler(trafficker Trafficker) (handler *eventHandler, err error) {
+	var wp *ants.Pool
+	if wp, err = ants.NewPool(ants.DefaultAntsPoolSize, ants.WithNonblocking(true)); err != nil {
+		return
+	}
+	handler = &eventHandler{
+		trafficker: trafficker,
+		workerPool: wp,
+	}
+	return
+}
+
+type (
+	Trafficker interface {
+		OnBoot() error
+		OnTraffic(c gnet.Conn, packet []byte)
+	}
+	eventHandler struct {
+		trafficker Trafficker
+		workerPool *ants.Pool
+	}
+)
+
+func (e *eventHandler) OnBoot(eng gnet.Engine) (action gnet.Action) {
+	if err := e.trafficker.OnBoot(); err != nil {
+		action = gnet.Shutdown
+	}
+	return
+}
+
+func (e *eventHandler) OnShutdown(eng gnet.Engine) {
+	return
+}
+
+func (e *eventHandler) OnOpen(c gnet.Conn) (out []byte, action gnet.Action) {
+	return
+}
+
+func (e *eventHandler) OnClose(c gnet.Conn, err error) (action gnet.Action) {
+	fmt.Println("断开")
+	return
+}
+
+func (e *eventHandler) OnTraffic(c gnet.Conn) (action gnet.Action) {
+	buf, err := c.Next(-1)
+	if err != nil {
+		return
+	}
+	var packet = make([]byte, len(buf))
+	copy(packet, buf)
+	err = e.workerPool.Submit(func() {
+		e.trafficker.OnTraffic(c, packet)
+	})
+
+	return
+}
+
+func (e *eventHandler) OnTick() (delay time.Duration, action gnet.Action) {
+	return
+}

server/v2/server.go

@@ -0,0 +1,20 @@
+package server
+
+import "github.com/panjf2000/gnet/v2"
+
+func NewServer(trafficker Trafficker) *Server {
+	srv := &Server{
+		trafficker: trafficker,
+	}
+	return srv
+}
+
+type Server struct {
+	trafficker Trafficker
+}
+
+func (s *Server) Run(protoAddr string) (err error) {
+	var handler *eventHandler
+	handler, err = newEventHandler(s.trafficker)
+	return gnet.Run(handler, protoAddr)
+}

server/v2/traffickers/http.go

@@ -0,0 +1,42 @@
+package traffickers
+
+import (
+	"bufio"
+	"bytes"
+	"github.com/kercylan98/minotaur/server/v2"
+	"github.com/kercylan98/minotaur/utils/hub"
+	"github.com/panjf2000/gnet/v2"
+	netHttp "net/http"
+)
+
+func Http(handler netHttp.Handler) server.Trafficker {
+	return &http{
+		handler: handler,
+	}
+}
+
+type http struct {
+	handler netHttp.Handler
+	rwp     *hub.ObjectPool[*httpResponseWriter]
+}
+
+func (h *http) OnBoot() error {
+	h.rwp = hub.NewObjectPool[httpResponseWriter](func() *httpResponseWriter {
+		return new(httpResponseWriter)
+	}, func(data *httpResponseWriter) {
+		data.reset()
+	})
+	return nil
+}
+
+func (h *http) OnTraffic(c gnet.Conn, packet []byte) {
+	httpRequest, err := netHttp.ReadRequest(bufio.NewReader(bytes.NewReader(packet)))
+	if err != nil {
+		return
+	}
+
+	responseWriter := h.rwp.Get()
+	responseWriter.init(c)
+
+	h.handler.ServeHTTP(responseWriter, httpRequest)
+}

server/v2/traffickers/http_response_writer.go

@@ -0,0 +1,58 @@
+package traffickers
+
+import (
+	"bytes"
+	"github.com/panjf2000/gnet/v2"
+	netHttp "net/http"
+	"strconv"
+	"sync"
+)
+
+type httpResponseWriter struct {
+	c          gnet.Conn
+	statusCode int
+	header     netHttp.Header
+}
+
+func (w *httpResponseWriter) init(c gnet.Conn) {
+	w.c = c
+	w.statusCode = 200
+	w.header = make(netHttp.Header)
+}
+
+func (w *httpResponseWriter) reset() {
+	w.c = nil
+	w.statusCode = 200
+	w.header = nil
+}
+
+func (w *httpResponseWriter) Header() netHttp.Header {
+	return w.header
+}
+
+func (w *httpResponseWriter) Write(b []byte) (n int, err error) {
+	var buf bytes.Buffer
+	buf.WriteString("HTTP/1.1 ")
+	buf.WriteString(netHttp.StatusText(w.statusCode))
+	buf.WriteString("\r\n")
+	w.header.Set("Content-Length", strconv.Itoa(len(b)))
+	if err = w.header.Write(&buf); err != nil {
+		return
+	}
+	buf.WriteString("\r\n")
+	buf.Write(b)
+	res := buf.Bytes()
+	var wg sync.WaitGroup
+	wg.Add(1)
+	err = w.c.AsyncWrite(res, func(c gnet.Conn, e error) error {
+		err = e
+		wg.Done()
+		return nil
+	})
+	wg.Wait()
+	return len(res), err
+}
+
+func (w *httpResponseWriter) WriteHeader(statusCode int) {
+	w.statusCode = statusCode
+}