k2safe/go-omni
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add bounded native http server

Browse Source
This commit is contained in:
k2safe
2026-09-21 01:58:50 +08:00
parent 3bbf4627a1
commit 05b4088a25
14 changed files with 1058 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
CHANGELOG.md
View File

@@ -1,5 +1,12 @@
# Changelog
## v0.5.0
- 新增 `transport/native` 第一版自研 HTTP/1.1 服务端，接管 TCP 监听、连接、Keep-Alive、请求解析、响应写回和优雅关闭。
- 新增 `App.RunWith` / `omni.Runner` 启动边界；native 通过标准 `http.Handler` 接入，现有 Handler、Service 和 Repository 无需修改。
- native 支持 `Content-Length` 请求体、请求头 / 请求体限制、读写超时和服务关闭时的请求取消；通过对象池和 `MaxConnections` / `MaxConcurrentRequests` 控制高并发内存；chunked 请求体暂返回 501。
- `omni.New()` 默认仍使用只依赖标准库的 `transport/nethttp` 路由，native 作为显式启用的网络层预览实现。
## v0.4.0
- **破坏性变更**：移除 Gin 适配器及其全部依赖，`omni.New()` 默认使用只依赖标准库的 `transport/nethttp`。

37
README.md
View File

@@ -5,7 +5,7 @@ GoOmni 是一套面向业务复用的 Go 后端基础框架。当前默认使用
远程 module 地址是 `git.yuntuo.hk/k2safe/go-omni`。业务项目可以直接引用指定版本：
```sh
go get git.yuntuo.hk/k2safe/go-omni@v0.4.0
go get git.yuntuo.hk/k2safe/go-omni@v0.5.0
```
`v0.4.0` 移除了 Gin 和 Gin 专属的 `WithMode` 配置；`omni.New()` 直接使用标准库路由，已有业务 Handler / Service 无需修改。
@@ -13,7 +13,7 @@ go get git.yuntuo.hk/k2safe/go-omni@v0.4.0
创建新业务项目：
```sh
go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.4.0
go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.5.0
go-omni new --module example.com/billing ./billing
cd billing
go mod tidy
@@ -43,6 +43,7 @@ go-omni/
├── auth/ # 认证和权限接口
├── database/ # database/sql 生命周期和事务
├── transport/nethttp/ # 标准库网络适配器
├── transport/native/ # 自研 HTTP/1.1 服务端预览
├── examples/ # 示例项目
└── cmd/ # go-omni 项目脚手架
```
@@ -94,6 +95,34 @@ application := omni.NewWithRouter(myRouter)
`app.Engine()` 返回标准 `http.Handler`，`ctx.Raw()` 返回适配器提供的 `http.ResponseWriter`，业务 API 不锁定具体网络实现。
需要试用第一版自研网络层时，业务注册代码无需改变，只替换启动器：
```go
import (
"context"
"log"
"os"
"os/signal"
"syscall"
omni "git.yuntuo.hk/k2safe/go-omni"
"git.yuntuo.hk/k2safe/go-omni/transport/native"
)
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer cancel()
app := omni.New()
app.GET("/health", func(ctx *omni.Context) error {
return ctx.OK(map[string]string{"status": "ok"})
})
if err := app.RunWith(ctx, native.New(nil), ":8080"); err != nil {
log.Fatal(err)
}
```
`transport/native` 当前支持 HTTP/1.1、Keep-Alive、`Content-Length` 请求体、读写超时和优雅关闭；chunked 请求体暂返回 501。它会复用连接状态和响应对象，并可通过 `MaxConnections`、`MaxConcurrentRequests` 限制内存峰值。它仍通过标准 HTTP 请求/响应接口连接业务层，适合在现有 Handler / Service 上做网络层验证。
业务代码可以直接使用 `git.yuntuo.hk/k2safe/go-omni/logx` 打结构化日志，不需要依赖具体网络库：
```go
@@ -171,10 +200,10 @@ userService := service.NewUsers(services)
从远程版本生成并升级一个独立业务模块，可运行：
```sh
python3 scripts/verifyBusiness.py --version v0.4.0
python3 scripts/verifyBusiness.py --version v0.5.0
```
该验收会在临时目录中先固定 `v0.2.0`，再升级到 `v0.4.0`，检查无本地 replace、配置加载、结构化日志、requestId、CRUD、校验错误、业务错误和优雅退出。
该验收会在临时目录中先固定 `v0.2.0`，再升级到 `v0.5.0`，检查无本地 replace、配置加载、结构化日志、requestId、CRUD、校验错误、业务错误和优雅退出。
标准库路由和完整 GoOmni 请求链的可重复性能基线见 [PERFORMANCE.md](PERFORMANCE.md)，运行 `python3 scripts/verifyPerformance.py` 可重新采样。

8
ROADMAP.md
View File

@@ -35,10 +35,12 @@
- 业务试用没有要求修改 Handler / Service；现有公共 API 足以支撑一套分层业务模块。
- 自动验收入口：`python3 scripts/verifyBusiness.py --version v0.4.0`。
## 5. 稳定性与性能基线（进行中）
## 5. 自研网络层、稳定性与性能基线（进行中，v0.5.0 预览）
- 已加入标准库路由和完整 GoOmni 请求链 benchmark，输出分配次数和耗时，作为未来自研网络层的对照基线。
- 已具备全量 race、vet、build 验收；生产压测、长连接和真实部署规模验证仍待完成。
- 新增 `transport/native` 第一版自研 HTTP/1.1 服务端：自行管理 TCP listener、Keep-Alive、Content-Length 请求解析、响应写回、读写超时和优雅关闭；通过 `App.RunWith` 接入，业务 Handler / Service 保持不变。
- native 当前拒绝 chunked 请求体，仍使用标准 `http.Request` / `http.ResponseWriter` 作为桥接；解析器、连接抽象和生产压测将在后续阶段继续下沉。
- 已具备全量 race、vet、build 验收；生产压测、长连接规模和真实部署验证仍待完成。
- 自动验收入口：`python3 scripts/verifyPerformance.py`。
以上阶段以验收结果判断完成。当前仍是基础框架试用阶段；MySQL、适配层契约和一次业务模块试用已完成，稳定性与性能验证、自研网络内核尚未完成。
以上阶段以验收结果判断完成。当前仍是基础框架试用阶段；MySQL、适配层契约和一次业务模块试用已完成，自研网络层已有可运行预览，稳定性与生产规模验证尚未完成。

26
app/app.go
View File

@@ -29,6 +29,13 @@ type App struct {
closeErr error
}
// Runner is the small lifecycle boundary used by an HTTP server
// implementation. The default App.RunContext uses net/http; a custom runner
// can own the listener, protocol parser and connection lifecycle.
type Runner interface {
Run(context.Context, http.Handler, string) error
}
// New creates an application using GoOmni's standard-library router.
func New(options ...config.Option) *App {
settings := configure(options...)
@@ -150,6 +157,25 @@ func (a *App) RunContext(ctx context.Context, address ...string) error {
return a.serveContext(ctx, server, listener)
}
// RunWith starts a custom server runner while keeping App-owned cleanup in the
// same lifecycle. The runner receives the transport-neutral App handler.
func (a *App) RunWith(ctx context.Context, runner Runner, address ...string) error {
if runner == nil {
return errors.Join(errors.New("go-omni: nil runner"), a.Close())
}
if ctx == nil {
ctx = context.Background()
}
listenOn := a.config.Address
if listenOn == "" {
listenOn = ":8080"
}
if len(address) > 0 && address[0] != "" {
listenOn = address[0]
}
return errors.Join(runner.Run(ctx, a, listenOn), a.Close())
}
func (a *App) serveContext(ctx context.Context, server *http.Server, listener net.Listener) error {
serveErr := make(chan error, 1)
go func() { serveErr <- server.Serve(listener) }()

8
cmd/go-omni/README.md
View File

@@ -3,7 +3,7 @@
从远程安装，生成一个带 Handler / Service / Repository / Model 分层的服务骨架：
```sh
go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.4.0
go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.5.0
go-omni new --module example.com/billing ./billing
cd billing
go mod tidy
@@ -16,7 +16,7 @@ go run ./cmd/server
需要选择其他已发布版本时，使用完整版本号：
```sh
go-omni new --framework-version v0.4.0 --module example.com/billing ./billing
go-omni new --framework-version v0.5.0 --module example.com/billing ./billing
```
`--framework-version` 禁用自动本地替换；`main`、`latest` 等浮动引用不接受为固定版本。
@@ -48,8 +48,8 @@ MySQL 项目额外包含 `schema.sql`、MySQL Repository 和可选的数据库
发布验收使用 Python 3 标准库脚本，不需要额外 Python 依赖：
```sh
python3 scripts/verifyScaffold.py --version v0.4.0
python3 scripts/verifyDatabase.py --version v0.4.0
python3 scripts/verifyScaffold.py --version v0.5.0
python3 scripts/verifyDatabase.py --version v0.5.0
```
脚本在源码目录外安装远程 CLI、生成项目，确认固定版本且没有 `replace`，再下载依赖、

