tiny

HTTP Server Component

A complete example of an HTTP server component with Kubernetes service exposure, modeled on the http_server component from http-module.

Overview

This component runs an HTTP listener inside the module pod and hands each incoming request to the flow. It demonstrates:

  • Server lifecycle driven by a Start port (not by settings)
  • Blocking request/response through module.Result
  • Kubernetes Service/Ingress exposure via the ClientAware capability
  • Restart survival via node metadata

The defining pattern: when an HTTP request arrives, the component emits on its request source port and blocks on the returned module.Result. Whatever flows back through the wire graph into the response port becomes the value of that Result — that is the HTTP response. No pending-request maps, no correlation IDs.

Key Types

package server

import (
    "net/url"

    "github.com/tiny-systems/module/module"
)

const (
    ComponentName = "http_server"
    ResponsePort  = "response"
    RequestPort   = "request"
    StartPort     = "start"
    StopPort      = "stop"

    metadataKeyStart = "http-start"
    metadataKeyPort  = "port"
)

type StartContext any

type Header struct {
    Key   string `json:"key" required:"true" title:"Key"`
    Value string `json:"value" required:"true" title:"Value"`
}

type Settings struct {
    EnableStatusPort bool `json:"enableStatusPort" required:"true" title:"Enable status port" description:"Status port notifies when server is up or down"`
    EnableStopPort   bool `json:"enableStopPort" required:"true" title:"Enable stop port" description:"Stop port stops the running server when it receives any message"`
}

// Start is the Start port's payload — wiring a signal into it launches the server
type Start struct {
    Context      StartContext `json:"context,omitempty" configurable:"true" title:"Context" description:"Start context"`
    AutoHostName bool         `json:"autoHostName" title:"Automatically generate hostname" description:"Use cluster auto subdomain setup if any."`
    Hostnames    []string     `json:"hostnames,omitempty" title:"Hostnames" description:"List of virtual hosts this server should be bound to."`
    ReadTimeout  int          `json:"readTimeout" required:"true" title:"Read Timeout"`
    WriteTimeout int          `json:"writeTimeout" required:"true" title:"Write Timeout"`
}

// Request is emitted on the request source port for every incoming HTTP call
type Request struct {
    Context       StartContext `json:"context"`
    RequestURI    string       `json:"requestURI" required:"true"`
    RequestParams url.Values   `json:"requestParams" required:"true"`
    Host          string       `json:"host" required:"true"`
    Method        string       `json:"method" required:"true" title:"Method" enum:"GET,POST,PATCH,PUT,DELETE"`
    RealIP        string       `json:"realIP"`
    Headers       []Header     `json:"headers,omitempty"`
    Body          string       `json:"body"`
    Scheme        string       `json:"scheme"`
}

// Response is what the flow must deliver back into the response port
type Response struct {
    StatusCode  int      `json:"statusCode" required:"true" title:"Status Code" minimum:"100" default:"200" maximum:"599"`
    ContentType string   `json:"contentType" required:"true"`
    Headers     []Header `json:"headers,omitempty" title:"Response headers"`
    Body        string   `json:"body" title:"Response body" format:"textarea"`
}

// Control is the _control port schema — read the public URL here
type Control struct {
    Status     string   `json:"status" title:"Status" readonly:"true"`
    ListenAddr []string `json:"listenAddr" title:"Listen Address" readonly:"true"`
}

Implementation Walkthrough

Component and Lifecycle Capabilities

type Component struct {
    module.Base

    settings      Settings
    startSettings Start
    portMgr       *portmanager.Manager
    // ... cancel funcs, listen addresses, locks
}

// OnClient receives the K8s client (ClientAware capability) and initializes
// the port manager that maintains the shared Service/Ingress.
func (h *Component) OnClient(k8sClient module.K8sClient) {
    h.portMgr = portmanager.New(k8sClient.GetK8sClient(), k8sClient.GetNamespace())
}

// OnSettings receives Settings.
func (h *Component) OnSettings(_ context.Context, msg any) error {
    in, ok := msg.(Settings)
    if !ok {
        return fmt.Errorf("invalid settings message")
    }
    h.settings = in
    return nil
}

// OnReconcile restores a running server from node metadata after pod restart.
func (h *Component) OnReconcile(ctx context.Context, node v1alpha1.TinyNode) error {
    // read metadataKeyStart / metadataKeyPort from node.Status.Metadata,
    // relaunch the listener on context.Background() if it should be running
    h.handleReconcile(node)
    return nil
}

// SyncRPC declares that this component blocks holding a live connection until
// the response port receives the result — its request→response path must be
// delivered over request/reply, never durable fire-and-forget.
func (h *Component) SyncRPC() module.SyncRPCInfo {
    return module.SyncRPCInfo{}
}

Handle: Start, Stop, and Response

func (h *Component) Handle(ctx context.Context, handler module.Handler, port string, msg any) module.Result {
    switch port {
    case StartPort:
        if err := h.handleStart(ctx, handler, msg); err != nil {
            return module.Fail(err)
        }
        return module.Result{}
    case StopPort:
        if err := h.handleStop(); err != nil {
            return module.Fail(err)
        }
        return module.Result{}
    case ResponsePort:
        // The flow delivered the HTTP response. Wrap it in Ok so it travels
        // back through the Result chain to the goroutine holding the socket.
        in, ok := msg.(Response)
        if !ok {
            return module.Fail(fmt.Errorf("invalid response message: got %T", msg))
        }
        return module.Ok(in)
    default:
        return module.Fail(fmt.Errorf("port %s is not supported", port))
    }
}

