nethttp

func Call ¶ added in v0.10.0

func Call[Req, Resp any](
	ctx context.Context,
	client *http.Client,
	baseURL string,
	handle *rest.RouteHandle[Req, Resp],
	req Req,
	vars map[string]string,
	opts CallOptions,
) (Resp, error)

Call executes a typed HTTP request for the given route handle against baseURL.

The concrete URL is built as baseURL + handle.BuildPath(vars) + "?" + queryString. For body-bearing methods (POST, PUT, PATCH) req is JSON-encoded as the request body; for other methods (GET, HEAD, DELETE) req is ignored.

All parameters are validated against their registered codecs before the request is sent: path variables via rest.RouteHandle.BuildPath, query parameters via rest.RouteHandle.ValidateQuery, cookies via rest.RouteHandle.ValidateCookies, and request headers via rest.RouteHandle.ValidateHeaders. A validation failure returns the corresponding rest error type (e.g. rest.PathParamError, rest.QueryParamError) without sending any request.

Security requirements: if the route declares non-nil Security (or inherits global security), CallOptions.CredentialFunc is called to obtain the Authorization headers. A nil CredentialFunc on a secured route is not an error — the request is sent without credential injection; use CallOptions.ExtraHeaders to supply static credentials instead.

On a 2xx response the body is decoded into Resp using the route's response codec. On a non-2xx response UnexpectedStatusError is returned.

CallOptions.Observer receives stats.Observer.RecordRequest after every call (success or failure) with the route path template (not the concrete URL), status code, and total duration. Per-field validation errors are reported separately via stats.Observer.RecordValidationError.

Example — GET with path variable:

handle := getUserRoute.ClientHandle()
user, err := nethttp.Call(ctx, http.DefaultClient, "https://api.example.com",
    handle, struct{}{}, map[string]string{"id": "f47ac10b"},
    nethttp.CallOptions{Observer: obs})

Example — POST with body and bearer token:

handle := createUserRoute.ClientHandle()
resp, err := nethttp.Call(ctx, http.DefaultClient, "https://api.example.com",
    handle, createReq, nil,
    nethttp.CallOptions{
        CredentialFunc: func(ctx context.Context, reqs []route.SecurityRequirement) (http.Header, error) {
            h := make(http.Header)
            h.Set("Authorization", "Bearer "+token)
            return h, nil
        },
    })

// Define the route — use ClientHandle() when no OpenAPI spec is needed.
type Item struct{ ID, Name string }
itemCodec := codex.Struct[Item](
	codex.OptionalField("id", codex.String(),
		func(i Item) string { return i.ID },
		func(i *Item, v string) { i.ID = v },
	),
	codex.RequiredField("name", codex.String().Refine(validate.NonEmptyString),
		func(i Item) string { return i.Name },
		func(i *Item, v string) { i.Name = v },
	),
)
getRoute := rest.NewRoute[getReq, Item]("GET", "/items/{id}",
	codex.Struct[getReq](), itemCodec,
	rest.PathParam{Name: "id"}.WithCodec(codex.String().Refine(validate.NonEmptyString)),
).ClientHandle()

// Validate path params before any HTTP call.
_, err := nethttp.Call(context.Background(), http.DefaultClient, "https://api.example.com",
	getRoute, getReq{}, map[string]string{"id": ""},
	nethttp.CallOptions{})
if err != nil {
	var pathErr rest.PathParamError
	if errors.As(err, &pathErr) {
		fmt.Printf("param %q rejected: %v\n", pathErr.Name, pathErr.Err)
	}
}

Output:
param "id" rejected: constraint failed (non-empty): expected non-empty string

func CallAdapter ¶ added in v0.11.0

func CallAdapter[Req, Resp any](
	client *http.Client,
	baseURL string,
	handle *rest.RouteHandle[Req, Resp],
	opts CallStreamOptions,
) ports.IOAdapter[Req, Resp]

CallAdapter returns a ports.IOAdapter that sends each item as an HTTP request, emitting responses downstream. Use with ports.IOPort.Bind:

domain.Calibration.Bind(ctx, nethttp.CallAdapter(
    httpClient, "http://svc:8080", calibHandle, nethttp.CallStreamOptions{}))

func CallHandle ¶ added in v0.12.0

func CallHandle[Req, Resp any](
	ctx context.Context,
	client *http.Client,
	baseURL string,
	handle *rest.RouteHandle[Req, Resp],
	req Req,
	opts CallOptions,
) (Resp, error)

