From 28185d97bdbc5d18e1e91c45eef30eaf7f86a430 Mon Sep 17 00:00:00 2001 From: Patricio Whittingslow Date: Sat, 25 Jul 2026 20:30:36 -0300 Subject: [PATCH] add low level Handle function and more tests --- http/httphi/httphi.go | 118 +++++++++++++------------ http/httphi/router_test.go | 173 +++++++++++++++++++++++++++++++++++-- http/httpraw/header.go | 1 + 3 files changed, 229 insertions(+), 63 deletions(-) diff --git a/http/httphi/httphi.go b/http/httphi/httphi.go index ccdc062..f401725 100644 --- a/http/httphi/httphi.go +++ b/http/httphi/httphi.go @@ -20,13 +20,14 @@ type conn = io.ReadWriteCloser // Router hosts concurrent safe data. type Router struct { - mu sync.Mutex - gen atomic.Uint32 - numGoro int - reqBuf int - respBuf int - pendingConns chan job - mux Mux + mu sync.Mutex + gen atomic.Uint32 + numGoro int + reqBuf int + respBuf int + normalizeKeys bool + pendingConns chan job + mux Mux globbuf []byte exchs []Exchange @@ -99,6 +100,7 @@ func (r *Router) Configure(cfg RouterConfig) error { r.reqBuf = cfg.RequestBufferSize r.respBuf = cfg.ResponseMinBufferSize r.mux = cfg.Mux + r.normalizeKeys = cfg.NormalizeOutgoingKeys if !workerMode { r.backoff = cfg.Backoff r.numGoro = 0 @@ -127,6 +129,52 @@ func (r *Router) Configure(cfg RouterConfig) error { return nil } +// Handle is a extremely low-level HTTP handling method used internally in [Router]. +// Requires exchange to be acquired and configured. Will panic if any argument is nil. +func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error { + reqhdr := &exch.reqHdr + reqhdr.Reset(nil) + var consecutiveBackoffs uint + for { + n, err := reqhdr.ReadFromLimited(exch.rw, reqhdr.BufferFree()) + if err != nil { + exch.readErr = err + return err + } else if n == 0 { + backoff.Do(consecutiveBackoffs) + consecutiveBackoffs++ + continue + } + consecutiveBackoffs = 0 + const asRequest = false + needMore, err := reqhdr.TryParse(asRequest) + if !needMore && err == nil { + // Done! + break + } else if err != nil { + return err + } + } + // Setup Exchange fields necessary for correct functioning. + parsed := reqhdr.BufferParsed() + exch.respRemains = reqhdr.BufferReceived() - parsed + exch.respHeaderOff = uint16(parsed) + exch.respHeaderLen = 0 + // Mux URI. + uri := reqhdr.RequestURI() + meth := reqhdr.Method() + handler := mux.LookupHandler(MethodFromBytes(meth), uri) + if handler != nil { + handler(exch) + exch.FlushHeader() + } else { + exch.WriteHeader(404) + } + // TODO write response from exchange here. + exch.rw.Close() + return nil +} + func (r *Router) Handle(conn conn) error { exch := r.getExch(conn) if exch == nil { @@ -161,50 +209,14 @@ func (r *Router) goroWorker(gen uint32, queue chan job, backoff lneto.BackoffStr func (r *Router) goroHandle(exch *Exchange, backoff lneto.BackoffStrategy, mux Mux) { defer r.freeExch(exch) - reqhdr := &exch.reqHdr - reqhdr.Reset(nil) - var consecutiveBackoffs uint - for { - n, err := reqhdr.ReadFromLimited(exch.rw, reqhdr.BufferFree()) - if err != nil { + err := Handle(exch, mux, backoff) + if err != nil { + if exch.readErr != nil { r.error("goroHandle:ReadFromLimited", slog.String("err", err.Error())) - exch.rw.Close() - return - } else if n == 0 { - backoff(consecutiveBackoffs) - consecutiveBackoffs++ - continue - } - consecutiveBackoffs = 0 - const asRequest = false - needMore, err := reqhdr.TryParse(asRequest) - if !needMore && err == nil { - // Done! - break - } else if err != nil { - r.error("goroHandle:TryParse", slog.String("err", err.Error())) - exch.rw.Close() - return + } else { + r.error("goroHandle:TryParse?", slog.String("err", err.Error())) } } - r.info("goroHandle:headerParsedSuccess") - // Setup Exchange fields necessary for correct functioning. - parsed := reqhdr.BufferParsed() - exch.respRemains = reqhdr.BufferReceived() - parsed - exch.respHeaderOff = uint16(parsed) - exch.respHeaderLen = 0 - // Mux URI. - uri := reqhdr.RequestURI() - meth := reqhdr.Method() - handler := mux.LookupHandler(MethodFromBytes(meth), uri) - if handler != nil { - handler(exch) - } - // Reuse request space as response header start. - - // TODO write response from exchange here. - exch.rw.Close() - } func (r *Router) freeExch(exch *Exchange) { @@ -245,17 +257,13 @@ func (r *Router) getExch(conn conn) (exch *Exchange) { if r.numGoro == 0 { exch := new(Exchange) - exch.Configure(make([]byte, r.respBuf+r.reqBuf), r.reqBuf, false) - + exch.Configure(make([]byte, r.respBuf+r.reqBuf), r.reqBuf, r.normalizeKeys) + exch.Acquire(conn) // Fresh exchange, CAS cannot fail. return exch } return nil } -func (r *Router) allocBuffers(exch *Exchange) { - -} - func (r *Router) error(msg string, attrs ...slog.Attr) { internal.LogAttrs(r.log, slog.LevelError, msg, attrs...) } @@ -279,6 +287,7 @@ type Exchange struct { headerWritten bool normalizeKeys bool nextFree *Exchange + readErr error } func (exch *Exchange) Configure(rawbuf []byte, requestLim int, normalizeKeys bool) { @@ -295,6 +304,7 @@ func (exch *Exchange) Acquire(conn conn) bool { if !exch.used.CompareAndSwap(false, true) { return false } + exch.readErr = nil exch.respTopWritten = 0 exch.respHeaderOff = 0 exch.respHeaderLen = 0 diff --git a/http/httphi/router_test.go b/http/httphi/router_test.go index 5bbf98b..3f11e56 100644 --- a/http/httphi/router_test.go +++ b/http/httphi/router_test.go @@ -6,6 +6,7 @@ import ( "io" "net" "strings" + "sync" "testing" "time" @@ -13,26 +14,62 @@ import ( "github.com/soypat/lneto/internal" ) +// rwconn is a in-memory conn. The router handles connections on another +// goroutine so every field is guarded; onClose lets tests await the handler. type rwconn struct { + mu sync.Mutex readable bytes.Buffer written bytes.Buffer closed bool + onClose chan struct{} deadline time.Time } +// newConn returns a conn preloaded with request and whose Close is observable +// with [rwconn.AwaitClose]. +func newConn(request string) *rwconn { + r := &rwconn{onClose: make(chan struct{})} + r.AddReadable([]byte(request)) + return r +} + func (r *rwconn) Close() error { - r.closed = true + r.mu.Lock() + defer r.mu.Unlock() + if !r.closed { + r.closed = true + if r.onClose != nil { + close(r.onClose) + } + } return nil } + +// AwaitClose blocks until the connection is closed by its handler or timeout elapses. +func (r *rwconn) AwaitClose(t *testing.T, timeout time.Duration) { + t.Helper() + select { + case <-r.onClose: + case <-time.After(timeout): + t.Fatal("timed out awaiting connection close by handler") + } +} + func (r *rwconn) Read(b []byte) (int, error) { + r.mu.Lock() + defer r.mu.Unlock() if r.closed { return 0, net.ErrClosed } else if r.deadlineExceeded() { return 0, context.DeadlineExceeded + } else if r.readable.Len() == 0 { + return 0, io.EOF } return r.readable.Read(b) } func (r *rwconn) Write(b []byte) (int, error) { + r.mu.Lock() + defer r.mu.Unlock() if r.closed { return 0, net.ErrClosed } else if r.deadlineExceeded() { @@ -43,8 +80,16 @@ func (r *rwconn) Write(b []byte) (int, error) { func (r *rwconn) deadlineExceeded() bool { return !r.deadline.IsZero() && time.Since(r.deadline) > 0 } -func (r *rwconn) AddReadable(b []byte) { r.readable.Write(b) } -func (r *rwconn) ViewWritten() []byte { return r.written.Bytes() } +func (r *rwconn) AddReadable(b []byte) { + r.mu.Lock() + defer r.mu.Unlock() + r.readable.Write(b) +} +func (r *rwconn) ViewWritten() string { + r.mu.Lock() + defer r.mu.Unlock() + return r.written.String() +} var _ Mux = (*sliceMux)(nil) @@ -99,29 +144,139 @@ func configSynchronousRouter(t *testing.T, router *Router, bufferSize int, mux M func TestRouterGet(t *testing.T) { const bufferSize = 1024 - const expectResponse = "its time" var ( sm sliceMux router Router - conn rwconn ) - conn.deadline = time.Now().Add(10 * time.Millisecond) // TODO: remove after tests run without blocking forever. sm.Handle("GET /", staticPage(t, expectResponse)) configSynchronousRouter(t, &router, bufferSize, &sm) - err := router.Handle(&conn) + + conn := newConn("GET / HTTP/1.1\r\nHost: tinygo.org\r\n\r\n") + err := router.Handle(conn) if err != nil { t.Fatal(err) } + conn.AwaitClose(t, time.Second) + + got := conn.ViewWritten() + if !strings.HasPrefix(got, "HTTP/1.1 200 OK\r\n") { + t.Errorf("want 200 status line, got %q", got) + } + if !strings.HasSuffix(got, expectResponse) { + t.Errorf("want body %q at end of response, got %q", expectResponse, got) + } +} + +// The handler observes the request line and header fields the router parsed. +func TestRouterRequestVisibleToHandler(t *testing.T) { + const bufferSize = 1024 + var ( + sm sliceMux + router Router + ) + var gotMethod, gotURI, gotHost string + sm.Handle("GET /index.html", func(ex *Exchange) { + gotMethod = string(ex.RequestMethod()) + gotURI = string(ex.RequestURI()) + gotHost = string(ex.RequestHeader("Host")) + ex.WriteHeader(200) + }) + configSynchronousRouter(t, &router, bufferSize, &sm) + + conn := newConn("GET /index.html HTTP/1.1\r\nHost: tinygo.org\r\n\r\n") + if err := router.Handle(conn); err != nil { + t.Fatal(err) + } + conn.AwaitClose(t, time.Second) + + if gotMethod != "GET" { + t.Errorf("want method %q, got %q", "GET", gotMethod) + } + if gotURI != "/index.html" { + t.Errorf("want URI %q, got %q", "/index.html", gotURI) + } + if gotHost != "tinygo.org" { + t.Errorf("want Host %q, got %q", "tinygo.org", gotHost) + } +} + +// Router must route on method and URI, and must not invoke a handler for +// requests it has no registration for. +func TestRouterMux(t *testing.T) { + const bufferSize = 1024 + for _, test := range []struct { + name string + request string + want string // Response body, empty means no handler must run. + }{ + {name: "get root", request: "GET / HTTP/1.1\r\nHost: h\r\n\r\n", want: "root"}, + {name: "get page", request: "GET /page HTTP/1.1\r\nHost: h\r\n\r\n", want: "page"}, + {name: "any method", request: "DELETE /any HTTP/1.1\r\nHost: h\r\n\r\n", want: "any"}, + {name: "method mismatch", request: "POST / HTTP/1.1\r\nHost: h\r\n\r\n", want: ""}, + {name: "unknown uri", request: "GET /nowhere HTTP/1.1\r\nHost: h\r\n\r\n", want: ""}, + } { + t.Run(test.name, func(t *testing.T) { + var ( + sm sliceMux + router Router + ) + sm.Handle("GET /", staticPage(t, "root")) + sm.Handle("GET /page", staticPage(t, "page")) + sm.Handle("/any", staticPage(t, "any")) // No method: matches any. + configSynchronousRouter(t, &router, bufferSize, &sm) + + conn := newConn(test.request) + if err := router.Handle(conn); err != nil { + t.Fatal(err) + } + conn.AwaitClose(t, time.Second) + + got := conn.ViewWritten() + if test.want == "" { + if strings.Contains(got, "root") || strings.Contains(got, "page") || strings.Contains(got, "any") { + t.Errorf("no handler must run, got response %q", got) + } + return + } + if !strings.HasSuffix(got, test.want) { + t.Errorf("want body %q, got response %q", test.want, got) + } + }) + } +} + +// A request arriving in pieces (TCP segmentation) must still be handled. +func TestRouterSplitRequest(t *testing.T) { + const bufferSize = 1024 + const expectResponse = "split ok" + var ( + sm sliceMux + router Router + ) + sm.Handle("GET /", staticPage(t, expectResponse)) + configSynchronousRouter(t, &router, bufferSize, &sm) + + conn := newConn("GET / HTTP/1.1\r\nHo") + if err := router.Handle(conn); err != nil { + t.Fatal(err) + } + conn.AddReadable([]byte("st: tinygo.org\r\n\r\n")) + conn.AwaitClose(t, time.Second) + + if got := conn.ViewWritten(); !strings.HasSuffix(got, expectResponse) { + t.Errorf("want body %q, got response %q", expectResponse, got) + } } func staticPage(t *testing.T, page string) HandlerFunc { return func(ex *Exchange) { n, err := io.WriteString(ex, page) + // Handler runs on the router goroutine: Error, never Fatal. if err != nil { - t.Fatal(err) + t.Error(err) } else if n != len(page) { - t.Fatal("expected written ", len(page), "got", n) + t.Error("expected written ", len(page), "got", n) } } } diff --git a/http/httpraw/header.go b/http/httpraw/header.go index b172480..7aa1001 100644 --- a/http/httpraw/header.go +++ b/http/httpraw/header.go @@ -27,6 +27,7 @@ const ( flagReaderEOF // set if [Header.SetStatus] or [Header.SetStatusInt] has been called. FlagStatusSet + FlagReaderError ) func (f Flags) HasAny(checkThese Flags) bool {