2
cmd/go-omni/scaffold/version.go
View File

@@ -10,7 +10,7 @@ const frameworkModule = "git.yuntuo.hk/k2safe/go-omni"
// Source builds can carry an unpublished VCS version (including +dirty).
// Keep this fallback in sync with the next release; installed CLIs use their
// downloaded module version, which also has a module checksum.
const defaultFrameworkVersion = "v0.4.0"
const defaultFrameworkVersion = "v0.5.0"
var frameworkVersionPattern = regexp.MustCompile(`^v[01]\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)(-((0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*)(\.(0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*))*))?(\+[0-9A-Za-z-]+(\.[0-9A-Za-z-]+)*)?$`)

1
omni.go
View File

@@ -13,6 +13,7 @@ import (
)
type App = app.App
type Runner = app.Runner
type RouterGroup = app.RouterGroup
type Principal = auth.Principal
type Authenticator = auth.Authenticator

2
scripts/verifyBusiness.py
View File

@@ -25,7 +25,7 @@ def free_address():
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--go", default="go")
parser.add_argument("--version", default="v0.4.0", help="Target framework release")
parser.add_argument("--version", default="v0.5.0", help="Target framework release")
parser.add_argument("--from-version", default="v0.2.0", help="Version used before upgrade")
args = parser.parse_args()

2
scripts/verifyDatabase.py
View File