CallHandle is the single-call convenience wrapper around Call: it derives the path vars AND CallOptions.QueryParams/[HeaderParams]/ [CookieParams] from req automatically, using the route's role-aware merge-field accessors (rest.RouteHandle.PathMergeFields/ [QueryMergeFields]/[HeaderMergeFields]/[CookieMergeFields]) and codex.EncodeVars — one line instead of building each map by hand.

Any entry already present in opts.QueryParams/HeaderParams/CookieParams takes PRECEDENCE over the corresponding derived value for the same key — this lets a caller override a struct field's value, or add ad-hoc params the struct doesn't declare, without losing the one-line convenience for the common case. opts fields left nil are populated entirely from the derived values (or left nil if the route declares no merge fields for that role).

Call remains available as the lower-level escape hatch for callers that build the maps themselves — e.g. no merge fields declared, path vars from a non-struct source, or a route shared between multiple unrelated Req shapes.

Example:

handle := getUserActivity.ClientHandle()
activity, err := nethttp.CallHandle(ctx, client, baseURL, handle,
    GetUserActivityReq{ID: userID, Filter: "logins"}, nethttp.CallOptions{})

func DrainCallAdapter ¶ added in v0.11.0

func DrainCallAdapter[Req, Resp any](
	client *http.Client,
	baseURL string,
	handle *rest.RouteHandle[Req, Resp],
	opts DrainCallOptions,
) ports.SinkAdapter[Req]

DrainCallAdapter returns a ports.SinkAdapter that calls an HTTP endpoint for each item (fire-and-forget; response discarded). Use with ports.SinkPort.Bind:

domain.Events.Bind(ctx, nethttp.DrainCallAdapter(
    client, "http://audit-svc", auditHandle, nethttp.DrainCallOptions{}))

func Handler ¶

func Handler[Req, Resp any](handle *rest.RouteHandle[Req, Resp], fn HandlerFunc[Req, Resp], opts Options) http.Handler

Handler wraps a rest.RouteHandle and a HandlerFunc into an http.Handler.

For body-bearing methods (POST, PUT, PATCH) the request body is read, decoded, and validated using the route's codec before fn is called. For other methods (GET, HEAD, DELETE) fn is called with the zero value of Req.

On success the response is JSON-encoded and written with the HTTP status from the route descriptor's primary response (the first entry in Responses).

Pass a zero-value Options{} for default behaviour (JSON error envelope, 1 MiB body limit, application/json Content-Type check, no-op observer).

func HandlerLatest ¶ added in v0.11.0

func HandlerLatest[Req, Resp any](
	handle *rest.RouteHandle[Req, Resp],
	src gstream.Stream[Resp],
	opts Options,
) http.Handler

HandlerLatest returns an http.Handler that responds to every request with the most recently emitted value from src.

A background goroutine reads src.Values and atomically stores each value. On the first request before any value is available, the handler calls opts.ErrorHandler with HTTP 503 and NoLatestValueError. Errors from src.Errors are reported to opts.Observer but do not affect responses.

Use HandlerLatest for "get current OEE", "get latest sensor reading", or any "current state" REST endpoint backed by a continuously running stream pipeline.

Codec coverage — all HTTP layers validated ¶

Handler validates all codec layers before the fn fires: body codec, query params, cookie params, header params, path params, and security. The decoded [Req] value and all param values are validated but not used for computation — the response is always the latest stream value. This ensures only well-formed requests receive a cached response; invalid requests produce the standard 400.

func IngestAdapter ¶ added in v0.11.0

func IngestAdapter[T any](
	mux *http.ServeMux,
	handle *rest.RouteHandle[T, struct{}],
	opts IngestAdapterOptions,
) ports.SourceAdapter[T]

IngestAdapter returns a ports.SourceAdapter that accepts HTTP requests as pipeline items. When Activate is called it registers a handler with mux. Use with ports.SourcePort.Bind:

domain.SensorReadings.Bind(ctx, nethttp.IngestAdapter(
    mux, ingestHandle, nethttp.IngestAdapterOptions{Buffer: 8}))

func LatestAdapter ¶ added in v0.12.0

func LatestAdapter[Resp any](
	mux *http.ServeMux,
	handle *rest.RouteHandle[struct{}, Resp],
	opts Options,
) ports.LatestAdapter[Resp]

LatestAdapter returns a ports.LatestAdapter that serves a ports.LatestPort's cached value as a GET endpoint — the port-based successor to HandlerLatest/RegisterLatest (which own their own cache cell; the port owns it here). Use with ports.LatestPort.Bind:

handle, _ := domain.Latest.PluginRESTPattern(domain.LatestPattern)
must(domain.Latest.Bind(ctx, nethttp.LatestAdapter(mux, handle, nethttp.Options{})))
go domain.Latest.Feed(ctx, readings)

Before the first value arrives the handler responds 503 Service Unavailable with NoLatestValueError (same semantics as HandlerLatest). All codec layers (params, security) validate exactly as with Handler.

func PipelineAdapter ¶ added in v0.11.0

func PipelineAdapter[Req, Resp any](
	mux *http.ServeMux,
	handle *rest.RouteHandle[Req, Resp],
	opts PipelineAdapterOptions,
) ports.ToolAdapter[Req, Resp]

PipelineAdapter returns a ports.ToolAdapter that registers the pipeline function as an HTTP endpoint via PipelineHandler. When ports.ToolPort.Bind is called it registers the handler with mux. Use with ports.ToolPort.Bind:

domain.OEEToolPort.Bind(ctx, nethttp.PipelineAdapter(mux, httpHandle,
    nethttp.PipelineAdapterOptions{}))

func PipelineHandler ¶ added in v0.11.0

func PipelineHandler[Req, Resp any](
	handle *rest.RouteHandle[Req, Resp],
	fn PipelineHandlerFunc[Req, Resp],
	opts Options,
) http.Handler

PipelineHandler wraps a PipelineHandlerFunc into an http.Handler. All codec validation, param validation, security enforcement, and observer integration follow the same path as plain Handler — PipelineHandler is a thin wrapper that adapts the function signature and collects the result via gstream.Collect.

Use PipelineHandler when the handler body benefits from:

• gstream.Tap for declarative intermediate observation (log/metrics/audit)
• gstream.Apply for multi-step forge function composition
• gstream.MapErr for per-step typed error recovery

For simple one-step handlers, use plain Handler for lower overhead.

Codec coverage — all HTTP layers ¶

Before fn is called, Handler has already validated and decoded:

• Request body (→ req Req)
• Query, cookie, header, path params (all registered rest.Param codecs)
• Security credentials + SecurityFunc

After fn returns, Handler validates:

• Response body (handle.Encode)
• Response header and cookie params (ValidateResponseHeaders / ValidateResponseCookies)

Accessing path/query/cookie/header param values inside the pipeline ¶

The decoded [Req] value (body) is passed directly to fn. To access path, query, cookie, or header param values inside the pipeline (already codec- validated by Handler), call RequestFromContext on the ctx passed to fn:

nethttp.RegisterPipeline(mux, handle,
    func(ctx context.Context, body SensorBody) stream.Stream[OEEResult] {
        r, _ := nethttp.RequestFromContext(ctx)
        sensorID := r.PathValue("sensorID") // already validated
        s := stream.Single(ctx, body)
        s = stream.Tap(ctx, s, func(v SensorBody) {
            slog.Info("request", "sensor", sensorID, "value", v.Value)
        })
        return stream.Apply(ctx, s, oeeCalcFn, opts)
    }, opts)

Response headers and cookies inside the pipeline ¶

Call WithResponseHeaders or WithResponseCookies anywhere inside the pipeline fn (including within gstream.Tap or forge functions). The maps are reference types stored in ctx — writes in the pipeline goroutines are visible to Handler after gstream.Collect returns. This is safe for sequential pipelines (Single → Apply chain). Parallel pipelines that write to response headers concurrently should use a mutex or avoid this pattern.

func PollAdapter ¶ added in v0.11.0

func PollAdapter[Req, Resp any](
	client *http.Client,
	baseURL string,
	handle *rest.RouteHandle[Req, Resp],
	req Req,
	interval time.Duration,
	opts PollStreamOptions,
) ports.SourceAdapter[Resp]

PollAdapter returns a ports.SourceAdapter that polls an HTTP endpoint at interval, emitting each response. Use with ports.SourcePort.Bind:

domain.Configs.Bind(ctx, nethttp.PollAdapter(
    client, "http://config-svc", configHandle, ConfigReq{}, 5*time.Minute, nethttp.PollStreamOptions{}))

func Register ¶

func Register[Req, Resp any](mux *http.ServeMux, handle *rest.RouteHandle[Req, Resp], fn HandlerFunc[Req, Resp], opts Options)

Register registers the route on mux using its method and path from the route descriptor. It uses the Go 1.22+ enhanced ServeMux pattern "METHOD /path" so each registration is scoped to a single method.

Pass a zero-value Options{} for default behaviour.

func RegisterLatest ¶ added in v0.11.0

func RegisterLatest[Req, Resp any](
	mux *http.ServeMux,
	handle *rest.RouteHandle[Req, Resp],
	src gstream.Stream[Resp],
	opts Options,
)