Serving a Request: Block on the Result

// Registered as the catch-all HTTP handler (echo framework in the real component)
func (h *Component) handleHTTPRequest(c echo.Context, handler module.Handler) error {
    req := h.buildRequest(c) // fills Request from the incoming http.Request

    // Emit on the request port and BLOCK until the flow's response port
    // handler returns. The Result's Value is the Response the flow produced.
    resp := handler(c.Request().Context(), RequestPort, req)
    if err := resp.Err(); err != nil {
        return err
    }

    respObj, ok := resp.Value().(Response)
    if !ok {
        return fmt.Errorf("invalid response: got %T", resp.Value())
    }

    h.writeResponse(c, respObj) // status code, headers, body
    return nil
}

Exposing the Port on the Cluster

Once the listener binds, the component exposes it through the module's shared Service/Ingress and reports the public URLs on the _control port:

publicURLs, err := h.portMgr.ExposePort(ctx, autoHostName, h.startSettings.Hostnames, port)

ExposePort(ctx context.Context, autoHostName string, hostnames []string, port int) ([]string, error) is idempotent — a running server re-asserts it periodically so an externally reset Service heals itself. Cleanup happens via DisclosePort in OnDestroy (leader only), not on every shutdown.

Declared Ports

func (h *Component) Ports() []module.Port {
    ports := []module.Port{
        {Name: v1alpha1.ClientPort},    // requests the K8s client (OnClient)
        {Name: v1alpha1.ReconcilePort}, // enables metadata persistence
        {Name: v1alpha1.SettingsPort, Label: "Settings", Configuration: h.settings},
        {
            Name:                  RequestPort,
            Label:                 "Request",
            Source:                true,
            Configuration:         Request{},
            Position:              module.Right,
            ResponseConfiguration: Response{}, // shape flowing BACK through this port
        },
        {
            Name:          ResponsePort,
            Label:         "Response",
            Position:      module.Right,
            Configuration: Response{StatusCode: 200},
        },
        {Name: v1alpha1.ControlPort, Label: "Dashboard", Source: true, Configuration: h.getControl()},
        {Name: StartPort, Label: "Start", Position: module.Left, Configuration: h.startSettings},
    }
    if h.settings.EnableStopPort {
        ports = append(ports, module.Port{Name: StopPort, Label: "Stop", Position: module.Left, Configuration: Stop{}})
    }
    return ports
}

Note ResponseConfiguration on the request port: it declares the schema of the data expected to flow back, so the editor can validate the response mapping.

Usage

  1. Start it: the server does NOT run until a message arrives on start. Wire a signal node (common-module) into start — prefer signal over cron, since a cron would re-launch on every tick.
  2. Read the URL: after start, the public URL appears in the _control port's listenAddr (or enable the status port). Never guess the address.
  3. Wire the loop: request → (processing nodes) → response. The edge into response maps the final data:
{
  "statusCode": 200,
  "contentType": "application/json",
  "body": "{\"status\": \"ok\", \"received\": \"{{$.context.body}}\"}"
}
  1. Stop it: enable the stop port in settings and send it any message. Context cancellation alone will NOT stop the server — the runtime is distributed and durable, and reconcile deliberately re-hosts the listener after transport timeouts.

Key Patterns Demonstrated

1. Blocking Request/Response via Result

The HTTP goroutine blocks on handler(...); the flow's answer arrives as Result.Value(). This is why handler Results must be returned up the chain by every component in between — a dropped Result is a timed-out HTTP request.

2. ClientAware for Cluster Resources

Declaring {Name: v1alpha1.ClientPort} plus implementing OnClient(client module.K8sClient) gives the component raw cluster access (GetK8sClient(), GetNamespace()) for managing Services and Ingresses.

3. Durable Start Intent

The start configuration and bound port are persisted to node metadata through the reconcile port. A restarted pod finds them in OnReconcile and re-hosts the server on context.Background(). A cancelled Start context is treated as a transport timeout, not a stop — only an explicit Stop (or node deletion) tears the server down.

4. State-Dependent Control Schema

_control shows Running + public URLs while up, Not running otherwise — the dashboard re-renders whatever Ports() currently returns.

Visual Flow

External                +--------------------------------------+
Request --------------->|  http_server                         |
                        |                                      |
   signal ──► start ───►|  listener :port ── ExposePort ─► K8s |
                        |      │                    Service/    |
                        |      ▼                    Ingress     |
                        |  request (Source) ───────────────────┼──► flow nodes
                        |      ▲                               |        │
                        |      │ Result.Value() = Response     |        │
                        |  response ◄──────────────────────────┼────────+
                        +--------------------------------------+

Extension Ideas

  1. TLS: accept PEM cert/key in the Start payload (the shipped component does)
  2. Status Port: emit up/down transitions for monitoring flows
  3. Body Size Limits: cap request bodies via settings
  4. Auth Middleware: pair with the basicauth component from http-module