@@ -145,7 +145,7 @@ class Server:
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--go", default="go")
parser.add_argument("--version", default="v0.4.0")
parser.add_argument("--version", default="v0.5.0")
parser.add_argument("--cli", help="Use a locally built CLI")
parser.add_argument("--framework-dir", help="Local framework source for pre-release checks")
parser.add_argument("--image", default="mysql:8.4")

2
scripts/verifyScaffold.py
View File

@@ -91,7 +91,7 @@ def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None):
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--go", default="go", help="Go executable")
parser.add_argument("--version", default="v0.4.0", help="Released CLI version to install")
parser.add_argument("--version", default="v0.5.0", help="Released CLI version to install")
parser.add_argument("--cli", help="Use a locally built CLI instead of installing")
parser.add_argument("--framework-version", help="Explicit version passed to the CLI")
parser.add_argument("--framework-dir", help="Local checkout for pre-release verification")

4
scripts/verifyTransports.py
View File

@@ -13,7 +13,7 @@ from verifyScaffold import MODULE, run
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--go", default="go")
parser.add_argument("--version", default="v0.4.0")
parser.add_argument("--version", default="v0.5.0")
parser.add_argument("--framework-dir", help="Use a local checkout before release")
args = parser.parse_args()
env = dict(os.environ, GOWORK="off")
@@ -55,7 +55,7 @@ func TestContract(t *testing.T) {{
count = sum(event.get("Action") == "pass" and event.get("Test", "").startswith(name + "/")
for event in events)
print(f"PASS {adapter}: {count} contract checks, including business handlers, cancellation and concurrent requests", flush=True)
print("PASS independent module: both adapters accepted by the same public contract suite", flush=True)
print("PASS independent module: nethttp accepted by the public contract suite", flush=True)
if __name__ == "__main__":

22
transport/README.md
View File

@@ -14,6 +14,24 @@ registerRoutes(application) // 与 omni.New() 复用同一份业务注册代码
`transport/nethttp` 是当前内置的标准库路由实现。它提供一套可移植的路径、方法和错误回退规则；自研网络层可以直接实现同一个 `httpx.Router` 契约。
`transport/native` 是第一版自研 HTTP/1.1 服务端：它自己管理 TCP listener、连接、Keep-Alive、请求解析、响应写回和优雅关闭，再把请求交给现有 `http.Handler`。当前只支持 `Content-Length` 请求体，chunked 请求会返回 501；它仍使用标准 `http.Request` / `http.ResponseWriter` 作为业务桥接，因此不会影响现有 Handler / Service。
服务端不是把连接交给一个全局串行 worker，而是每条连接独立处理请求；Keep-Alive 负责复用客户端连接，`Server` 内部的 `sync.Pool` 负责复用小型 Reader、响应对象和连接状态。Reader 默认只有 4 KiB，需要处理大请求头时按需增长。需要保护内存上限时，可以设置 `MaxConnections` 和 `MaxConcurrentRequests`；超出连接上限的连接会被关闭，超出请求并发上限的请求返回 503，不会无限排队。
这个设计借鉴了 Nginx 的资源边界，但没有照搬它的进程模型：Nginx 用少量 worker 进程和事件机制管理大量连接，并用 `worker_connections` 限制每个 worker 的同时连接数；Go runtime 自己已经用 epoll / kqueue 等网络轮询器唤醒阻塞的 goroutine，因此 native 采用“accept loop + 每连接轻量 goroutine + 请求并发闸门”的等价实现。这样业务 Handler 可以继续执行同步代码，慢请求不会阻塞其他连接；只有在真实压测证明 goroutine 调度或协议状态机成为瓶颈时，才需要继续下沉为手写事件循环。
使用自研服务端：
```go
application := omni.New()
server := native.New(nil)
if err := application.RunWith(ctx, server, ":8080"); err != nil {
log.Fatal(err)
}
```
`omni.New()` 默认仍使用 `transport/nethttp` 路由；native 服务端是显式启用的预览实现，等协议和连接边界稳定后再考虑切换默认启动器。
## 可移植路由规则（v0.3.0）
| 项目 | 约定 |
@@ -67,9 +85,9 @@ func TestContract(t *testing.T) {
```sh
go test -race ./transport/... ./examples/reference
python3 scripts/verifyTransports.py --version v0.4.0
python3 scripts/verifyTransports.py --version v0.5.0
```
第一条验证标准库适配器复用完整 Handler / Service / Repository。第二条在源码目录外创建无 replace 的消费模块，从远程版本导入同一套验收包；自研适配器也应使用同一个公共测试包。
当前替换边界是路由与 HTTP 适配层，`App.Run` 仍使用标准库 `net/http.Server`。未来自研监听、连接管理或协议解析时，可让自研服务调用 `App.ServeHTTP`（提供标准 HTTP 请求/响应桥接），由自研启动层管理监听与关闭，并在退出时调用 `App.Close`；本阶段没有实现新的网络内核。
当前替换边界分成两层：`transport/nethttp` 提供标准库路由，`transport/native` 进一步接管 TCP 监听、连接管理和 HTTP/1.1 协议 I/O。`App.Run` / `App.RunContext` 仍使用标准库 `net/http.Server`，需要显式调用 `App.RunWith(ctx, native.New(nil), address)` 才会启用 native 预览实现。native 当前通过标准 `http.Request` / `http.ResponseWriter` 桥接到业务层，后续可在不改 Handler / Service 的前提下逐步替换解析器和响应写回。

655
transport/native/server.go Normal file
View File

@@ -0,0 +1,655 @@
// Package native provides GoOmni's first project-owned HTTP/1.1 server.
//
// It deliberately keeps the public handler boundary as net/http so existing
// GoOmni business code remains unchanged while connection management and
// protocol I/O move out of net/http.Server. The parser currently supports
// Content-Length request bodies and rejects chunked requests.
package native
import (
"bufio"
"bytes"
"context"
"errors"
"io"
"net"
"net/http"
"net/url"
"sort"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
)
const (
defaultReadHeaderTimeout = 5 * time.Second
defaultReadTimeout = 30 * time.Second
defaultWriteTimeout = 30 * time.Second
defaultIdleTimeout = 60 * time.Second
defaultMaxHeaderBytes = 1 << 20
defaultMaxBodyBytes = 8 << 20
defaultReadBufferSize = 4 << 10
maxPooledResponseBytes = 64 << 10
)
// Server is a small HTTP/1.1 server with project-owned connection lifecycle.
// It supports keep-alive, graceful shutdown, bounded concurrency and standard
// http.Handler values. Zero MaxConnections and MaxConcurrentRequests mean
// unlimited; pooled protocol objects are reused automatically.
type Server struct {
ReadHeaderTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
IdleTimeout time.Duration
ReadBufferSize int
MaxHeaderBytes int
MaxBodyBytes int64
MaxConnections int
MaxConcurrentRequests int
mu sync.Mutex
handler http.Handler
listener net.Listener
conns map[*connection]struct{}
closing bool
cancel context.CancelFunc
wg sync.WaitGroup
connSlots chan struct{}
requestSlots chan struct{}
readerPool sync.Pool
writerPool sync.Pool
connectionPool sync.Pool
}
type connection struct {
server *Server
conn net.Conn
active atomic.Bool
closeWhenIdle atomic.Bool
closed atomic.Bool
slotHeld bool
closeDone chan struct{}
}
// New creates a native server. The handler may also be supplied later through
// App.RunWith, which is useful when the App owns the lifecycle.
func New(handler http.Handler) *Server {
server := &Server{
handler: handler,
ReadHeaderTimeout: defaultReadHeaderTimeout,
ReadTimeout: defaultReadTimeout,
WriteTimeout: defaultWriteTimeout,
IdleTimeout: defaultIdleTimeout,
ReadBufferSize: defaultReadBufferSize,
MaxHeaderBytes: defaultMaxHeaderBytes,
MaxBodyBytes: defaultMaxBodyBytes,
conns: make(map[*connection]struct{}),
}
server.readerPool.New = func() any { return bufio.NewReaderSize(nil, server.readBufferSize()) }
server.writerPool.New = func() any { return newResponseWriter("") }
server.connectionPool.New = func() any { return &connection{server: server} }
return server
}
func (s *Server) readBufferSize() int {
if s.ReadBufferSize > 0 {
return s.ReadBufferSize
}
return defaultReadBufferSize
}
// Run implements app.Runner without importing the app package.
func (s *Server) Run(ctx context.Context, handler http.Handler, address string) error {
if handler != nil {
s.handler = handler
}
if s.handler == nil {
return errors.New("go-omni/native: nil handler")
}
if address == "" {
address = ":8080"
}
listener, err := net.Listen("tcp", address)
if err != nil {
return err
}
return s.Serve(ctx, listener)
}
// Serve accepts connections from listener until the context is canceled or an
// unrecoverable accept error occurs. The listener is owned by the Server.
func (s *Server) Serve(ctx context.Context, listener net.Listener) error {
if listener == nil {
return errors.New("go-omni/native: nil listener")
}
if s.handler == nil {
return errors.New("go-omni/native: nil handler")
}
if ctx == nil {
ctx = context.Background()
}
s.mu.Lock()
if s.listener != nil {
s.mu.Unlock()
return errors.New("go-omni/native: server is already running")
}
s.listener = listener
s.closing = false
if s.MaxConnections > 0 {
s.connSlots = make(chan struct{}, s.MaxConnections)
} else {
s.connSlots = nil
}
if s.MaxConcurrentRequests > 0 {
s.requestSlots = make(chan struct{}, s.MaxConcurrentRequests)
} else {
s.requestSlots = nil
}
baseCtx, cancel := context.WithCancel(context.Background())
s.cancel = cancel
s.mu.Unlock()
defer cancel()
shutdownDone := make(chan struct{})
go func() {
select {
case <-ctx.Done():
timeout := s.shutdownTimeout()
shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout)
_ = s.Shutdown(shutdownCtx)
cancel()
case <-shutdownDone:
}
}()
var serveErr error
for {
conn, err := listener.Accept()
if err != nil {
if ctx.Err() != nil || s.isClosing() {
break
}
serveErr = err
break
}
if slots := s.connSlots; slots != nil {
select {
case slots <- struct{}{}:
default:
_ = conn.Close()
continue
}
}
state := s.connectionPool.Get().(*connection)
state.server = s
state.conn = conn
state.active.Store(false)
state.closeWhenIdle.Store(false)
state.closed.Store(false)
state.slotHeld = s.connSlots != nil
state.closeDone = make(chan struct{})
s.mu.Lock()
if s.closing {
s.mu.Unlock()
state.close()
s.recycleConnection(state)
continue
}
s.conns[state] = struct{}{}
s.wg.Add(1)
s.mu.Unlock()
go s.serveConnection(baseCtx, state)
}
close(shutdownDone)
s.mu.Lock()
if s.listener == listener {
s.listener = nil
}
s.mu.Unlock()
if ctx.Err() != nil || s.isClosing() {
timeout := s.shutdownTimeout()
shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout)
shutdownErr := s.Shutdown(shutdownCtx)
cancel()
return errors.Join(shutdownErr, serveErr)
}
_ = listener.Close()
timeout := s.shutdownTimeout()
shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout)
shutdownErr := s.Shutdown(shutdownCtx)
cancel()
return errors.Join(shutdownErr, serveErr)
}
// Shutdown stops accepting requests, cancels active request contexts and waits
// for connections to finish. On timeout, active sockets are closed.
func (s *Server) Shutdown(ctx context.Context) error {
if ctx == nil {
ctx = context.Background()
}
s.mu.Lock()
s.closing = true
if s.cancel != nil {
s.cancel()
}
listener := s.listener
states := make([]*connection, 0, len(s.conns))
for state := range s.conns {
state.closeWhenIdle.Store(true)
states = append(states, state)
}
s.listener = nil
s.mu.Unlock()
if listener != nil {
_ = listener.Close()
}
for _, state := range states {
if !state.active.Load() {
state.close()
}
}
done := make(chan struct{})
go func() {
s.wg.Wait()
close(done)
}()
select {
case <-done:
return nil
case <-ctx.Done():
for _, state := range states {
state.close()
}
<-done
return ctx.Err()
}
}
func (s *Server) shutdownTimeout() time.Duration {
if value := s.WriteTimeout; value > 0 {
return value
}
return defaultWriteTimeout
}
func (s *Server) isClosing() bool {
s.mu.Lock()
defer s.mu.Unlock()
return s.closing
}
func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
defer s.wg.Done()
defer func() {
state.close()
s.recycleConnection(state)
}()
reader := s.readerPool.Get().(*bufio.Reader)
reader.Reset(state.conn)
defer func() {
reader.Reset(nil)
s.readerPool.Put(reader)
}()
firstRequest := true
for {
readTimeout := s.ReadHeaderTimeout
if readTimeout <= 0 {
readTimeout = defaultReadHeaderTimeout
}
if !firstRequest && s.IdleTimeout > 0 {
readTimeout = s.IdleTimeout
} else if s.IdleTimeout > 0 && s.IdleTimeout < readTimeout {
readTimeout = s.IdleTimeout
}
_ = state.conn.SetReadDeadline(time.Now().Add(readTimeout))
request, closeAfter, err := readRequest(reader, state.conn, s.MaxHeaderBytes, s.MaxBodyBytes, func() {
if s.ReadTimeout > 0 {
_ = state.conn.SetReadDeadline(time.Now().Add(s.ReadTimeout))
} else {
_ = state.conn.SetReadDeadline(time.Time{})
}
})
if err != nil {
if !errors.Is(err, io.EOF) && !errors.Is(err, net.ErrClosed) && !isTimeout(err) {
status := http.StatusBadRequest
message := "bad request\n"
var protocolErr *protocolError
if errors.As(err, &protocolErr) {
status = protocolErr.status
message = protocolErr.message + "\n"
}
_ = writeSimple(state.conn, status, message, true, s.WriteTimeout)
}
return
}
_ = state.conn.SetReadDeadline(time.Time{})
firstRequest = false
requestContext, cancel := context.WithCancel(baseCtx)
request = request.WithContext(requestContext)
state.active.Store(true)
writer := s.writerPool.Get().(*responseWriter)
writer.reset(request.Method)
requestSlot := false
if slots := s.requestSlots; slots != nil {
select {
case slots <- struct{}{}:
requestSlot = true
case <-baseCtx.Done():
cancel()
writer.reset("")
s.writerPool.Put(writer)
state.active.Store(false)
return
default:
writer.Header().Set("Content-Type", "text/plain; charset=utf-8")
writer.WriteHeader(http.StatusServiceUnavailable)
writer.body.WriteString("server busy\n")
closing := s.isClosing() || state.closeWhenIdle.Load()
_ = writer.writeTo(state.conn, closeAfter || closing, s.WriteTimeout)
cancel()
writer.reset("")
s.writerPool.Put(writer)
state.active.Store(false)
if closeAfter || closing {
return
}
continue
}
}
handler := s.handler
handler.ServeHTTP(writer, request)
cancel()
closing := s.isClosing() || state.closeWhenIdle.Load()
if err := writer.writeTo(state.conn, closeAfter || closing, s.WriteTimeout); err != nil {
if requestSlot {
<-s.requestSlots
}
writer.reset("")
s.writerPool.Put(writer)
state.active.Store(false)
return
}
if requestSlot {
<-s.requestSlots
}
writer.reset("")
s.writerPool.Put(writer)
state.active.Store(false)
if closeAfter || closing {
return
}
}
}
func (c *connection) close() {
if c.closed.CompareAndSwap(false, true) {
defer close(c.closeDone)
_ = c.conn.Close()
c.server.mu.Lock()
delete(c.server.conns, c)
c.server.mu.Unlock()
}
}
func (s *Server) recycleConnection(c *connection) {
<-c.closeDone
if c.slotHeld {
if slots := s.connSlots; slots != nil {
<-slots
}
c.slotHeld = false
}
s.connectionPool.Put(c)
}
type protocolError struct {
status int
message string
}
func (e *protocolError) Error() string { return e.message }
func readRequest(reader *bufio.Reader, conn net.Conn, maxHeaderBytes int, maxBodyBytes int64, setBodyDeadline func()) (*http.Request, bool, error) {
if maxHeaderBytes <= 0 {
maxHeaderBytes = defaultMaxHeaderBytes
}
if maxBodyBytes <= 0 {
maxBodyBytes = defaultMaxBodyBytes
}
used := 0
readLine := func() (string, error) {
line, err := reader.ReadString('\n')
used += len(line)
if used > maxHeaderBytes {
return "", &protocolError{status: http.StatusRequestHeaderFieldsTooLarge, message: "request headers too large"}
}
if err != nil {
return "", err
}
if !strings.HasSuffix(line, "\n") {
return "", &protocolError{status: http.StatusBadRequest, message: "malformed request line"}
}
return strings.TrimSuffix(strings.TrimSuffix(line, "\n"), "\r"), nil
}
line, err := readLine()
if err != nil {
return nil, false, err
}
parts := strings.Split(line, " ")
if len(parts) != 3 || !validToken(parts[0]) {
return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed request line"}
}
protoMajor, protoMinor, ok := parseProtocol(parts[2])
if !ok {
return nil, true, &protocolError{status: http.StatusHTTPVersionNotSupported, message: "unsupported HTTP version"}
}
target := parts[1]
var requestURL *url.URL
if target == "*" {
requestURL = &url.URL{Path: "*"}
} else {
requestURL, err = url.ParseRequestURI(target)
if err != nil || requestURL.Path == "" || !strings.HasPrefix(requestURL.Path, "/") {
return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed request target"}
}
}
header := make(http.Header)
for {
line, err = readLine()
if err != nil {
return nil, true, err
}
if line == "" {
break
}
if strings.HasPrefix(line, " ") || strings.HasPrefix(line, "\t") {
return nil, true, &protocolError{status: http.StatusBadRequest, message: "folded headers are not supported"}
}
index := strings.IndexByte(line, ':')
if index <= 0 {
return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed header"}
}
name := strings.TrimSpace(line[:index])
value := strings.TrimSpace(line[index+1:])
if !validToken(name) || strings.ContainsAny(value, "\r\n") {
return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed header"}
}
header.Add(name, value)
}
if value := header.Get("Expect"); value != "" {
return nil, true, &protocolError{status: http.StatusExpectationFailed, message: "unsupported expectation"}
}
if value := header.Get("Transfer-Encoding"); value != "" && !strings.EqualFold(value, "identity") {
return nil, true, &protocolError{status: http.StatusNotImplemented, message: "chunked requests are not supported"}
}
contentLength := int64(0)
if value := header.Get("Content-Length"); value != "" {
contentLength, err = strconv.ParseInt(strings.TrimSpace(value), 10, 64)
if err != nil || contentLength < 0 {
return nil, true, &protocolError{status: http.StatusBadRequest, message: "invalid content length"}
}
}
if contentLength > maxBodyBytes {
return nil, true, &protocolError{status: http.StatusRequestEntityTooLarge, message: "request body too large"}
}
body := make([]byte, contentLength)
if contentLength > 0 {
if setBodyDeadline != nil {
setBodyDeadline()
}
if _, err := io.ReadFull(reader, body); err != nil {
return nil, true, err
}
}
request := &http.Request{
Method: parts[0],
URL: requestURL,
Proto: parts[2],
ProtoMajor: protoMajor,
ProtoMinor: protoMinor,
Header: header,
Body: io.NopCloser(bytes.NewReader(body)),
ContentLength: contentLength,
Host: header.Get("Host"),
RequestURI: target,
RemoteAddr: conn.RemoteAddr().String(),
}
closeAfter := strings.EqualFold(header.Get("Connection"), "close") ||
(protoMajor == 1 && protoMinor == 0 && !strings.Contains(strings.ToLower(header.Get("Connection")), "keep-alive"))
request.Close = closeAfter
return request, closeAfter, nil
}
func parseProtocol(value string) (int, int, bool) {
if value != "HTTP/1.1" && value != "HTTP/1.0" {
return 0, 0, false
}
return 1, int(value[len(value)-1] - '0'), true
}
func validToken(value string) bool {
if value == "" {
return false
}
for _, char := range value {
if char <= ' ' || strings.ContainsRune("()<>@,;:\\\"/[]?={} \t", char) {
return false
}
}
return true
}
func isTimeout(err error) bool {
var netErr net.Error
return errors.As(err, &netErr) && netErr.Timeout()
}
func writeSimple(conn net.Conn, status int, body string, closeAfter bool, timeout time.Duration) error {
writer := newResponseWriter("")
writer.status = status
writer.wroteHeader = true
writer.body.WriteString(body)
writer.header.Set("Content-Type", "text/plain; charset=utf-8")
return writer.writeTo(conn, closeAfter, timeout)
}
type responseWriter struct {
header http.Header
status int
wroteHeader bool
body bytes.Buffer
requestMethod string
}
func newResponseWriter(method string) *responseWriter {
return &responseWriter{header: make(http.Header), requestMethod: method}
}
func (w *responseWriter) reset(method string) {
for key := range w.header {
delete(w.header, key)
}
w.status = 0
w.wroteHeader = false
if w.body.Cap() > maxPooledResponseBytes {
w.body = bytes.Buffer{}
} else {
w.body.Reset()
}
w.requestMethod = method
}
func (w *responseWriter) Header() http.Header { return w.header }
func (w *responseWriter) WriteHeader(status int) {
if w.wroteHeader {
return
}
w.status = status
w.wroteHeader = true
}
func (w *responseWriter) Write(data []byte) (int, error) {
if !w.wroteHeader {
w.WriteHeader(http.StatusOK)
}
return w.body.Write(data)
}
func (w *responseWriter) writeTo(conn net.Conn, closeAfter bool, timeout time.Duration) error {
if timeout > 0 {
_ = conn.SetWriteDeadline(time.Now().Add(timeout))
}
if !w.wroteHeader {
w.status = http.StatusOK
w.wroteHeader = true
}
body := w.body.Bytes()
noBody := w.requestMethod == http.MethodHead || w.status == http.StatusNoContent ||
w.status == http.StatusNotModified || (w.status >= 100 && w.status < 200)
if noBody {
body = nil
}
if w.header.Get("Content-Length") == "" && w.status != http.StatusNoContent && w.status != http.StatusNotModified {
w.header.Set("Content-Length", strconv.Itoa(len(w.body.Bytes())))
}
if closeAfter {
w.header.Set("Connection", "close")
}
keys := make([]string, 0, len(w.header))
for key := range w.header {
keys = append(keys, key)
}
sort.Strings(keys)
var output bytes.Buffer
output.WriteString("HTTP/1.1 ")
output.WriteString(strconv.Itoa(w.status))
output.WriteByte(' ')
output.WriteString(http.StatusText(w.status))
output.WriteString("\r\n")
for _, key := range keys {
for _, value := range w.header.Values(key) {
if strings.ContainsAny(value, "\r\n") {
return errors.New("invalid response header")
}
output.WriteString(key)
output.WriteString(": ")
output.WriteString(value)
output.WriteString("\r\n")
}
}
output.WriteString("\r\n")
output.Write(body)
_, err := conn.Write(output.Bytes())
_ = conn.SetWriteDeadline(time.Time{})
return err
}