RegisterLatest wires HandlerLatest onto mux using the route's method and path. Mirrors Register.

func RegisterPipeline ¶ added in v0.11.0

func RegisterPipeline[Req, Resp any](
	mux *http.ServeMux,
	handle *rest.RouteHandle[Req, Resp],
	fn PipelineHandlerFunc[Req, Resp],
	opts Options,
)

RegisterPipeline wires PipelineHandler onto mux. Mirrors Register.

func RegisterSSE ¶ added in v0.8.0

func RegisterSSE[Req, Event any](mux *http.ServeMux, handle *rest.SSERouteHandle[Req, Event], fn SSEHandlerFunc[Req, Event], opts Options)

RegisterSSE wires an rest.SSERouteHandle onto mux as a GET SSE endpoint.

func RequestFromContext ¶

func RequestFromContext(ctx context.Context) (*http.Request, bool)

RequestFromContext retrieves the *http.Request stored in ctx by Handler. Returns false if the context was not created by this package.

func ResponseHeadersFromContext ¶ added in v0.8.0

func ResponseHeadersFromContext(ctx context.Context) (http.Header, bool)

ResponseHeadersFromContext retrieves response headers previously stored by WithResponseHeaders. Returns false if no headers were set.

func SSEAdapter ¶ added in v0.11.0

func SSEAdapter[Event any](
	mux *http.ServeMux,
	handle *rest.SSERouteHandle[struct{}, Event],
	opts SSEAdapterOptions,
) ports.SinkAdapter[Event]

SSEAdapter returns a ports.SinkAdapter that serves each item from the SinkPort as an SSE event to all connected clients. Use with ports.SinkPort.Bind:

domain.OEEResults.Bind(ctx, nethttp.SSEAdapter(mux, sseHandle, nethttp.SSEAdapterOptions{}))

func SSEHandler ¶ added in v0.8.0

func SSEHandler[Req, Event any](handle *rest.SSERouteHandle[Req, Event], fn SSEHandlerFunc[Req, Event], opts Options) http.Handler

SSEHandler wraps an rest.SSERouteHandle and a user-supplied SSEHandlerFunc into an http.Handler that streams Server-Sent Events.

The handler sets Content-Type: text/event-stream, Cache-Control: no-cache, and Connection: keep-alive, then calls fn. The send func provided to fn validates the event via the codec, encodes it as JSON, writes "data: <json>\n\n" to the response, and flushes. If the event fails validation, send returns an error without writing anything.

fn should honour ctx.Done() for clean client-disconnect handling.

func SetCookie ¶ added in v0.8.0

func SetCookie(w http.ResponseWriter, name, value string, opts CookieOptions) error

SetCookie writes a Set-Cookie header on w with secure defaults: Secure, HttpOnly, SameSite=Strict, Path="/".

If CookieOptions.Codec is non-nil, value is validated first using that codec. A validation failure returns a rest.CookieParamError without writing any header — the same error type returned by rest.RouteHandle.ValidateCookies on the read side.

Example — symmetric read/write validation from a single codec:

sessionCodec := codex.String().Refine(validate.MinLen(32))

// Read (adapter validates automatically via CookieParam):
rest.CookieParam{Name: "session_token"}.WithCodec(sessionCodec)

// Write (handler sets the cookie with the same codec):
if err := nethttp.SetCookie(w, "session_token", token,
    nethttp.CookieOptions{MaxAge: 3600}.WithCodec(sessionCodec),
); err != nil {
    http.Error(w, err.Error(), http.StatusInternalServerError)
    return
}

func WithResponseCookies ¶ added in v0.8.0

func WithResponseCookies(ctx context.Context, cookies ...PendingCookie)

WithResponseCookies deposits one or more PendingCookie values into ctx. Handler validates their values against the route's [ResponseCookieParam] codecs and then writes Set-Cookie headers on success. Call this inside a HandlerFunc to emit response cookies.

func WithResponseHeaders ¶ added in v0.8.0

func WithResponseHeaders(ctx context.Context, h http.Header)

WithResponseHeaders copies the key-value pairs from h into the response header map stored in ctx (pre-allocated by Handler before calling the HandlerFunc). Call this inside a HandlerFunc to emit response headers such as Location, ETag, or custom headers without direct access to http.ResponseWriter.

resp, _ := svc.Create(ctx, req)
h := make(http.Header)
h.Set("Location", "/users/"+resp.ID)
nethttp.WithResponseHeaders(ctx, h) // mutates the header map in ctx