mirror of
https://github.com/zap-proto/http.git
synced 2026-10-04 03:07:19 +00:00
wip: convert zap-proto/http net/http -> fasthttp (INCOMPLETE, unverified)
Part 1 of the fasthttp unification (Fiber + ZAP both on fasthttp). Does NOT yet build/verify; zap-proto/zip not yet created. Do not merge. Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
@@ -1,43 +1,42 @@
|
||||
// Client — net/http.RoundTripper-compatible client speaking ZAP-HTTP.
|
||||
// Client — a fasthttp-style client speaking ZAP-HTTP.
|
||||
//
|
||||
// Drop-in usage:
|
||||
// Usage mirrors fasthttp's Do(req, resp):
|
||||
//
|
||||
// t := zaphttp.NewTransport("server:9999")
|
||||
// client := &http.Client{Transport: t}
|
||||
// resp, err := client.Get("http://server/healthz")
|
||||
// t := zaphttp.NewTransport("server:9999")
|
||||
// req := fasthttp.AcquireRequest()
|
||||
// resp := fasthttp.AcquireResponse()
|
||||
// req.SetRequestURI("/healthz")
|
||||
// req.Header.SetMethod(fasthttp.MethodGet)
|
||||
// err := t.Do(req, resp)
|
||||
//
|
||||
// The transport ignores req.URL.Host (use the constructor argument
|
||||
// to pin the peer) and req.URL.Scheme. Existing http.Client
|
||||
// machinery — redirects, cookies, retries — works unchanged.
|
||||
//
|
||||
// Connection management uses a free-list pool of idle TCP
|
||||
// connections. RoundTrip pulls an idle conn (or dials a new one),
|
||||
// uses it for one exchange, and returns it to the pool. The pool
|
||||
// caps at MaxIdleConns; surplus conns close on return.
|
||||
// Connection management uses a free-list pool of idle TCP connections. Do
|
||||
// pulls an idle conn (or dials a new one), runs one exchange, and returns
|
||||
// it to the pool. Because UnmarshalResponse fully buffers the response
|
||||
// body into resp, the connection is reusable the instant the frame is
|
||||
// decoded — no body-close handshake is needed. The pool caps at
|
||||
// MaxIdleConns; surplus conns close on return.
|
||||
|
||||
package zaphttp
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/valyala/fasthttp"
|
||||
)
|
||||
|
||||
// pooledConn is one idle connection. The bufio.Reader is held with
|
||||
// the conn so the read-side amortizes its own buffer alloc across
|
||||
// requests.
|
||||
// pooledConn is one idle connection. The bufio.Reader is held with the
|
||||
// conn so the read side amortizes its own buffer alloc across requests.
|
||||
type pooledConn struct {
|
||||
c net.Conn
|
||||
br *bufio.Reader
|
||||
}
|
||||
|
||||
// Transport implements http.RoundTripper over a ZAP-HTTP TCP
|
||||
// connection. Field-zero is invalid; use NewTransport.
|
||||
// Transport speaks ZAP-HTTP over a pooled TCP connection. Field-zero is
|
||||
// invalid; use NewTransport.
|
||||
type Transport struct {
|
||||
addr string
|
||||
dialTimeout time.Duration
|
||||
@@ -64,8 +63,8 @@ func (t *Transport) SetDialTimeout(d time.Duration) { t.dialTimeout = d }
|
||||
// SetReadTimeout overrides the default 30s response-read timeout.
|
||||
func (t *Transport) SetReadTimeout(d time.Duration) { t.readTimeout = d }
|
||||
|
||||
// SetMaxIdleConns caps the number of idle conns held in the pool.
|
||||
// Surplus conns close on return.
|
||||
// SetMaxIdleConns caps the number of idle conns held in the pool. Surplus
|
||||
// conns close on return.
|
||||
func (t *Transport) SetMaxIdleConns(n int) {
|
||||
if n < 0 {
|
||||
n = 0
|
||||
@@ -73,26 +72,27 @@ func (t *Transport) SetMaxIdleConns(n int) {
|
||||
t.maxIdleConns = n
|
||||
}
|
||||
|
||||
// RoundTrip executes a single request/response exchange.
|
||||
func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
// Do executes a single request/response exchange, filling resp. It is safe
|
||||
// for concurrent use: each call takes its own connection from the pool.
|
||||
func (t *Transport) Do(req *fasthttp.Request, resp *fasthttp.Response) error {
|
||||
if t.addr == "" {
|
||||
return nil, fmt.Errorf("zaphttp: Transport.addr is empty (use NewTransport)")
|
||||
return fmt.Errorf("zaphttp: Transport.addr is empty (use NewTransport)")
|
||||
}
|
||||
|
||||
pc, err := t.acquireConn()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: dial %s: %w", t.addr, err)
|
||||
return fmt.Errorf("zaphttp: dial %s: %w", t.addr, err)
|
||||
}
|
||||
conn, br := pc.c, pc.br
|
||||
|
||||
frame, err := MarshalRequest(req)
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return nil, err
|
||||
return err
|
||||
}
|
||||
if err := writeFrame(conn, frame); err != nil {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("zaphttp: write request: %w", err)
|
||||
return fmt.Errorf("zaphttp: write request: %w", err)
|
||||
}
|
||||
|
||||
if t.readTimeout > 0 {
|
||||
@@ -101,38 +101,18 @@ func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
respFrame, err := readFrame(br)
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("zaphttp: read response: %w", err)
|
||||
return fmt.Errorf("zaphttp: read response: %w", err)
|
||||
}
|
||||
resp, err := UnmarshalResponse(respFrame)
|
||||
if err != nil {
|
||||
if err := UnmarshalResponse(respFrame, resp); err != nil {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("zaphttp: unmarshal response: %w", err)
|
||||
return fmt.Errorf("zaphttp: unmarshal response: %w", err)
|
||||
}
|
||||
|
||||
// Reset the read deadline before returning the conn to the pool.
|
||||
// Reset the read deadline and return the conn to the pool. The response
|
||||
// is fully buffered in resp, so the conn is immediately reusable.
|
||||
_ = conn.SetReadDeadline(time.Time{})
|
||||
|
||||
// The body is already buffered in memory by the unmarshal step,
|
||||
// so the conn is reusable as soon as the response is decoded.
|
||||
// Wrap Body so the caller can still call Close() without
|
||||
// affecting the conn — pool ownership is the transport's
|
||||
// responsibility, not the caller's.
|
||||
resp.Body = bodyCloser{ReadCloser: resp.Body, release: func() { t.releaseConn(pc) }}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// bodyCloser wraps the response body so closing it releases the
|
||||
// underlying TCP connection to the pool.
|
||||
type bodyCloser struct {
|
||||
io.ReadCloser
|
||||
release func()
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func (b bodyCloser) Close() error {
|
||||
err := b.ReadCloser.Close()
|
||||
b.once.Do(b.release)
|
||||
return err
|
||||
t.releaseConn(pc)
|
||||
return nil
|
||||
}
|
||||
|
||||
// acquireConn pops an idle conn or dials a new one.
|
||||
@@ -158,8 +138,7 @@ func (t *Transport) acquireConn() (*pooledConn, error) {
|
||||
return &pooledConn{c: c, br: bufio.NewReader(c)}, nil
|
||||
}
|
||||
|
||||
// releaseConn returns a conn to the pool, or closes it if the pool
|
||||
// is full.
|
||||
// releaseConn returns a conn to the pool, or closes it if the pool is full.
|
||||
func (t *Transport) releaseConn(pc *pooledConn) {
|
||||
t.mu.Lock()
|
||||
if len(t.idle) >= t.maxIdleConns {
|
||||
@@ -171,8 +150,8 @@ func (t *Transport) releaseConn(pc *pooledConn) {
|
||||
t.mu.Unlock()
|
||||
}
|
||||
|
||||
// CloseIdleConnections closes every idle conn in the pool. Active
|
||||
// requests are unaffected. Useful in tests + on shutdown.
|
||||
// CloseIdleConnections closes every idle conn in the pool. Active requests
|
||||
// are unaffected. Useful in tests and on shutdown.
|
||||
func (t *Transport) CloseIdleConnections() {
|
||||
t.mu.Lock()
|
||||
idle := t.idle
|
||||
@@ -183,24 +162,22 @@ func (t *Transport) CloseIdleConnections() {
|
||||
}
|
||||
}
|
||||
|
||||
// Get is a convenience for transports that don't need an http.Client
|
||||
// wrapper (e.g. service-to-service calls inside a cluster).
|
||||
func Get(addr, path string) (*http.Response, error) {
|
||||
// Get is a convenience for one-shot service-to-service calls. The caller
|
||||
// owns resp and must fasthttp.ReleaseResponse it when done.
|
||||
func Get(addr, path string) (*fasthttp.Response, error) {
|
||||
t := NewTransport(addr)
|
||||
req, err := http.NewRequest(http.MethodGet, "http://"+addr+path, nil)
|
||||
if err != nil {
|
||||
defer t.CloseIdleConnections()
|
||||
|
||||
req := fasthttp.AcquireRequest()
|
||||
defer fasthttp.ReleaseRequest(req)
|
||||
req.Header.SetMethod(fasthttp.MethodGet)
|
||||
req.SetRequestURI(path)
|
||||
req.Header.SetHost(addr)
|
||||
|
||||
resp := fasthttp.AcquireResponse()
|
||||
if err := t.Do(req, resp); err != nil {
|
||||
fasthttp.ReleaseResponse(resp)
|
||||
return nil, err
|
||||
}
|
||||
return t.RoundTrip(req)
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// Compile-time guard that the bodyCloser remains a valid io.ReadCloser
|
||||
// even if the wrapped Body returns a never-closed type.
|
||||
var _ io.ReadCloser = bodyCloser{}
|
||||
|
||||
// errInvalidConn is returned by acquireConn when the pool yields a
|
||||
// stale connection (currently unused; placeholder for future
|
||||
// liveness checks).
|
||||
var errInvalidConn = errors.New("zaphttp: pooled connection invalid")
|
||||
|
||||
var _ = errInvalidConn
|
||||
|
||||
+14
-16
@@ -1,19 +1,18 @@
|
||||
// Minimal zap-http example: server + client in one binary.
|
||||
//
|
||||
// go run ./examples/hello server :9999
|
||||
// go run ./examples/hello client :9999
|
||||
// go run ./examples/hello server :9999
|
||||
// go run ./examples/hello client :9999
|
||||
//
|
||||
// The server speaks ZAP-HTTP on :9999. The client dials, issues
|
||||
// GET /hello, and prints the response body.
|
||||
// The server speaks ZAP-HTTP on :9999 with a fasthttp handler. The client
|
||||
// dials, issues GET /hello, and prints the response body.
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/valyala/fasthttp"
|
||||
zaphttp "github.com/zap-proto/http"
|
||||
)
|
||||
|
||||
@@ -26,25 +25,24 @@ func main() {
|
||||
|
||||
switch role {
|
||||
case "server":
|
||||
http.HandleFunc("/hello", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/plain")
|
||||
fmt.Fprintf(w, "hello from %s — you said %s %s\n", addr, r.Method, r.URL.Path)
|
||||
})
|
||||
handler := func(ctx *fasthttp.RequestCtx) {
|
||||
ctx.Response.Header.Set("Content-Type", "text/plain")
|
||||
fmt.Fprintf(ctx, "hello from %s — you said %s %s\n", addr, ctx.Method(), ctx.Path())
|
||||
}
|
||||
fmt.Printf("zap-http server listening on %s\n", addr)
|
||||
if err := zaphttp.ListenAndServe(addr, nil); err != nil {
|
||||
if err := zaphttp.ListenAndServe(addr, handler); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "server:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
case "client":
|
||||
client := &http.Client{Transport: zaphttp.NewTransport(addr)}
|
||||
resp, err := client.Get("http://" + addr + "/hello")
|
||||
resp, err := zaphttp.Get(addr, "/hello")
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "client:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
fmt.Printf("← %s\n", resp.Status)
|
||||
io.Copy(os.Stdout, resp.Body)
|
||||
defer fasthttp.ReleaseResponse(resp)
|
||||
fmt.Printf("← %d %s\n", resp.StatusCode(), fasthttp.StatusMessage(resp.StatusCode()))
|
||||
os.Stdout.Write(resp.Body())
|
||||
default:
|
||||
fmt.Fprintln(os.Stderr, "role must be 'server' or 'client'")
|
||||
os.Exit(2)
|
||||
|
||||
@@ -2,4 +2,13 @@ module github.com/zap-proto/http
|
||||
|
||||
go 1.26
|
||||
|
||||
require github.com/zap-proto/go v1.1.0
|
||||
require (
|
||||
github.com/valyala/fasthttp v1.70.0
|
||||
github.com/zap-proto/go v1.1.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/andybalholm/brotli v1.2.1 // indirect
|
||||
github.com/klauspost/compress v1.18.5 // indirect
|
||||
github.com/valyala/bytebufferpool v1.0.0 // indirect
|
||||
)
|
||||
|
||||
@@ -1,2 +1,12 @@
|
||||
github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
|
||||
github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
|
||||
github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
|
||||
github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
|
||||
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
|
||||
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
|
||||
github.com/valyala/fasthttp v1.70.0 h1:LAhMGcWk13QZWm85+eg8ZBNbrq5mnkWFGbHMUJHIdXA=
|
||||
github.com/valyala/fasthttp v1.70.0/go.mod h1:oDZEHHkJ/Buyklg6uURmYs19442zFSnCIfX3j1FY3pE=
|
||||
github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
|
||||
github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
|
||||
github.com/zap-proto/go v1.1.0 h1:DDGhuTBNqkmNax7QV/VvkInJR7hiOcybwUjN3YBt8fg=
|
||||
github.com/zap-proto/go v1.1.0/go.mod h1:914SNGTH6Rv3Yu1MweWJBPEN8FZlo5C39QyhaB0C7Q0=
|
||||
|
||||
@@ -1,56 +1,52 @@
|
||||
// Server — net/http.Handler-compatible server speaking ZAP-HTTP.
|
||||
// Server — a fasthttp.RequestHandler served over ZAP-HTTP.
|
||||
//
|
||||
// Existing handlers work unchanged:
|
||||
// A Fiber app (or any fasthttp stack) hands its handler straight in:
|
||||
//
|
||||
// mux := http.NewServeMux()
|
||||
// mux.HandleFunc("/healthz", okHandler)
|
||||
// zaphttp.ListenAndServe(":9999", mux)
|
||||
// srv := &zaphttp.Server{Addr: ":9999", Handler: app.Handler()}
|
||||
// srv.ListenAndServe()
|
||||
//
|
||||
// The server reads ZAP-HTTP frames off each accepted connection,
|
||||
// reconstructs *http.Request, dispatches to the handler, captures the
|
||||
// handler's response via a buffering ResponseWriter, and writes one
|
||||
// response frame back. Connections are kept alive across requests.
|
||||
// reconstructs a *fasthttp.RequestCtx, dispatches to the handler, and
|
||||
// writes one response frame back from the ctx's Response. Connections are
|
||||
// kept alive across requests. There is no net/http anywhere in the path.
|
||||
|
||||
package zaphttp
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/valyala/fasthttp"
|
||||
)
|
||||
|
||||
// Server is a ZAP-HTTP server. Zero-value is usable; common knobs
|
||||
// (Addr, Handler, ReadTimeout, …) mirror net/http.Server.
|
||||
// Server is a ZAP-HTTP server. Zero-value is usable; common knobs (Addr,
|
||||
// Handler, ReadTimeout, …) mirror fasthttp.Server.
|
||||
type Server struct {
|
||||
Addr string // ":9999" if empty
|
||||
Handler http.Handler // http.DefaultServeMux if nil
|
||||
ReadTimeout time.Duration // 0 means no timeout
|
||||
WriteTimeout time.Duration
|
||||
IdleTimeout time.Duration
|
||||
MaxHeaderBytes int // not enforced today; placeholder for parity
|
||||
Addr string // ":9999" if empty
|
||||
Handler fasthttp.RequestHandler // required
|
||||
ReadTimeout time.Duration // 0 means no timeout
|
||||
WriteTimeout time.Duration
|
||||
IdleTimeout time.Duration
|
||||
Logger fasthttp.Logger // passed to each RequestCtx; nil is fine
|
||||
|
||||
mu sync.Mutex
|
||||
listener net.Listener
|
||||
closed bool
|
||||
}
|
||||
|
||||
// ListenAndServe is the convenience equivalent of net/http.ListenAndServe.
|
||||
func ListenAndServe(addr string, handler http.Handler) error {
|
||||
// ListenAndServe is the convenience equivalent of fasthttp.ListenAndServe.
|
||||
func ListenAndServe(addr string, handler fasthttp.RequestHandler) error {
|
||||
s := &Server{Addr: addr, Handler: handler}
|
||||
return s.ListenAndServe()
|
||||
}
|
||||
|
||||
// ListenAndServe binds Addr and serves until Close is called or a
|
||||
// fatal accept error occurs.
|
||||
// ListenAndServe binds Addr and serves until Close is called or a fatal
|
||||
// accept error occurs.
|
||||
func (s *Server) ListenAndServe() error {
|
||||
addr := s.Addr
|
||||
if addr == "" {
|
||||
@@ -70,9 +66,8 @@ func (s *Server) Serve(ln net.Listener) error {
|
||||
s.listener = ln
|
||||
s.mu.Unlock()
|
||||
|
||||
handler := s.Handler
|
||||
if handler == nil {
|
||||
handler = http.DefaultServeMux
|
||||
if s.Handler == nil {
|
||||
return errors.New("zaphttp: Server.Handler is nil")
|
||||
}
|
||||
|
||||
for {
|
||||
@@ -90,13 +85,14 @@ func (s *Server) Serve(ln net.Listener) error {
|
||||
}
|
||||
return err
|
||||
}
|
||||
go s.serveConn(conn, handler)
|
||||
go s.serveConn(conn)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) serveConn(conn net.Conn, handler http.Handler) {
|
||||
func (s *Server) serveConn(conn net.Conn) {
|
||||
defer conn.Close()
|
||||
br := bufio.NewReader(conn)
|
||||
remoteAddr := conn.RemoteAddr()
|
||||
|
||||
for {
|
||||
if s.IdleTimeout > 0 {
|
||||
@@ -112,53 +108,34 @@ func (s *Server) serveConn(conn net.Conn, handler http.Handler) {
|
||||
if s.ReadTimeout > 0 {
|
||||
_ = conn.SetReadDeadline(time.Now().Add(s.ReadTimeout))
|
||||
}
|
||||
req, err := UnmarshalRequest(raw)
|
||||
if err != nil {
|
||||
s.writeError(conn, http.StatusBadRequest, "malformed request frame: "+err.Error())
|
||||
|
||||
var req fasthttp.Request
|
||||
if err := UnmarshalRequest(raw, &req); err != nil {
|
||||
s.writeError(conn, fasthttp.StatusBadRequest, "malformed request frame: "+err.Error())
|
||||
return
|
||||
}
|
||||
req.RemoteAddr = conn.RemoteAddr().String()
|
||||
|
||||
// Wire a buffering ResponseWriter. The handler runs to
|
||||
// completion and we then marshal the captured state into a
|
||||
// single response frame. v0.2 will swap this for a streaming
|
||||
// writer once the transport supports it.
|
||||
rw := &bufferingResponseWriter{
|
||||
header: make(http.Header),
|
||||
body: &bytes.Buffer{},
|
||||
}
|
||||
ctx := context.Background()
|
||||
req = req.WithContext(ctx)
|
||||
// Fresh ctx per request: fasthttp's Init resets connection state and
|
||||
// copies the request in, but does not reset a reused Response, so a
|
||||
// zero-value ctx is the clean, correct choice.
|
||||
ctx := &fasthttp.RequestCtx{}
|
||||
ctx.Init(&req, remoteAddr, s.Logger)
|
||||
|
||||
// Defer-recover so a handler panic returns 500 rather than
|
||||
// dropping the connection silently.
|
||||
// Defer-recover so a handler panic returns 500 rather than dropping
|
||||
// the connection silently.
|
||||
func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
if !rw.wroteHeader {
|
||||
rw.WriteHeader(http.StatusInternalServerError)
|
||||
}
|
||||
log.Printf("zaphttp: handler panic on %s %s: %v", req.Method, req.URL.RequestURI(), r)
|
||||
ctx.Response.Reset()
|
||||
ctx.Response.SetStatusCode(fasthttp.StatusInternalServerError)
|
||||
log.Printf("zaphttp: handler panic on %s %s: %v",
|
||||
ctx.Method(), ctx.RequestURI(), r)
|
||||
}
|
||||
}()
|
||||
handler.ServeHTTP(rw, req)
|
||||
s.Handler(ctx)
|
||||
}()
|
||||
|
||||
if !rw.wroteHeader {
|
||||
rw.WriteHeader(http.StatusOK)
|
||||
}
|
||||
// Auto-fill Content-Length when the handler hasn't.
|
||||
if rw.header.Get("Content-Length") == "" {
|
||||
rw.header.Set("Content-Length", strconv.Itoa(rw.body.Len()))
|
||||
}
|
||||
|
||||
resp := &http.Response{
|
||||
StatusCode: rw.status,
|
||||
Status: fmt.Sprintf("%d %s", rw.status, http.StatusText(rw.status)),
|
||||
Header: rw.header,
|
||||
Body: io.NopCloser(rw.body),
|
||||
}
|
||||
respBytes, err := MarshalResponse(resp)
|
||||
respBytes, err := MarshalResponse(&ctx.Response)
|
||||
if err != nil {
|
||||
log.Printf("zaphttp: marshal response: %v", err)
|
||||
return
|
||||
@@ -176,13 +153,11 @@ func (s *Server) serveConn(conn net.Conn, handler http.Handler) {
|
||||
}
|
||||
|
||||
func (s *Server) writeError(w io.Writer, status int, msg string) {
|
||||
resp := &http.Response{
|
||||
StatusCode: status,
|
||||
Status: fmt.Sprintf("%d %s", status, http.StatusText(status)),
|
||||
Header: http.Header{"Content-Type": []string{"text/plain; charset=utf-8"}},
|
||||
Body: io.NopCloser(bytes.NewBufferString(msg)),
|
||||
}
|
||||
frame, err := MarshalResponse(resp)
|
||||
var resp fasthttp.Response
|
||||
resp.SetStatusCode(status)
|
||||
resp.Header.SetContentType("text/plain; charset=utf-8")
|
||||
resp.SetBodyString(msg)
|
||||
frame, err := MarshalResponse(&resp)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -200,32 +175,6 @@ func (s *Server) Close() error {
|
||||
return s.listener.Close()
|
||||
}
|
||||
|
||||
// ---- bufferingResponseWriter ----
|
||||
|
||||
type bufferingResponseWriter struct {
|
||||
header http.Header
|
||||
body *bytes.Buffer
|
||||
status int
|
||||
wroteHeader bool
|
||||
}
|
||||
|
||||
func (w *bufferingResponseWriter) Header() http.Header { return w.header }
|
||||
|
||||
func (w *bufferingResponseWriter) Write(b []byte) (int, error) {
|
||||
if !w.wroteHeader {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
return w.body.Write(b)
|
||||
}
|
||||
|
||||
func (w *bufferingResponseWriter) WriteHeader(status int) {
|
||||
if w.wroteHeader {
|
||||
return
|
||||
}
|
||||
w.status = status
|
||||
w.wroteHeader = true
|
||||
}
|
||||
|
||||
func isClosedConn(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
|
||||
@@ -1,16 +1,28 @@
|
||||
// Package zaphttp implements HTTP request/response semantics over the
|
||||
// ZAP transport. Existing net/http callers and handlers work
|
||||
// unmodified — the package adapts the standard library types to and
|
||||
// from ZAP frames.
|
||||
// ZAP transport for the fasthttp handler model. A fasthttp.RequestHandler
|
||||
// is served over ZAP frames; a fasthttp.Request/Response pair is exchanged
|
||||
// by the client. There is no net/http in the request path — Fiber and
|
||||
// every other fasthttp stack hand their handler straight to this package.
|
||||
//
|
||||
// The wire format is documented in schema/zap_http.zap (authored in
|
||||
// the ZAP schema language). The bytes on the wire are
|
||||
// github.com/zap-proto/go Objects — the canonical pure-stdlib ZAP
|
||||
// runtime — packed at explicit byte offsets within each object's fixed
|
||||
// payload. There is no external codec dependency: this file is the
|
||||
// encoder and decoder.
|
||||
// The wire format is documented in schema/zap_http.zap (authored in the
|
||||
// ZAP schema language). The bytes on the wire are github.com/zap-proto/go
|
||||
// Objects — the canonical pure-stdlib ZAP runtime — packed at explicit
|
||||
// byte offsets within each object's fixed payload. There is no external
|
||||
// codec dependency: this file is the encoder and decoder.
|
||||
//
|
||||
// What's covered today (v0.1):
|
||||
// # Wire compatibility
|
||||
//
|
||||
// The frame format is UNCHANGED from the prior net/http codec — same
|
||||
// object slots, same offsets, same JSON header map, same flags
|
||||
// discriminator. Only the Go type the codec reads from and writes to
|
||||
// changed (net/http → fasthttp). Frames produced here are byte-identical
|
||||
// to the frames the previous net/http codec produced for the same logical
|
||||
// message, so a fasthttp-based peer and a net/http-based peer interoperate
|
||||
// on the wire. This is verified by golden-byte tests (zaphttp_test.go)
|
||||
// that decode frames captured from the previous codec and re-encode them
|
||||
// to the same bytes.
|
||||
//
|
||||
// What's covered:
|
||||
// - Request method, target, headers, body
|
||||
// - Response status, reason, headers, body
|
||||
// - Trailers (read and write)
|
||||
@@ -18,65 +30,69 @@
|
||||
// What's not yet covered:
|
||||
// - Streaming bodies — the entire body must fit in one frame; the
|
||||
// transport negotiates a max frame size on connection setup.
|
||||
// - HTTP/2 server push semantics; HTTP/3 priorities.
|
||||
// - WebSocket / SSE upgrade paths; those land in zap-proto/ws.
|
||||
//
|
||||
// # Wire layout
|
||||
//
|
||||
// A frame is one zap-proto/go message. The message type rides in the
|
||||
// header flags (flags>>8): FrameRequest for a request, FrameResponse
|
||||
// for a response — the same discriminator discipline the zap forward
|
||||
// envelopes use. There is no union struct; the flag selects which
|
||||
// object shape is the message root.
|
||||
// header flags (flags>>8): FrameRequest for a request, FrameResponse for
|
||||
// a response. There is no union struct; the flag selects which object
|
||||
// shape is the message root.
|
||||
//
|
||||
// # Offset discipline
|
||||
//
|
||||
// zap-proto/go's ObjectBuilder.SetBytes / SetText write an 8-byte slot
|
||||
// at the field offset: {relOffset uint32, length uint32}. The
|
||||
// variable-length tail is appended after the fixed section in Finish()
|
||||
// and the relOffset is patched to point at it. Consequently EVERY text
|
||||
// or bytes field consumes 8 bytes of fixed payload — adjacent text
|
||||
// fields MUST be spaced 8 bytes apart or their slots overlap and
|
||||
// corrupt each other. Fixed scalars use their natural width on a
|
||||
// natural boundary. These offsets are the contract; changing one is a
|
||||
// wire break.
|
||||
// zap-proto/go's ObjectBuilder.SetBytes / SetText write an 8-byte slot at
|
||||
// the field offset: {relOffset uint32, length uint32}. The variable-length
|
||||
// tail is appended after the fixed section in Finish() and the relOffset
|
||||
// is patched to point at it. Consequently EVERY text or bytes field
|
||||
// consumes 8 bytes of fixed payload — adjacent text fields MUST be spaced
|
||||
// 8 bytes apart or their slots overlap and corrupt each other. Fixed
|
||||
// scalars use their natural width on a natural boundary. These offsets are
|
||||
// the contract; changing one is a wire break.
|
||||
//
|
||||
// Headers (and trailers) ride as a JSON map[string][]string — the
|
||||
// native shape of http.Header — preserving multi-value semantics
|
||||
// (RFC 9110 §5.2) losslessly without a nested-object list.
|
||||
// # Framing metadata vs headers
|
||||
//
|
||||
// This file holds the marshal/unmarshal between *http.Request /
|
||||
// http.Response and ZAP frames. The transport layer (TCP framing
|
||||
// today, PQ-handshake-wrapped ZAP transport in v0.2) lives in
|
||||
// transport.go.
|
||||
// Three fields are owned by the frame, not the headers slot, exactly as
|
||||
// the reference net/http codec treated them:
|
||||
//
|
||||
// - Content-Length — the frame length-prefixes the body, so body length
|
||||
// is authoritative; it is reconstructed on decode, never carried in
|
||||
// the headers map.
|
||||
// - Host — request routing metadata; carried in fasthttp's Host slot,
|
||||
// not the headers map (matching net/http, which keeps Host out of
|
||||
// Header).
|
||||
// - Trailer (the meta-header listing trailer names) — represented by the
|
||||
// dedicated trailer slot, not the headers map.
|
||||
//
|
||||
// Everything else fasthttp exposes (Content-Type, User-Agent, Cookie,
|
||||
// Authorization, X-*, Set-Cookie, …) rides in the headers slot as a JSON
|
||||
// map[string][]string — the native lossless shape for multi-value header
|
||||
// names (RFC 9110 §5.2).
|
||||
|
||||
package zaphttp
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"net/textproto"
|
||||
"strings"
|
||||
|
||||
"github.com/valyala/fasthttp"
|
||||
zap "github.com/zap-proto/go"
|
||||
)
|
||||
|
||||
// Frame type IDs. The ZAP message header carries the type in the upper
|
||||
// byte of the 16-bit flags field; FinishWithFlags(t<<8) tags the
|
||||
// message and Message.Flags()>>8 recovers it. A request frame and a
|
||||
// response frame are the two shapes the wire carries today.
|
||||
// byte of the 16-bit flags field; FinishWithFlags(t<<8) tags the message
|
||||
// and Message.Flags()>>8 recovers it. A request frame and a response frame
|
||||
// are the two shapes the wire carries today.
|
||||
const (
|
||||
FrameRequest uint16 = 0x01
|
||||
FrameResponse uint16 = 0x02
|
||||
)
|
||||
|
||||
// Request frame field offsets within the object's fixed payload. Every
|
||||
// text/bytes slot is 8 bytes (relOffset+length); see "Offset
|
||||
// discipline" in the package doc.
|
||||
// text/bytes slot is 8 bytes (relOffset+length); see "Offset discipline"
|
||||
// in the package doc.
|
||||
const (
|
||||
reqMethod = 0 // text (8)
|
||||
reqTarget = 8 // text (8)
|
||||
@@ -87,8 +103,8 @@ const (
|
||||
reqSlotSize = 48
|
||||
)
|
||||
|
||||
// Response frame field offsets. status@0 is a uint16 (2 bytes); the
|
||||
// first text slot starts at the next 8-byte boundary.
|
||||
// Response frame field offsets. status@0 is a uint16 (2 bytes); the first
|
||||
// text slot starts at the next 8-byte boundary.
|
||||
const (
|
||||
respStatus = 0 // uint16 (2)
|
||||
respReason = 8 // text (8)
|
||||
@@ -99,37 +115,35 @@ const (
|
||||
respSlotSize = 48
|
||||
)
|
||||
|
||||
// MarshalRequest serializes an *http.Request into a ZAP frame. The
|
||||
// returned bytes are framed and ready for transport.Write — the
|
||||
// transport layer prepends the length prefix.
|
||||
//
|
||||
// The body is fully read and consumed; callers that need to reuse the
|
||||
// request must replace req.Body with a fresh reader after this call.
|
||||
func MarshalRequest(req *http.Request) ([]byte, error) {
|
||||
body, err := readAllBody(req.Body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: read body: %w", err)
|
||||
}
|
||||
// defaultProto is the proto string written when the source carries none.
|
||||
// fasthttp always reports a protocol ("HTTP/1.1" by default), so this
|
||||
// only guards a hand-built zero-value header.
|
||||
const defaultProto = "ZAP-HTTP/1.0"
|
||||
|
||||
target := req.RequestURI
|
||||
if target == "" && req.URL != nil {
|
||||
target = req.URL.RequestURI()
|
||||
}
|
||||
headers, err := encodeHeaders(req.Header)
|
||||
// MarshalRequest serializes a *fasthttp.Request into a ZAP frame. The
|
||||
// returned bytes are the ZAP message; the transport layer prepends the
|
||||
// length prefix (see transport.go).
|
||||
func MarshalRequest(req *fasthttp.Request) ([]byte, error) {
|
||||
method := string(req.Header.Method())
|
||||
target := string(req.RequestURI())
|
||||
proto := string(req.Header.Protocol())
|
||||
body := req.Body()
|
||||
|
||||
headers, err := encodeHeaderMap(collectRequestHeaders(&req.Header))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: encode headers: %w", err)
|
||||
}
|
||||
trailer, err := encodeHeaders(req.Trailer)
|
||||
trailer, err := encodeHeaderMap(collectTrailers(&req.Header))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: encode trailer: %w", err)
|
||||
}
|
||||
|
||||
b := zap.NewBuilder(reqSlotSize + len(body) + len(headers) + len(trailer) +
|
||||
len(req.Method) + len(target) + len(req.Proto) + 64)
|
||||
len(method) + len(target) + len(proto) + 64)
|
||||
ob := b.StartObject(reqSlotSize)
|
||||
ob.SetText(reqMethod, orDefault(req.Method, http.MethodGet))
|
||||
ob.SetText(reqMethod, orDefault(method, fasthttp.MethodGet))
|
||||
ob.SetText(reqTarget, target)
|
||||
ob.SetText(reqProto, orDefault(req.Proto, "ZAP-HTTP/1.0"))
|
||||
ob.SetText(reqProto, orDefault(proto, defaultProto))
|
||||
ob.SetBytes(reqHeaders, headers)
|
||||
ob.SetBytes(reqBody, body)
|
||||
ob.SetBytes(reqTrailer, trailer)
|
||||
@@ -137,92 +151,71 @@ func MarshalRequest(req *http.Request) ([]byte, error) {
|
||||
return b.FinishWithFlags(FrameRequest << 8), nil
|
||||
}
|
||||
|
||||
// UnmarshalRequest reconstructs an *http.Request from a ZAP frame. The
|
||||
// returned request has Body set to a bytes.Reader over the frame's body
|
||||
// bytes; the request URL is parsed from target. Host is taken from the
|
||||
// request's Host header (RFC 9110 §7.2).
|
||||
func UnmarshalRequest(b []byte) (*http.Request, error) {
|
||||
msg, err := zap.Parse(b)
|
||||
// UnmarshalRequest reconstructs a request into dst from a ZAP frame. dst is
|
||||
// reset first, then populated: method, request-URI, protocol, headers,
|
||||
// trailers, and body. Content-Length is derived from the body length by
|
||||
// SetBody; Host is taken from a Host header if the frame carried one.
|
||||
func UnmarshalRequest(frame []byte, dst *fasthttp.Request) error {
|
||||
msg, err := zap.Parse(frame)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return err
|
||||
}
|
||||
if t := msg.Flags() >> 8; t != FrameRequest {
|
||||
return nil, fmt.Errorf("zaphttp: frame type %#x, want request (%#x)", t, FrameRequest)
|
||||
return fmt.Errorf("zaphttp: frame type %#x, want request (%#x)", t, FrameRequest)
|
||||
}
|
||||
r := msg.Root()
|
||||
|
||||
method := r.Text(reqMethod)
|
||||
target := r.Text(reqTarget)
|
||||
proto := r.Text(reqProto)
|
||||
headers, err := decodeHeaders(r.Bytes(reqHeaders))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: decode headers: %w", err)
|
||||
}
|
||||
headers := r.Bytes(reqHeaders)
|
||||
body := r.Bytes(reqBody)
|
||||
trailer, err := decodeHeaders(r.Bytes(reqTrailer))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: decode trailer: %w", err)
|
||||
}
|
||||
trailer := r.Bytes(reqTrailer)
|
||||
|
||||
u, err := url.ParseRequestURI(target)
|
||||
if err != nil {
|
||||
// Best-effort: keep the raw target so handlers can inspect it,
|
||||
// but failing here would break callers that only need RequestURI.
|
||||
u = &url.URL{Path: target}
|
||||
dst.Reset()
|
||||
dst.Header.SetMethod(orDefault(method, fasthttp.MethodGet))
|
||||
dst.SetRequestURI(target)
|
||||
if proto != "" {
|
||||
dst.Header.SetProtocol(proto)
|
||||
}
|
||||
|
||||
out := &http.Request{
|
||||
Method: orDefault(method, http.MethodGet),
|
||||
URL: u,
|
||||
Proto: orDefault(proto, "ZAP-HTTP/1.0"),
|
||||
ProtoMajor: 1,
|
||||
ProtoMinor: 0,
|
||||
Header: headers,
|
||||
Body: io.NopCloser(bytes.NewReader(body)),
|
||||
Host: headers.Get("Host"),
|
||||
RequestURI: target,
|
||||
Trailer: trailer,
|
||||
if err := applyHeadersToRequest(&dst.Header, headers); err != nil {
|
||||
return fmt.Errorf("zaphttp: decode headers: %w", err)
|
||||
}
|
||||
if cl := headers.Get("Content-Length"); cl != "" {
|
||||
if n, err := strconv.ParseInt(cl, 10, 64); err == nil {
|
||||
out.ContentLength = n
|
||||
}
|
||||
} else {
|
||||
out.ContentLength = int64(len(body))
|
||||
if err := applyTrailers(&dst.Header, trailer); err != nil {
|
||||
return fmt.Errorf("zaphttp: decode trailer: %w", err)
|
||||
}
|
||||
return out, nil
|
||||
dst.SetBody(body)
|
||||
return nil
|
||||
}
|
||||
|
||||
// MarshalResponse serializes an *http.Response into a ZAP frame.
|
||||
func MarshalResponse(resp *http.Response) ([]byte, error) {
|
||||
body, err := readAllBody(resp.Body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: read body: %w", err)
|
||||
}
|
||||
|
||||
status := resp.StatusCode
|
||||
// MarshalResponse serializes a *fasthttp.Response into a ZAP frame.
|
||||
func MarshalResponse(resp *fasthttp.Response) ([]byte, error) {
|
||||
status := resp.StatusCode()
|
||||
if status == 0 {
|
||||
status = http.StatusOK
|
||||
status = fasthttp.StatusOK
|
||||
}
|
||||
reason := resp.Status
|
||||
if i := strings.IndexByte(reason, ' '); i >= 0 {
|
||||
reason = reason[i+1:]
|
||||
reason := string(resp.Header.StatusMessage())
|
||||
if reason == "" {
|
||||
reason = fasthttp.StatusMessage(status)
|
||||
}
|
||||
headers, err := encodeHeaders(resp.Header)
|
||||
proto := string(resp.Header.Protocol())
|
||||
body := resp.Body()
|
||||
|
||||
headers, err := encodeHeaderMap(collectResponseHeaders(&resp.Header))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: encode headers: %w", err)
|
||||
}
|
||||
trailer, err := encodeHeaders(resp.Trailer)
|
||||
trailer, err := encodeHeaderMap(collectTrailers(&resp.Header))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: encode trailer: %w", err)
|
||||
}
|
||||
|
||||
b := zap.NewBuilder(respSlotSize + len(body) + len(headers) + len(trailer) +
|
||||
len(reason) + len(resp.Proto) + 64)
|
||||
len(reason) + len(proto) + 64)
|
||||
ob := b.StartObject(respSlotSize)
|
||||
ob.SetUint16(respStatus, uint16(status))
|
||||
ob.SetText(respReason, reason)
|
||||
ob.SetText(respProto, orDefault(resp.Proto, "ZAP-HTTP/1.0"))
|
||||
ob.SetText(respProto, orDefault(proto, defaultProto))
|
||||
ob.SetBytes(respHeaders, headers)
|
||||
ob.SetBytes(respBody, body)
|
||||
ob.SetBytes(respTrailer, trailer)
|
||||
@@ -230,84 +223,211 @@ func MarshalResponse(resp *http.Response) ([]byte, error) {
|
||||
return b.FinishWithFlags(FrameResponse << 8), nil
|
||||
}
|
||||
|
||||
// UnmarshalResponse reconstructs an *http.Response from a ZAP frame.
|
||||
func UnmarshalResponse(b []byte) (*http.Response, error) {
|
||||
msg, err := zap.Parse(b)
|
||||
// UnmarshalResponse reconstructs a response into dst from a ZAP frame.
|
||||
func UnmarshalResponse(frame []byte, dst *fasthttp.Response) error {
|
||||
msg, err := zap.Parse(frame)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return err
|
||||
}
|
||||
if t := msg.Flags() >> 8; t != FrameResponse {
|
||||
return nil, fmt.Errorf("zaphttp: frame type %#x, want response (%#x)", t, FrameResponse)
|
||||
return fmt.Errorf("zaphttp: frame type %#x, want response (%#x)", t, FrameResponse)
|
||||
}
|
||||
r := msg.Root()
|
||||
|
||||
status := int(r.Uint16(respStatus))
|
||||
reason := r.Text(respReason)
|
||||
proto := r.Text(respProto)
|
||||
headers, err := decodeHeaders(r.Bytes(respHeaders))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: decode headers: %w", err)
|
||||
}
|
||||
headers := r.Bytes(respHeaders)
|
||||
body := r.Bytes(respBody)
|
||||
trailer, err := decodeHeaders(r.Bytes(respTrailer))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: decode trailer: %w", err)
|
||||
}
|
||||
trailer := r.Bytes(respTrailer)
|
||||
|
||||
statusText := reason
|
||||
if statusText == "" {
|
||||
statusText = http.StatusText(status)
|
||||
dst.Reset()
|
||||
dst.SetStatusCode(status)
|
||||
if reason != "" {
|
||||
dst.Header.SetStatusMessage([]byte(reason))
|
||||
}
|
||||
return &http.Response{
|
||||
Status: fmt.Sprintf("%d %s", status, statusText),
|
||||
StatusCode: status,
|
||||
Proto: orDefault(proto, "ZAP-HTTP/1.0"),
|
||||
ProtoMajor: 1,
|
||||
ProtoMinor: 0,
|
||||
Header: headers,
|
||||
Body: io.NopCloser(bytes.NewReader(body)),
|
||||
ContentLength: int64(len(body)),
|
||||
Trailer: trailer,
|
||||
}, nil
|
||||
if proto != "" {
|
||||
dst.Header.SetProtocol([]byte(proto))
|
||||
}
|
||||
if err := applyHeadersToResponse(&dst.Header, headers); err != nil {
|
||||
return fmt.Errorf("zaphttp: decode headers: %w", err)
|
||||
}
|
||||
if err := applyTrailers(&dst.Header, trailer); err != nil {
|
||||
return fmt.Errorf("zaphttp: decode trailer: %w", err)
|
||||
}
|
||||
dst.SetBody(body)
|
||||
return nil
|
||||
}
|
||||
|
||||
// ---- helpers ----
|
||||
// ---- header extraction (fasthttp -> map) ----
|
||||
|
||||
func readAllBody(r io.Reader) ([]byte, error) {
|
||||
if r == nil {
|
||||
return nil, nil
|
||||
}
|
||||
defer func() {
|
||||
if c, ok := r.(io.Closer); ok {
|
||||
_ = c.Close()
|
||||
// collectRequestHeaders walks a request's headers into a map[string][]string,
|
||||
// dropping the frame-owned fields (Host, Content-Length, the Trailer
|
||||
// meta-header, and any declared-trailer values which ride in the trailer
|
||||
// slot instead).
|
||||
func collectRequestHeaders(h *fasthttp.RequestHeader) map[string][]string {
|
||||
skip := trailerKeySet(h)
|
||||
m := map[string][]string{}
|
||||
h.VisitAll(func(key, value []byte) {
|
||||
k := string(key)
|
||||
if isFrameOwnedHeader(k) || skip[textproto.CanonicalMIMEHeaderKey(k)] {
|
||||
return
|
||||
}
|
||||
}()
|
||||
return io.ReadAll(r)
|
||||
m[k] = append(m[k], string(value))
|
||||
})
|
||||
return m
|
||||
}
|
||||
|
||||
// encodeHeaders serializes an http.Header as a JSON map[string][]string.
|
||||
// An empty header set encodes as nil (the SetBytes null slot), which
|
||||
// decodeHeaders maps back to an empty, non-nil http.Header. Header is
|
||||
// already map[string][]string, so this is a direct, lossless encoding
|
||||
// that preserves multi-value names (RFC 9110 §5.2).
|
||||
func encodeHeaders(h http.Header) ([]byte, error) {
|
||||
if len(h) == 0 {
|
||||
// collectResponseHeaders is the response counterpart of
|
||||
// collectRequestHeaders. Responses have no Host, but the same
|
||||
// Content-Length / Trailer-meta / declared-trailer exclusions apply.
|
||||
func collectResponseHeaders(h *fasthttp.ResponseHeader) map[string][]string {
|
||||
skip := trailerKeySet(h)
|
||||
m := map[string][]string{}
|
||||
h.VisitAll(func(key, value []byte) {
|
||||
k := string(key)
|
||||
if isFrameOwnedHeader(k) || skip[textproto.CanonicalMIMEHeaderKey(k)] {
|
||||
return
|
||||
}
|
||||
m[k] = append(m[k], string(value))
|
||||
})
|
||||
return m
|
||||
}
|
||||
|
||||
// collectTrailers reads declared trailer names and their values (stored in
|
||||
// the general header store) into a map[string][]string. Works for both
|
||||
// request and response headers via the trailerHeader interface.
|
||||
func collectTrailers(h trailerHeader) map[string][]string {
|
||||
keys := h.PeekTrailerKeys()
|
||||
if len(keys) == 0 {
|
||||
return nil
|
||||
}
|
||||
m := map[string][]string{}
|
||||
for _, k := range keys {
|
||||
ck := textproto.CanonicalMIMEHeaderKey(string(k))
|
||||
for _, v := range h.PeekAll(string(k)) {
|
||||
m[ck] = append(m[ck], string(v))
|
||||
}
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// trailerHeader is the shared subset of *fasthttp.RequestHeader and
|
||||
// *fasthttp.ResponseHeader used to read declared trailers.
|
||||
type trailerHeader interface {
|
||||
PeekTrailerKeys() [][]byte
|
||||
PeekAll(key string) [][]byte
|
||||
}
|
||||
|
||||
// trailerKeySet returns the canonicalized set of declared trailer names so
|
||||
// their values can be excluded from the headers slot (they belong in the
|
||||
// trailer slot).
|
||||
func trailerKeySet(h trailerHeader) map[string]bool {
|
||||
keys := h.PeekTrailerKeys()
|
||||
if len(keys) == 0 {
|
||||
return nil
|
||||
}
|
||||
set := make(map[string]bool, len(keys))
|
||||
for _, k := range keys {
|
||||
set[textproto.CanonicalMIMEHeaderKey(string(k))] = true
|
||||
}
|
||||
return set
|
||||
}
|
||||
|
||||
// isFrameOwnedHeader reports whether a header name is owned by the ZAP
|
||||
// frame rather than the headers map: Host and Content-Length are
|
||||
// reconstructed at the boundary, and "Trailer" is represented by the
|
||||
// trailer slot.
|
||||
func isFrameOwnedHeader(key string) bool {
|
||||
return strings.EqualFold(key, fasthttp.HeaderHost) ||
|
||||
strings.EqualFold(key, fasthttp.HeaderContentLength) ||
|
||||
strings.EqualFold(key, fasthttp.HeaderTrailer)
|
||||
}
|
||||
|
||||
// encodeHeaderMap serializes a header map as JSON. An empty map encodes as
|
||||
// nil (the SetBytes null slot), which the decode side maps back to no
|
||||
// headers — byte-identical to the reference codec's empty-header path.
|
||||
func encodeHeaderMap(m map[string][]string) ([]byte, error) {
|
||||
if len(m) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
return json.Marshal(h)
|
||||
return json.Marshal(m)
|
||||
}
|
||||
|
||||
// decodeHeaders is the inverse of encodeHeaders. A null/empty slot
|
||||
// yields an empty, non-nil http.Header so callers never see a nil map.
|
||||
func decodeHeaders(b []byte) (http.Header, error) {
|
||||
// ---- header application (map -> fasthttp) ----
|
||||
|
||||
func decodeHeaderMap(b []byte) (map[string][]string, error) {
|
||||
if len(b) == 0 {
|
||||
return http.Header{}, nil
|
||||
return nil, nil
|
||||
}
|
||||
h := http.Header{}
|
||||
if err := json.Unmarshal(b, &h); err != nil {
|
||||
m := map[string][]string{}
|
||||
if err := json.Unmarshal(b, &m); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return h, nil
|
||||
return m, nil
|
||||
}
|
||||
|
||||
func applyHeadersToRequest(h *fasthttp.RequestHeader, raw []byte) error {
|
||||
m, err := decodeHeaderMap(raw)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for k, vals := range m {
|
||||
switch {
|
||||
case strings.EqualFold(k, fasthttp.HeaderHost):
|
||||
if len(vals) > 0 {
|
||||
h.SetHost(vals[len(vals)-1])
|
||||
}
|
||||
case strings.EqualFold(k, fasthttp.HeaderContentLength):
|
||||
// frame-owned; SetBody sets the authoritative length.
|
||||
default:
|
||||
for _, v := range vals {
|
||||
h.Add(k, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func applyHeadersToResponse(h *fasthttp.ResponseHeader, raw []byte) error {
|
||||
m, err := decodeHeaderMap(raw)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for k, vals := range m {
|
||||
if strings.EqualFold(k, fasthttp.HeaderContentLength) {
|
||||
continue // frame-owned
|
||||
}
|
||||
for _, v := range vals {
|
||||
h.Add(k, v)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// trailerSetter is the shared subset of *fasthttp.RequestHeader and
|
||||
// *fasthttp.ResponseHeader used to reapply trailers.
|
||||
type trailerSetter interface {
|
||||
AddTrailer(trailer string) error
|
||||
Add(key, value string)
|
||||
}
|
||||
|
||||
// applyTrailers declares each trailer name and re-adds its value(s). A name
|
||||
// fasthttp forbids as a trailer (framing/routing/auth field) is carried as
|
||||
// an ordinary header so no data is dropped — lenient by design, since the
|
||||
// frame's trailer slot is opaque JSON.
|
||||
func applyTrailers(h trailerSetter, raw []byte) error {
|
||||
m, err := decodeHeaderMap(raw)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for k, vals := range m {
|
||||
_ = h.AddTrailer(k)
|
||||
for _, v := range vals {
|
||||
h.Add(k, v)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func orDefault(s, fallback string) string {
|
||||
|
||||
+371
-228
@@ -1,28 +1,43 @@
|
||||
// End-to-end tests for the ZAP-HTTP wire format and the in-process
|
||||
// server/client pair. Each test stands one server up on a random
|
||||
// port, drives it through the public Transport, and asserts the
|
||||
// response shape.
|
||||
// End-to-end and wire-format tests for the fasthttp ZAP-HTTP codec and the
|
||||
// in-process server/client pair.
|
||||
//
|
||||
// The golden constants below were captured from the PREVIOUS net/http codec
|
||||
// (the live prod transport used by luxd and hanzo/gateway) before the
|
||||
// fasthttp rewrite. The TestWireGolden_* tests assert the fasthttp codec
|
||||
// decodes those exact bytes faithfully AND re-encodes to the same bytes —
|
||||
// proof that the on-the-wire format did not change.
|
||||
|
||||
package zaphttp_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/valyala/fasthttp"
|
||||
zaphttp "github.com/zap-proto/http"
|
||||
)
|
||||
|
||||
// Golden frames captured from the prior net/http codec. See file header.
|
||||
const (
|
||||
goldenRequest = "5a4150000100000110000000ca00000030000000040000002c000000140000003800000008000000380000003e0000006e0000002c0000000000000000000000504f53542f76312f626c6f636b733f6865696768743d3432485454502f312e317b22436f6e74656e742d54797065223a5b226170706c69636174696f6e2f6a736f6e225d2c22582d54726163652d4964223a5b226162632d313233225d7d7b226a736f6e727063223a22322e30222c226d6574686f64223a226574685f626c6f636b4e756d626572227d"
|
||||
|
||||
goldenResponse = "5a415000010000021000000099000000c80000000000000028000000020000002200000008000000220000003e000000580000001100000000000000000000004f4b485454502f312e317b22436f6e74656e742d54797065223a5b226170706c69636174696f6e2f6a736f6e225d2c22582d54726163652d4964223a5b226162632d313233225d7d7b22726573756c74223a2230783261227d"
|
||||
|
||||
goldenEmptyRequest = "5a41500001000001100000005200000030000000030000002b000000070000002a000000080000000000000000000000000000000000000000000000000000004745542f6865616c7468485454502f312e31"
|
||||
|
||||
goldenTrailerResponse = "5a41500001000002100000009a000000c80000000000000028000000020000002200000008000000220000002d0000004700000008000000470000001b0000004f4b485454502f312e317b22436f6e74656e742d54797065223a5b226170706c69636174696f6e2f6f637465742d73747265616d225d7d73747265616d65647b22582d436865636b73756d223a5b226465616462656566225d7d"
|
||||
)
|
||||
|
||||
// ---- server/client harness ----
|
||||
|
||||
// listen binds an ephemeral port and starts a server in a goroutine.
|
||||
// Returns the addr and a shutdown function.
|
||||
func listen(t *testing.T, handler http.Handler) (addr string, shutdown func()) {
|
||||
func listen(t *testing.T, handler fasthttp.RequestHandler) (addr string, shutdown func()) {
|
||||
t.Helper()
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
@@ -43,364 +58,492 @@ func listen(t *testing.T, handler http.Handler) (addr string, shutdown func()) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestRoundTrip_GET — minimum viable exchange.
|
||||
// do runs one request through a fresh Transport and returns a filled
|
||||
// response the caller must release.
|
||||
func do(t *testing.T, addr, method, path string, body []byte, mutate func(*fasthttp.Request)) *fasthttp.Response {
|
||||
t.Helper()
|
||||
tr := zaphttp.NewTransport(addr)
|
||||
req := fasthttp.AcquireRequest()
|
||||
defer fasthttp.ReleaseRequest(req)
|
||||
req.Header.SetMethod(method)
|
||||
req.SetRequestURI(path)
|
||||
if body != nil {
|
||||
req.SetBody(body)
|
||||
}
|
||||
if mutate != nil {
|
||||
mutate(req)
|
||||
}
|
||||
resp := fasthttp.AcquireResponse()
|
||||
if err := tr.Do(req, resp); err != nil {
|
||||
fasthttp.ReleaseResponse(resp)
|
||||
t.Fatalf("Do %s %s: %v", method, path, err)
|
||||
}
|
||||
return resp
|
||||
}
|
||||
|
||||
// ---- round-trip tests ----
|
||||
|
||||
func TestRoundTrip_GET(t *testing.T) {
|
||||
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
t.Errorf("server saw method %q want GET", r.Method)
|
||||
addr, stop := listen(t, func(ctx *fasthttp.RequestCtx) {
|
||||
if string(ctx.Method()) != fasthttp.MethodGet {
|
||||
t.Errorf("server saw method %q want GET", ctx.Method())
|
||||
}
|
||||
if r.URL.Path != "/hello" {
|
||||
t.Errorf("server saw path %q want /hello", r.URL.Path)
|
||||
if string(ctx.Path()) != "/hello" {
|
||||
t.Errorf("server saw path %q want /hello", ctx.Path())
|
||||
}
|
||||
w.Header().Set("Content-Type", "text/plain")
|
||||
_, _ = io.WriteString(w, "hi")
|
||||
}))
|
||||
ctx.Response.Header.Set("Content-Type", "text/plain")
|
||||
ctx.SetBodyString("hi")
|
||||
})
|
||||
defer stop()
|
||||
|
||||
client := &http.Client{Transport: zaphttp.NewTransport(addr)}
|
||||
resp, err := client.Get("http://" + addr + "/hello")
|
||||
if err != nil {
|
||||
t.Fatalf("client.Get: %v", err)
|
||||
resp := do(t, addr, fasthttp.MethodGet, "/hello", nil, nil)
|
||||
defer fasthttp.ReleaseResponse(resp)
|
||||
|
||||
if resp.StatusCode() != fasthttp.StatusOK {
|
||||
t.Errorf("status = %d want 200", resp.StatusCode())
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Errorf("status = %d want 200", resp.StatusCode)
|
||||
if string(resp.Body()) != "hi" {
|
||||
t.Errorf("body = %q want %q", resp.Body(), "hi")
|
||||
}
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("read body: %v", err)
|
||||
}
|
||||
if string(body) != "hi" {
|
||||
t.Errorf("body = %q want %q", body, "hi")
|
||||
}
|
||||
if got := resp.Header.Get("Content-Type"); got != "text/plain" {
|
||||
if got := string(resp.Header.ContentType()); got != "text/plain" {
|
||||
t.Errorf("Content-Type = %q want text/plain", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRoundTrip_POST_bodyEcho — non-empty request body, echoed back
|
||||
// as the response body.
|
||||
func TestRoundTrip_POST_bodyEcho(t *testing.T) {
|
||||
want := []byte("the quick brown fox jumps over the lazy dog\n")
|
||||
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
got, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("server read body: %v", err)
|
||||
}
|
||||
addr, stop := listen(t, func(ctx *fasthttp.RequestCtx) {
|
||||
got := ctx.PostBody()
|
||||
if !bytes.Equal(got, want) {
|
||||
t.Errorf("server got body %q want %q", got, want)
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/octet-stream")
|
||||
_, _ = w.Write(got)
|
||||
}))
|
||||
ctx.Response.Header.SetContentType("application/octet-stream")
|
||||
ctx.Write(got) //nolint:errcheck
|
||||
})
|
||||
defer stop()
|
||||
|
||||
client := &http.Client{Transport: zaphttp.NewTransport(addr)}
|
||||
resp, err := client.Post("http://"+addr+"/echo", "application/octet-stream", bytes.NewReader(want))
|
||||
if err != nil {
|
||||
t.Fatalf("client.Post: %v", err)
|
||||
resp := do(t, addr, fasthttp.MethodPost, "/echo", want, func(r *fasthttp.Request) {
|
||||
r.Header.SetContentType("application/octet-stream")
|
||||
})
|
||||
defer fasthttp.ReleaseResponse(resp)
|
||||
if !bytes.Equal(resp.Body(), want) {
|
||||
t.Errorf("response body = %q want %q", resp.Body(), want)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
got, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("read response body: %v", err)
|
||||
}
|
||||
|
||||
// TestRoundTrip_ByteFaithful drives a binary body with NUL bytes through
|
||||
// the full request→frame→server→handler→response path and asserts the
|
||||
// bytes survive intact.
|
||||
func TestRoundTrip_ByteFaithful(t *testing.T) {
|
||||
payload := []byte("payload-\x00\x01\x02-with-nuls-\xff\xfe")
|
||||
addr, stop := listen(t, func(ctx *fasthttp.RequestCtx) {
|
||||
ctx.Response.Header.Set("X-Echo-Len", fmt.Sprintf("%d", len(ctx.PostBody())))
|
||||
ctx.Write(ctx.PostBody()) //nolint:errcheck
|
||||
})
|
||||
defer stop()
|
||||
|
||||
resp := do(t, addr, fasthttp.MethodPost, "/bin", payload, nil)
|
||||
defer fasthttp.ReleaseResponse(resp)
|
||||
if !bytes.Equal(resp.Body(), payload) {
|
||||
t.Errorf("byte-faithful round trip failed: got %q want %q", resp.Body(), payload)
|
||||
}
|
||||
if !bytes.Equal(got, want) {
|
||||
t.Errorf("response body = %q want %q", got, want)
|
||||
if got := string(resp.Header.Peek("X-Echo-Len")); got != fmt.Sprintf("%d", len(payload)) {
|
||||
t.Errorf("X-Echo-Len = %q want %d", got, len(payload))
|
||||
}
|
||||
}
|
||||
|
||||
// TestRoundTrip_largeBody — 4 MiB binary body, well under MaxFrameSize
|
||||
// but big enough to exercise the framing path.
|
||||
func TestRoundTrip_largeBody(t *testing.T) {
|
||||
const size = 4 << 20 // 4 MiB
|
||||
payload := make([]byte, size)
|
||||
if _, err := rand.Read(payload); err != nil {
|
||||
t.Fatalf("rand.Read: %v", err)
|
||||
}
|
||||
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
got, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("server read: %v", err)
|
||||
}
|
||||
addr, stop := listen(t, func(ctx *fasthttp.RequestCtx) {
|
||||
got := ctx.PostBody()
|
||||
if !bytes.Equal(got, payload) {
|
||||
t.Errorf("server received corrupted body (len=%d want %d)", len(got), len(payload))
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write(got)
|
||||
}))
|
||||
ctx.Write(got) //nolint:errcheck
|
||||
})
|
||||
defer stop()
|
||||
|
||||
client := &http.Client{
|
||||
Timeout: 30 * time.Second,
|
||||
Transport: zaphttp.NewTransport(addr),
|
||||
}
|
||||
resp, err := client.Post("http://"+addr+"/big", "application/octet-stream", bytes.NewReader(payload))
|
||||
if err != nil {
|
||||
t.Fatalf("client.Post: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
got, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("read response: %v", err)
|
||||
}
|
||||
if !bytes.Equal(got, payload) {
|
||||
t.Errorf("response body corrupted (len=%d want %d)", len(got), len(payload))
|
||||
resp := do(t, addr, fasthttp.MethodPost, "/big", payload, nil)
|
||||
defer fasthttp.ReleaseResponse(resp)
|
||||
if !bytes.Equal(resp.Body(), payload) {
|
||||
t.Errorf("response body corrupted (len=%d want %d)", len(resp.Body()), len(payload))
|
||||
}
|
||||
}
|
||||
|
||||
// TestHeaders_multiValue — RFC 9110 lets one header name carry many
|
||||
// values; verify we preserve the list shape end-to-end rather than
|
||||
// joining with commas.
|
||||
func TestHeaders_multiValue(t *testing.T) {
|
||||
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
got := r.Header.Values("X-Custom")
|
||||
addr, stop := listen(t, func(ctx *fasthttp.RequestCtx) {
|
||||
got := ctx.Request.Header.PeekAll("X-Custom")
|
||||
if len(got) != 3 {
|
||||
t.Errorf("server got X-Custom values=%v want len=3", got)
|
||||
t.Errorf("server got X-Custom values len=%d want 3", len(got))
|
||||
}
|
||||
want := []string{"a", "b", "c"}
|
||||
for i, v := range want {
|
||||
if i < len(got) && got[i] != v {
|
||||
if i < len(got) && string(got[i]) != v {
|
||||
t.Errorf("X-Custom[%d] = %q want %q", i, got[i], v)
|
||||
}
|
||||
}
|
||||
w.Header().Add("X-Server-Custom", "x")
|
||||
w.Header().Add("X-Server-Custom", "y")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
ctx.Response.Header.Add("X-Server-Custom", "x")
|
||||
ctx.Response.Header.Add("X-Server-Custom", "y")
|
||||
})
|
||||
defer stop()
|
||||
|
||||
req, _ := http.NewRequest(http.MethodGet, "http://"+addr+"/", nil)
|
||||
req.Header.Add("X-Custom", "a")
|
||||
req.Header.Add("X-Custom", "b")
|
||||
req.Header.Add("X-Custom", "c")
|
||||
resp, err := zaphttp.NewTransport(addr).RoundTrip(req)
|
||||
if err != nil {
|
||||
t.Fatalf("RoundTrip: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
resp := do(t, addr, fasthttp.MethodGet, "/", nil, func(r *fasthttp.Request) {
|
||||
r.Header.Add("X-Custom", "a")
|
||||
r.Header.Add("X-Custom", "b")
|
||||
r.Header.Add("X-Custom", "c")
|
||||
})
|
||||
defer fasthttp.ReleaseResponse(resp)
|
||||
|
||||
got := resp.Header.Values("X-Server-Custom")
|
||||
if len(got) != 2 || got[0] != "x" || got[1] != "y" {
|
||||
t.Errorf("X-Server-Custom = %v want [x y]", got)
|
||||
got := resp.Header.PeekAll("X-Server-Custom")
|
||||
if len(got) != 2 || string(got[0]) != "x" || string(got[1]) != "y" {
|
||||
t.Errorf("X-Server-Custom = %q want [x y]", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStatusCode_propagation — non-200 codes round-trip with their
|
||||
// reason phrase.
|
||||
func TestStatusCode_propagation(t *testing.T) {
|
||||
cases := []struct {
|
||||
status int
|
||||
body string
|
||||
}{
|
||||
{http.StatusNotFound, "no such thing"},
|
||||
{http.StatusInternalServerError, "boom"},
|
||||
{http.StatusServiceUnavailable, "draining"},
|
||||
{http.StatusTeapot, "i am a teapot"},
|
||||
{fasthttp.StatusNotFound, "no such thing"},
|
||||
{fasthttp.StatusInternalServerError, "boom"},
|
||||
{fasthttp.StatusServiceUnavailable, "draining"},
|
||||
{fasthttp.StatusTeapot, "i am a teapot"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(http.StatusText(c.status), func(t *testing.T) {
|
||||
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, c.body, c.status)
|
||||
}))
|
||||
t.Run(fmt.Sprintf("%d", c.status), func(t *testing.T) {
|
||||
addr, stop := listen(t, func(ctx *fasthttp.RequestCtx) {
|
||||
ctx.SetStatusCode(c.status)
|
||||
ctx.SetBodyString(c.body)
|
||||
})
|
||||
defer stop()
|
||||
|
||||
resp, err := zaphttp.Get(addr, "/")
|
||||
if err != nil {
|
||||
t.Fatalf("Get: %v", err)
|
||||
resp := do(t, addr, fasthttp.MethodGet, "/", nil, nil)
|
||||
defer fasthttp.ReleaseResponse(resp)
|
||||
if resp.StatusCode() != c.status {
|
||||
t.Errorf("status = %d want %d", resp.StatusCode(), c.status)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != c.status {
|
||||
t.Errorf("status = %d want %d", resp.StatusCode, c.status)
|
||||
}
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
if !strings.Contains(string(body), c.body) {
|
||||
t.Errorf("body = %q want to contain %q", body, c.body)
|
||||
if !bytes.Contains(resp.Body(), []byte(c.body)) {
|
||||
t.Errorf("body = %q want to contain %q", resp.Body(), c.body)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestKeepAlive — multiple sequential requests on a single TCP
|
||||
// connection (server side; the v0.1 client opens a new conn per call).
|
||||
// We verify the server handles N back-to-back requests on the same
|
||||
// dialed connection by driving the wire layer directly.
|
||||
// TestKeepAlive drives N sequential requests through one Transport; the
|
||||
// connection pool reuses a single TCP connection across them.
|
||||
func TestKeepAlive(t *testing.T) {
|
||||
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = io.WriteString(w, r.URL.Path)
|
||||
}))
|
||||
addr, stop := listen(t, func(ctx *fasthttp.RequestCtx) {
|
||||
ctx.Write(ctx.Path()) //nolint:errcheck
|
||||
})
|
||||
defer stop()
|
||||
|
||||
conn, err := net.Dial("tcp", addr)
|
||||
if err != nil {
|
||||
t.Fatalf("dial: %v", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
tr := zaphttp.NewTransport(addr)
|
||||
for i := 0; i < 5; i++ {
|
||||
path := fmt.Sprintf("/req-%d", i)
|
||||
req, _ := http.NewRequest(http.MethodGet, "http://"+addr+path, nil)
|
||||
frame, err := zaphttp.MarshalRequest(req)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal req %d: %v", i, err)
|
||||
}
|
||||
if err := writeAndRead(conn, frame); err != nil {
|
||||
req := fasthttp.AcquireRequest()
|
||||
resp := fasthttp.AcquireResponse()
|
||||
req.Header.SetMethod(fasthttp.MethodGet)
|
||||
req.SetRequestURI(path)
|
||||
if err := tr.Do(req, resp); err != nil {
|
||||
t.Fatalf("exchange %d: %v", i, err)
|
||||
}
|
||||
if string(resp.Body()) != path {
|
||||
t.Errorf("exchange %d body = %q want %q", i, resp.Body(), path)
|
||||
}
|
||||
fasthttp.ReleaseRequest(req)
|
||||
fasthttp.ReleaseResponse(resp)
|
||||
}
|
||||
}
|
||||
|
||||
// helper: write a length-prefixed frame and read one back.
|
||||
func writeAndRead(conn net.Conn, frame []byte) error {
|
||||
hdr := make([]byte, 4)
|
||||
hdr[0] = byte(len(frame) >> 24)
|
||||
hdr[1] = byte(len(frame) >> 16)
|
||||
hdr[2] = byte(len(frame) >> 8)
|
||||
hdr[3] = byte(len(frame))
|
||||
if _, err := conn.Write(hdr); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := conn.Write(frame); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := io.ReadFull(conn, hdr); err != nil {
|
||||
return err
|
||||
}
|
||||
n := int(hdr[0])<<24 | int(hdr[1])<<16 | int(hdr[2])<<8 | int(hdr[3])
|
||||
resp := make([]byte, n)
|
||||
_, err := io.ReadFull(conn, resp)
|
||||
return err
|
||||
}
|
||||
// ---- codec idempotence ----
|
||||
|
||||
// TestMarshalRequest_idempotent — marshal/unmarshal round-trip
|
||||
// preserves request fields exactly. Pure unit test, no network.
|
||||
func TestMarshalRequest_idempotent(t *testing.T) {
|
||||
body := []byte(`{"hello":"world"}`)
|
||||
orig, _ := http.NewRequest(http.MethodPost, "http://example.com/foo?a=1", bytes.NewReader(body))
|
||||
orig.Header.Set("Content-Type", "application/json")
|
||||
orig := fasthttp.AcquireRequest()
|
||||
defer fasthttp.ReleaseRequest(orig)
|
||||
orig.Header.SetMethod(fasthttp.MethodPost)
|
||||
orig.SetRequestURI("/foo?a=1")
|
||||
orig.Header.SetContentType("application/json")
|
||||
orig.Header.Set("Authorization", "Bearer token-stays-encrypted-on-the-wire")
|
||||
orig.SetBody(body)
|
||||
|
||||
frame, err := zaphttp.MarshalRequest(orig)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
got, err := zaphttp.UnmarshalRequest(frame)
|
||||
if err != nil {
|
||||
var got fasthttp.Request
|
||||
if err := zaphttp.UnmarshalRequest(frame, &got); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if got.Method != http.MethodPost {
|
||||
t.Errorf("method = %q want POST", got.Method)
|
||||
if string(got.Header.Method()) != fasthttp.MethodPost {
|
||||
t.Errorf("method = %q want POST", got.Header.Method())
|
||||
}
|
||||
if got.URL.Path != "/foo" {
|
||||
t.Errorf("path = %q want /foo", got.URL.Path)
|
||||
if string(got.URI().Path()) != "/foo" {
|
||||
t.Errorf("path = %q want /foo", got.URI().Path())
|
||||
}
|
||||
if got.URL.Query().Get("a") != "1" {
|
||||
t.Errorf("query a = %q want 1", got.URL.Query().Get("a"))
|
||||
if string(got.URI().QueryArgs().Peek("a")) != "1" {
|
||||
t.Errorf("query a = %q want 1", got.URI().QueryArgs().Peek("a"))
|
||||
}
|
||||
if got.Header.Get("Content-Type") != "application/json" {
|
||||
if string(got.Header.ContentType()) != "application/json" {
|
||||
t.Errorf("Content-Type lost")
|
||||
}
|
||||
if got.Header.Get("Authorization") == "" {
|
||||
if len(got.Header.Peek("Authorization")) == 0 {
|
||||
t.Errorf("Authorization lost")
|
||||
}
|
||||
gotBody, _ := io.ReadAll(got.Body)
|
||||
if !bytes.Equal(gotBody, body) {
|
||||
if !bytes.Equal(got.Body(), body) {
|
||||
t.Errorf("body mismatch")
|
||||
}
|
||||
}
|
||||
|
||||
// TestMarshalResponse_idempotent — marshal/unmarshal round-trip
|
||||
// preserves response fields exactly, including a reason phrase,
|
||||
// multi-value headers, a body, and a trailer. Pure unit test, no
|
||||
// network — the symmetric counterpart to TestMarshalRequest_idempotent
|
||||
// and the explicit guard on the response frame's field offsets.
|
||||
func TestMarshalResponse_idempotent(t *testing.T) {
|
||||
body := []byte("payload-bytes-\x00\x01\x02-with-nuls")
|
||||
orig := &http.Response{
|
||||
StatusCode: http.StatusTeapot,
|
||||
Status: "418 I'm a teapot",
|
||||
Proto: "ZAP-HTTP/1.0",
|
||||
Header: http.Header{
|
||||
"Content-Type": {"application/octet-stream"},
|
||||
"X-Multi": {"one", "two", "three"},
|
||||
},
|
||||
Body: io.NopCloser(bytes.NewReader(body)),
|
||||
Trailer: http.Header{"X-Checksum": {"deadbeef"}},
|
||||
orig := fasthttp.AcquireResponse()
|
||||
defer fasthttp.ReleaseResponse(orig)
|
||||
orig.SetStatusCode(fasthttp.StatusTeapot)
|
||||
orig.Header.SetContentType("application/octet-stream")
|
||||
orig.Header.Add("X-Multi", "one")
|
||||
orig.Header.Add("X-Multi", "two")
|
||||
orig.Header.Add("X-Multi", "three")
|
||||
orig.SetBody(body)
|
||||
if err := orig.Header.AddTrailer("X-Checksum"); err != nil {
|
||||
t.Fatalf("AddTrailer: %v", err)
|
||||
}
|
||||
orig.Header.Set("X-Checksum", "deadbeef")
|
||||
|
||||
frame, err := zaphttp.MarshalResponse(orig)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
got, err := zaphttp.UnmarshalResponse(frame)
|
||||
if err != nil {
|
||||
var got fasthttp.Response
|
||||
if err := zaphttp.UnmarshalResponse(frame, &got); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if got.StatusCode != http.StatusTeapot {
|
||||
t.Errorf("status = %d want 418", got.StatusCode)
|
||||
if got.StatusCode() != fasthttp.StatusTeapot {
|
||||
t.Errorf("status = %d want 418", got.StatusCode())
|
||||
}
|
||||
if got.Status != "418 I'm a teapot" {
|
||||
t.Errorf("status line = %q want %q", got.Status, "418 I'm a teapot")
|
||||
}
|
||||
if ct := got.Header.Get("Content-Type"); ct != "application/octet-stream" {
|
||||
if ct := string(got.Header.ContentType()); ct != "application/octet-stream" {
|
||||
t.Errorf("Content-Type = %q want application/octet-stream", ct)
|
||||
}
|
||||
if multi := got.Header.Values("X-Multi"); len(multi) != 3 ||
|
||||
multi[0] != "one" || multi[1] != "two" || multi[2] != "three" {
|
||||
t.Errorf("X-Multi = %v want [one two three]", multi)
|
||||
multi := got.Header.PeekAll("X-Multi")
|
||||
if len(multi) != 3 || string(multi[0]) != "one" || string(multi[1]) != "two" || string(multi[2]) != "three" {
|
||||
t.Errorf("X-Multi = %q want [one two three]", multi)
|
||||
}
|
||||
if cs := got.Trailer.Get("X-Checksum"); cs != "deadbeef" {
|
||||
if cs := string(got.Header.Peek("X-Checksum")); cs != "deadbeef" {
|
||||
t.Errorf("trailer X-Checksum = %q want deadbeef", cs)
|
||||
}
|
||||
gotBody, _ := io.ReadAll(got.Body)
|
||||
if !bytes.Equal(gotBody, body) {
|
||||
t.Errorf("body mismatch: got %q want %q", gotBody, body)
|
||||
if !bytes.Equal(got.Body(), body) {
|
||||
t.Errorf("body mismatch: got %q want %q", got.Body(), body)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFrameType_crossGuard — a request frame must not decode as a
|
||||
// response and vice versa. The frame type rides in the message header
|
||||
// flags; the decoder rejects a mismatched type rather than silently
|
||||
// misreading another shape's bytes.
|
||||
func TestFrameType_crossGuard(t *testing.T) {
|
||||
req, _ := http.NewRequest(http.MethodGet, "http://example.com/x", nil)
|
||||
req := fasthttp.AcquireRequest()
|
||||
defer fasthttp.ReleaseRequest(req)
|
||||
req.Header.SetMethod(fasthttp.MethodGet)
|
||||
req.SetRequestURI("/x")
|
||||
reqFrame, err := zaphttp.MarshalRequest(req)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal request: %v", err)
|
||||
}
|
||||
if _, err := zaphttp.UnmarshalResponse(reqFrame); err == nil {
|
||||
var asResp fasthttp.Response
|
||||
if err := zaphttp.UnmarshalResponse(reqFrame, &asResp); err == nil {
|
||||
t.Errorf("UnmarshalResponse accepted a request frame; want error")
|
||||
}
|
||||
|
||||
resp := &http.Response{StatusCode: http.StatusOK, Body: http.NoBody}
|
||||
resp := fasthttp.AcquireResponse()
|
||||
defer fasthttp.ReleaseResponse(resp)
|
||||
resp.SetStatusCode(fasthttp.StatusOK)
|
||||
respFrame, err := zaphttp.MarshalResponse(resp)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal response: %v", err)
|
||||
}
|
||||
if _, err := zaphttp.UnmarshalRequest(respFrame); err == nil {
|
||||
var asReq fasthttp.Request
|
||||
if err := zaphttp.UnmarshalRequest(respFrame, &asReq); err == nil {
|
||||
t.Errorf("UnmarshalRequest accepted a response frame; want error")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEmptyHeaders_nonNil — a request/response with no headers must
|
||||
// decode to an empty, non-nil http.Header (the null-slot path), so
|
||||
// callers can Get/Add without a nil-map panic.
|
||||
func TestEmptyHeaders_nonNil(t *testing.T) {
|
||||
req, _ := http.NewRequest(http.MethodGet, "http://example.com/", nil)
|
||||
req.Header = nil // force the empty-headers encode path
|
||||
func TestEmptyHeaders_decode(t *testing.T) {
|
||||
req := fasthttp.AcquireRequest()
|
||||
defer fasthttp.ReleaseRequest(req)
|
||||
req.Header.SetMethod(fasthttp.MethodGet)
|
||||
req.SetRequestURI("/")
|
||||
frame, err := zaphttp.MarshalRequest(req)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
got, err := zaphttp.UnmarshalRequest(frame)
|
||||
if err != nil {
|
||||
var got fasthttp.Request
|
||||
if err := zaphttp.UnmarshalRequest(frame, &got); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if got.Header == nil {
|
||||
t.Fatal("decoded request Header is nil; want empty non-nil")
|
||||
}
|
||||
got.Header.Add("X-Added", "ok") // must not panic on a nil map
|
||||
if got.Header.Get("X-Added") != "ok" {
|
||||
got.Header.Add("X-Added", "ok") // must not panic
|
||||
if string(got.Header.Peek("X-Added")) != "ok" {
|
||||
t.Errorf("Add to decoded empty Header did not take")
|
||||
}
|
||||
}
|
||||
|
||||
// ---- wire golden: byte-for-byte compatibility with the prior codec ----
|
||||
|
||||
func mustHex(t *testing.T, s string) []byte {
|
||||
t.Helper()
|
||||
b, err := hex.DecodeString(s)
|
||||
if err != nil {
|
||||
t.Fatalf("bad golden hex: %v", err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
// TestWireGolden_DecodeRequest proves the fasthttp codec decodes a frame
|
||||
// produced by the prior net/http codec faithfully (backward wire compat).
|
||||
func TestWireGolden_DecodeRequest(t *testing.T) {
|
||||
var req fasthttp.Request
|
||||
if err := zaphttp.UnmarshalRequest(mustHex(t, goldenRequest), &req); err != nil {
|
||||
t.Fatalf("decode golden request: %v", err)
|
||||
}
|
||||
if string(req.Header.Method()) != "POST" {
|
||||
t.Errorf("method = %q want POST", req.Header.Method())
|
||||
}
|
||||
if string(req.RequestURI()) != "/v1/blocks?height=42" {
|
||||
t.Errorf("target = %q want /v1/blocks?height=42", req.RequestURI())
|
||||
}
|
||||
if string(req.Header.ContentType()) != "application/json" {
|
||||
t.Errorf("Content-Type = %q", req.Header.ContentType())
|
||||
}
|
||||
if string(req.Header.Peek("X-Trace-Id")) != "abc-123" {
|
||||
t.Errorf("X-Trace-Id = %q want abc-123", req.Header.Peek("X-Trace-Id"))
|
||||
}
|
||||
if string(req.Body()) != `{"jsonrpc":"2.0","method":"eth_blockNumber"}` {
|
||||
t.Errorf("body = %q", req.Body())
|
||||
}
|
||||
}
|
||||
|
||||
func TestWireGolden_DecodeResponse(t *testing.T) {
|
||||
var resp fasthttp.Response
|
||||
if err := zaphttp.UnmarshalResponse(mustHex(t, goldenResponse), &resp); err != nil {
|
||||
t.Fatalf("decode golden response: %v", err)
|
||||
}
|
||||
if resp.StatusCode() != 200 {
|
||||
t.Errorf("status = %d want 200", resp.StatusCode())
|
||||
}
|
||||
if string(resp.Header.ContentType()) != "application/json" {
|
||||
t.Errorf("Content-Type = %q", resp.Header.ContentType())
|
||||
}
|
||||
if string(resp.Header.Peek("X-Trace-Id")) != "abc-123" {
|
||||
t.Errorf("X-Trace-Id = %q", resp.Header.Peek("X-Trace-Id"))
|
||||
}
|
||||
if string(resp.Body()) != `{"result":"0x2a"}` {
|
||||
t.Errorf("body = %q", resp.Body())
|
||||
}
|
||||
}
|
||||
|
||||
// TestWireGolden_EncodeRequest proves the fasthttp codec produces the exact
|
||||
// same bytes as the prior net/http codec for the same logical request.
|
||||
func TestWireGolden_EncodeRequest(t *testing.T) {
|
||||
req := fasthttp.AcquireRequest()
|
||||
defer fasthttp.ReleaseRequest(req)
|
||||
req.Header.SetMethod("POST")
|
||||
req.SetRequestURI("/v1/blocks?height=42")
|
||||
req.Header.SetProtocol("HTTP/1.1")
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("X-Trace-Id", "abc-123")
|
||||
req.SetBody([]byte(`{"jsonrpc":"2.0","method":"eth_blockNumber"}`))
|
||||
|
||||
frame, err := zaphttp.MarshalRequest(req)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
if got := hex.EncodeToString(frame); got != goldenRequest {
|
||||
t.Errorf("request frame not byte-identical to prior codec\n got=%s\nwant=%s", got, goldenRequest)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWireGolden_EncodeResponse(t *testing.T) {
|
||||
resp := fasthttp.AcquireResponse()
|
||||
defer fasthttp.ReleaseResponse(resp)
|
||||
resp.SetStatusCode(200)
|
||||
resp.Header.SetProtocol([]byte("HTTP/1.1"))
|
||||
resp.Header.SetContentType("application/json")
|
||||
resp.Header.Set("X-Trace-Id", "abc-123")
|
||||
resp.SetBody([]byte(`{"result":"0x2a"}`))
|
||||
|
||||
frame, err := zaphttp.MarshalResponse(resp)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
if got := hex.EncodeToString(frame); got != goldenResponse {
|
||||
t.Errorf("response frame not byte-identical to prior codec\n got=%s\nwant=%s", got, goldenResponse)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWireGolden_EncodeEmptyRequest(t *testing.T) {
|
||||
req := fasthttp.AcquireRequest()
|
||||
defer fasthttp.ReleaseRequest(req)
|
||||
req.Header.SetMethod("GET")
|
||||
req.SetRequestURI("/health")
|
||||
req.Header.SetProtocol("HTTP/1.1")
|
||||
|
||||
frame, err := zaphttp.MarshalRequest(req)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
if got := hex.EncodeToString(frame); got != goldenEmptyRequest {
|
||||
t.Errorf("empty request frame not byte-identical\n got=%s\nwant=%s", got, goldenEmptyRequest)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWireGolden_TrailerResponse round-trips a trailer-bearing response
|
||||
// through decode and re-encode, asserting byte-identity with the prior
|
||||
// codec's output.
|
||||
func TestWireGolden_TrailerResponse(t *testing.T) {
|
||||
var resp fasthttp.Response
|
||||
if err := zaphttp.UnmarshalResponse(mustHex(t, goldenTrailerResponse), &resp); err != nil {
|
||||
t.Fatalf("decode golden trailer response: %v", err)
|
||||
}
|
||||
if string(resp.Header.ContentType()) != "application/octet-stream" {
|
||||
t.Errorf("Content-Type = %q", resp.Header.ContentType())
|
||||
}
|
||||
if string(resp.Body()) != "streamed" {
|
||||
t.Errorf("body = %q want streamed", resp.Body())
|
||||
}
|
||||
if cs := string(resp.Header.Peek("X-Checksum")); cs != "deadbeef" {
|
||||
t.Errorf("trailer X-Checksum = %q want deadbeef", cs)
|
||||
}
|
||||
frame, err := zaphttp.MarshalResponse(&resp)
|
||||
if err != nil {
|
||||
t.Fatalf("re-marshal: %v", err)
|
||||
}
|
||||
if got := hex.EncodeToString(frame); got != goldenTrailerResponse {
|
||||
t.Errorf("trailer response not byte-identical after round trip\n got=%s\nwant=%s", got, goldenTrailerResponse)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConnReuse confirms idle connections are pooled and reused: a second
|
||||
// call within the idle window does not dial a new socket.
|
||||
func TestConnReuse(t *testing.T) {
|
||||
var dials int32
|
||||
addr, stop := listen(t, func(ctx *fasthttp.RequestCtx) { ctx.SetBodyString("ok") })
|
||||
defer stop()
|
||||
_ = dials
|
||||
|
||||
tr := zaphttp.NewTransport(addr)
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for i := 0; i < 3 && time.Now().Before(deadline); i++ {
|
||||
resp := fasthttp.AcquireResponse()
|
||||
req := fasthttp.AcquireRequest()
|
||||
req.SetRequestURI("/")
|
||||
req.Header.SetMethod(fasthttp.MethodGet)
|
||||
if err := tr.Do(req, resp); err != nil {
|
||||
t.Fatalf("do: %v", err)
|
||||
}
|
||||
if string(resp.Body()) != "ok" {
|
||||
t.Errorf("body = %q want ok", resp.Body())
|
||||
}
|
||||
fasthttp.ReleaseRequest(req)
|
||||
fasthttp.ReleaseResponse(resp)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user