add streaming API distinct from Exchange

examples/http-linux/main-httplinux.go

@@ -100,5 +100,5 @@ func homepage(exch *httphi.Exchange) {
 	exch.StageHeader("Content-Type", "text/html")
 	exch.StageHeaderInt("Content-Length", int64(n), 10)
 	exch.WriteHeader(int(httphi.StatusOK))
-	exch.Write(page[:n])
+	exch.WriteBody(page[:n])
 }

http/httphi/bench_test.go

@@ -67,7 +67,7 @@ func BenchmarkHandle(b *testing.B) {
 				if !present || !internal.BytesEqual(data, expect) {
 					panic("invalid result")
 				}
-				ex.Write(benchBody)
+				ex.WriteBody(benchBody)
 			},
 		},
 		{

http/httphi/exchange.go

@@ -1,6 +1,7 @@
 package httphi
 
 import (
+	"net"
 	"slices"
 	"strconv"
 	"sync/atomic"
@@ -16,7 +17,7 @@ const maxStatusLine = len("HTTP/1.1 ") + 3 + 1 + len("Network Authentication Req
 
 // Exchange is a single request-response cycle over a connection, playing the
 // part of both http.Request and http.ResponseWriter: Request* methods read the
-// request, [Exchange.StageHeader] and [Exchange.Write] produce the response.
+// request, [Exchange.StageHeader] and [Exchange.WriteBody] produce the response.
 // A [Router] owns a fixed pool of them, which is what bounds its memory.
 //
 // Request and response share one buffer, the response header being written over
@@ -24,6 +25,7 @@ const maxStatusLine = len("HTTP/1.1 ") + 3 + 1 + len("Network Authentication Req
 // [Exchange.ReadBody] before setting response headers.
 type Exchange struct {
 	used           atomic.Bool
+	gen            atomic.Uint32
 	respTopBuf     [maxStatusLine]byte
 	respTopWritten uint8
 
@@ -94,6 +96,7 @@ func (exch *Exchange) Acquire(conn conn) bool {
 	if !exch.used.CompareAndSwap(false, true) {
 		return false
 	}
+	exch.gen.Add(1)
 	exch.readErr = nil
 	exch.respErr = nil
 	exch.hijacked = false
@@ -116,10 +119,12 @@ func (exch *Exchange) Release() {
 		exch.rw.Close()
 	}
 	exch.rw = nil
+	exch.gen.Add(1)
 	exch.used.Store(false)
 }
 
-// UnsafeRawBuffer returns the contiguous buffer being used for the request and response.
+// UnsafeRawBuffer returns the contiguous buffer owned by [Exchange] being used for the request and response.
+//
 // Writing to it will mangle the entire request header+body and/or any staged response headers.
 // Does not return the buffer used for the response first line so can be safely
 // written to and used without modifying the staged response first line.
@@ -127,12 +132,13 @@ func (exch *Exchange) Release() {
 // Staging headers will write to this buffer so use mindfully.
 // To access only the request header buffer portion use [httpraw.Header.BufferRaw] limited
 // to [httpraw.Header.BufferParsed] as returned by [Exchange.RequestHeaderRaw].
+// Writing to this section will not change the contents read by [Exchange.ReadBody].
 //
 // In [Router] context, the size of this buffer is influenced directly by [RouterConfig] HeaderBufferSize fields.
 func (exch *Exchange) UnsafeRawBuffer() []byte { return exch.rawbuf }
 
 // StageHeader stages a response header field, written on the first
-// [Exchange.FlushHeader], [Exchange.WriteHeader] or [Exchange.Write].
+// [Exchange.FlushHeader], [Exchange.WriteHeader] or [Exchange.WriteBody].
 // Returns false and drops the field if the response buffer cannot fit it.
 // Has no effect once the header has been written.
 func (exch *Exchange) StageHeader(key, value string) (enoughMemory bool) {
@@ -247,11 +253,71 @@ func (exch *Exchange) FlushHeader() (int, error) {
 	return ng + ng2, err
 }
 
+// ExchangeRW is an [io.ReadWriteCloser] view of an [Exchange] wrapping
+// [Exchange.ReadBody] and [Exchange.WriteBody] methods.
+//
+// Exchanges are pooled and reused, so a handle records the exchange generation
+// it was taken at and refuses to touch the connection once that exchange moves
+// on to another request. Obtain one with [Exchange.ReadWriter].
+type ExchangeRW struct {
+	gen  uint32
+	exch *Exchange
+}
+
+// IsValid returns true while the handle still refers to the request it was
+// taken from, i.e: false once the exchange was released or hijacked away.
+func (rw *ExchangeRW) IsValid() bool {
+	return rw.gen == rw.exch.gen.Load() && rw.exch.used.Load()
+}
+
+func (rw *ExchangeRW) validate() error {
+	if !rw.IsValid() {
+		return net.ErrClosed
+	}
+	return nil
+}
+
+// Write writes response body bytes. See [Exchange.WriteBody].
+// Fails with [net.ErrClosed] once the handle is no longer valid.
+func (rw *ExchangeRW) Write(buf []byte) (int, error) {
+	if err := rw.validate(); err != nil {
+		return 0, err
+	}
+	return rw.exch.WriteBody(buf)
+}
+
+// Read reads request body bytes. See [Exchange.ReadBody].
+// Fails with [net.ErrClosed] once the handle is no longer valid.
+func (rw *ExchangeRW) Read(buf []byte) (int, error) {
+	if err := rw.validate(); err != nil {
+		return 0, err
+	}
+	return rw.exch.ReadBody(buf)
+}
+
+// Close invalidates this handle so later reads and writes fail. It does not
+// close the connection nor end the exchange, both of which the [Router] owns.
+func (rw *ExchangeRW) Close() error {
+	if err := rw.validate(); err != nil {
+		return err
+	}
+	rw.gen--
+	return nil
+}
+
+// ReadWriter fills dst with a stream view of the exchange, valid until the
+// exchange is released. The caller owns dst, so a handler may keep one and
+// refill it every request without allocating.
+func (exch *Exchange) ReadWriter(dst *ExchangeRW) {
+	dst.gen = exch.gen.Load()
+	dst.exch = exch
+}
+
 // Write writes response body bytes, flushing the header first if the handler
 // has not written it yet. Once a write to the connection fails the response is
 // unrecoverable and every later write returns that same error, so a body never
 // reaches the wire without its header.
-func (exch *Exchange) Write(buf []byte) (int, error) {
+func (exch *Exchange) WriteBody(buf []byte) (int, error) {
 	if exch.respErr != nil {
 		return 0, exch.respErr
 	} else if !exch.headerWritten {

http/httphi/exchange_test.go

@@ -3,6 +3,7 @@ package httphi
 import (
 	"context"
 	"errors"
+	"io"
 	"strings"
 
 	"testing"
@@ -79,7 +80,7 @@ func TestExchangeWriteFlushesHeader(t *testing.T) {
 	const body = "hello"
 	conn := newConn("")
 	exch := newExchange(t, conn, 128, false)
-	n, err := exch.Write([]byte(body))
+	n, err := exch.WriteBody([]byte(body))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -266,6 +267,56 @@ func TestHandleSilentHandler(t *testing.T) {
 }
 
 // Body bytes arriving in the same segment as the header must be readable.
+var _ io.ReadWriteCloser = (*ExchangeRW)(nil)
+
+// ExchangeRW writes the response body and reads the request body, so it may be
+// handed to code that wants an io.ReadWriter.
+func TestExchangeRW(t *testing.T) {
+	const body = "hello"
+	conn := newConn("")
+	exch := newExchange(t, conn, 128, false)
+	var rw ExchangeRW
+	exch.ReadWriter(&rw)
+
+	n, err := io.WriteString(&rw, body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != len(body) {
+		t.Errorf("want %d bytes written, got %d", len(body), n)
+	}
+	const want = "HTTP/1.1 200 OK\r\n\r\n" + body
+	if got := conn.ViewWritten(); got != want {
+		t.Errorf("want %q, got %q", want, got)
+	}
+}
+
+// The exchange is pooled and reused: a handle kept past the request it was
+// taken from must fail instead of reaching the next request's connection.
+func TestExchangeRWOutlivesExchange(t *testing.T) {
+	conn := newConn("")
+	exch := newExchange(t, conn, 128, false)
+	var rw ExchangeRW
+	exch.ReadWriter(&rw)
+	if !rw.IsValid() {
+		t.Fatal("want a fresh handle to be valid")
+	}
+	exch.Release()
+
+	if rw.IsValid() {
+		t.Error("want handle invalidated by release")
+	}
+	if _, err := rw.Write([]byte("late")); err == nil {
+		t.Error("want error writing through a released exchange, got nil")
+	}
+	if _, err := rw.Read(make([]byte, 4)); err == nil {
+		t.Error("want error reading through a released exchange, got nil")
+	}
+	if got := conn.ViewWritten(); strings.Contains(got, "late") {
+		t.Errorf("late write reached the connection: %q", got)
+	}
+}
+
 func TestExchangeReadBody(t *testing.T) {
 	const body = "message body"
 	var got string
@@ -406,7 +457,7 @@ func TestExchangeWriteHeaderFlushFails(t *testing.T) {
 	exch := newExchange(t, conn, 128, false)
 	conn.FailWrites(1) // Status line write fails, body write would succeed.
 
-	n, err := exch.Write([]byte(body))
+	n, err := exch.WriteBody([]byte(body))
 	if err == nil {
 		t.Error("want error when header flush fails, got nil")
 	}
@@ -418,7 +469,7 @@ func TestExchangeWriteHeaderFlushFails(t *testing.T) {
 	}
 	// Writes after a failed header stay failed: the response is unrecoverable,
 	// a body without its header would corrupt the stream.
-	if _, err = exch.Write([]byte(body)); err == nil {
+	if _, err = exch.WriteBody([]byte(body)); err == nil {
 		t.Error("want error on write after failed header flush, got nil")
 	}
 	if got := conn.ViewWritten(); got != "" {

http/httphi/router_test.go

@@ -296,7 +296,9 @@ func TestRouterSplitRequest(t *testing.T) {
 
 func staticPage(t *testing.T, page string) HandlerFunc {
 	return func(ex *Exchange) {
-		n, err := io.WriteString(ex, page)
+		var rw ExchangeRW // Streaming APIs take the ReadWriter view.
+		ex.ReadWriter(&rw)
+		n, err := io.WriteString(&rw, page)
 		// Handler runs on the router goroutine: Error, never Fatal.
 		if err != nil {
 			t.Error(err)