301
transport/native/server_test.go Normal file
View File

@@ -0,0 +1,301 @@
package native_test
import (
"bufio"
"context"
"errors"
"fmt"
"io"
"net"
"net/http"
"strings"
"testing"
"time"
omni "git.yuntuo.hk/k2safe/go-omni"
"git.yuntuo.hk/k2safe/go-omni/transport/native"
netrouter "git.yuntuo.hk/k2safe/go-omni/transport/nethttp"
)
func TestServerKeepAliveAndApplicationBridge(t *testing.T) {
application := omni.NewWithRouter(netrouter.New())
application.GET("/hello", func(ctx *omni.Context) error {
return ctx.OK(map[string]string{"message": "hello"})
})
server, listener, done := startServer(t, application)
defer stopServer(t, server, listener, done)
conn, err := net.Dial("tcp", listener.Addr().String())
if err != nil {
t.Fatal(err)
}
defer conn.Close()
reader := bufio.NewReader(conn)
for index, closeAfter := range []bool{false, true} {
connectionHeader := "keep-alive"
if closeAfter {
connectionHeader = "close"
}
fmt.Fprintf(conn, "GET /hello HTTP/1.1\r\nHost: test\r\nConnection: %s\r\n\r\n", connectionHeader)
request := &http.Request{Method: http.MethodGet}
response, err := http.ReadResponse(reader, request)
if err != nil {
t.Fatalf("response %d: %v", index, err)
}
body, err := io.ReadAll(response.Body)
response.Body.Close()
if err != nil || response.StatusCode != http.StatusOK || !strings.Contains(string(body), "\"message\":\"hello\"") {
t.Fatalf("response %d: status=%d body=%s error=%v", index, response.StatusCode, body, err)
}
}
}
func TestServerRejectsUnsupportedBodyAndLimits(t *testing.T) {
application := omni.NewWithRouter(netrouter.New())
application.POST("/body", func(ctx *omni.Context) error { return ctx.OK("ok") })
server := native.New(application)
server.MaxBodyBytes = 4
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
done := make(chan error, 1)
go func() { done <- server.Serve(context.Background(), listener) }()
waitForServer(t, listener.Addr().String())
defer stopServer(t, server, listener, done)
for _, requestText := range []string{
"POST /body HTTP/1.1\r\nHost: test\r\nTransfer-Encoding: chunked\r\n\r\n4\r\ntest\r\n0\r\n\r\n",
"POST /body HTTP/1.1\r\nHost: test\r\nContent-Length: 5\r\n\r\ntest!",
} {
conn, err := net.Dial("tcp", listener.Addr().String())
if err != nil {
t.Fatal(err)
}
_, _ = io.WriteString(conn, requestText)
response, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodPost})
if err != nil {
conn.Close()
t.Fatal(err)
}
response.Body.Close()
if response.StatusCode != http.StatusNotImplemented && response.StatusCode != http.StatusRequestEntityTooLarge {
t.Fatalf("unexpected protocol status %d", response.StatusCode)
}
conn.Close()
}
}
func TestServerShutdownCancelsActiveHandler(t *testing.T) {
started := make(chan struct{})
application := omni.NewWithRouter(netrouter.New())
application.GET("/wait", func(ctx *omni.Context) error {
close(started)
<-ctx.Context().Done()
return ctx.Context().Err()
})
server, listener, done := startServer(t, application)
conn, err := net.Dial("tcp", listener.Addr().String())
if err != nil {
t.Fatal(err)
}
_, _ = io.WriteString(conn, "GET /wait HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n")
select {
case <-started:
case <-time.After(2 * time.Second):
t.Fatal("handler did not start")
}
shutdownContext, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := server.Shutdown(shutdownContext); err != nil {
t.Fatal(err)
}
response, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodGet})
if err != nil {
t.Fatal(err)
}
response.Body.Close()
if response.StatusCode != http.StatusInternalServerError {
t.Fatalf("expected canceled handler to become 500, got %d", response.StatusCode)
}
conn.Close()
select {
case err := <-done:
if err != nil && !errors.Is(err, net.ErrClosed) {
t.Fatal(err)
}
case <-time.After(2 * time.Second):
t.Fatal("server did not stop")
}
}
func TestServerConcurrentRequestLimit(t *testing.T) {
started := make(chan struct{})
release := make(chan struct{})
application := omni.NewWithRouter(netrouter.New())
application.GET("/work", func(ctx *omni.Context) error {
close(started)
<-release
return ctx.OK("done")
})
server := native.New(application)
server.MaxConcurrentRequests = 1
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
done := make(chan error, 1)
go func() { done <- server.Serve(context.Background(), listener) }()
waitForServer(t, listener.Addr().String())
defer func() {
stopServer(t, server, listener, done)
}()
first, err := net.Dial("tcp", listener.Addr().String())
if err != nil {
t.Fatal(err)
}
defer first.Close()
_, _ = io.WriteString(first, "GET /work HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n")
select {
case <-started:
case <-time.After(2 * time.Second):
t.Fatal("first handler did not start")
}
second, err := net.Dial("tcp", listener.Addr().String())
if err != nil {
t.Fatal(err)
}
defer second.Close()
_, _ = io.WriteString(second, "GET /work HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n")
response, err := http.ReadResponse(bufio.NewReader(second), &http.Request{Method: http.MethodGet})
if err != nil {
t.Fatal(err)
}
response.Body.Close()
if response.StatusCode != http.StatusServiceUnavailable {
t.Fatalf("expected 503 at concurrency limit, got %d", response.StatusCode)
}
close(release)
response, err = http.ReadResponse(bufio.NewReader(first), &http.Request{Method: http.MethodGet})
if err != nil {
t.Fatal(err)
}
response.Body.Close()
if response.StatusCode != http.StatusOK {
t.Fatalf("expected first request to complete, got %d", response.StatusCode)
}
}
func TestServerReadHeaderTimeout(t *testing.T) {
application := omni.NewWithRouter(netrouter.New())
application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })
server := native.New(application)
server.ReadHeaderTimeout = 25 * time.Millisecond
server.IdleTimeout = 0
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
done := make(chan error, 1)
go func() { done <- server.Serve(context.Background(), listener) }()
waitForServer(t, listener.Addr().String())
defer stopServer(t, server, listener, done)
conn, err := net.Dial("tcp", listener.Addr().String())
if err != nil {
t.Fatal(err)
}
defer conn.Close()
conn.SetReadDeadline(time.Now().Add(time.Second))
_, _ = io.WriteString(conn, "GET /health HTTP/1.1\r\nHost: test")
buffer := make([]byte, 1)
if _, err := conn.Read(buffer); err == nil {
t.Fatal("expected incomplete headers to time out and close")
}
}
func TestAppRunWithNativeRunner(t *testing.T) {
application := omni.NewWithRouter(netrouter.New())
application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })
address := freeAddress(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan error, 1)
go func() { done <- application.RunWith(ctx, native.New(nil), address) }()
waitForServer(t, address)
response, err := http.Get("http://" + address + "/health")
if err != nil {
t.Fatal(err)
}
response.Body.Close()
if response.StatusCode != http.StatusOK {
t.Fatalf("unexpected status %d", response.StatusCode)
}
cancel()
select {
case err := <-done:
if err != nil {
t.Fatal(err)
}
case <-time.After(3 * time.Second):
t.Fatal("RunWith did not stop")
}
}
func startServer(t *testing.T, handler http.Handler) (*native.Server, net.Listener, <-chan error) {
t.Helper()
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
server := native.New(handler)
done := make(chan error, 1)
go func() { done <- server.Serve(context.Background(), listener) }()
waitForServer(t, listener.Addr().String())
return server, listener, done
}
func stopServer(t *testing.T, server *native.Server, listener net.Listener, done <-chan error) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
_ = server.Shutdown(ctx)
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("server did not stop")
}
_ = listener.Close()
}
func freeAddress(t *testing.T) string {
t.Helper()
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
address := listener.Addr().String()
listener.Close()
return address
}
func waitForServer(t *testing.T, address string) {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for {
conn, err := net.DialTimeout("tcp", address, 50*time.Millisecond)
if err == nil {
conn.Close()
return
}
if time.Now().After(deadline) {
t.Fatalf("server did not start: %v", err)
}
time.Sleep(10 * time.Millisecond)
}
}