tiny

Handling Messages

The Handle() method is where your component processes incoming messages on its business ports. Understanding message handling patterns is essential for building effective components.

Basic Message Handling

func (c *MyComponent) Handle(
    ctx context.Context,
    output module.Handler,
    port string,
    msg any,
) module.Result {
    if port == "input" {
        input, ok := msg.(InputMessage)
        if !ok {
            return module.Fail(fmt.Errorf("invalid message on port %q", port))
        }
        return output(ctx, "output", c.process(input))
    }
    return module.Fail(fmt.Errorf("unknown port %q", port))
}

The Handler Signature

type Handler func(ctx context.Context, port string, data any) module.Result
ParameterDescription
ctxContext for cancellation and tracing
portOutput port name to send to
dataData to send
Returnsmodule.Result — carries the downstream response or an error

The Result Type

Every Handler call and every Handle invocation returns module.Result. Construct it with the package helpers — never zero-value it manually outside test code:

module.Ok(v)      // success with payload (v may be nil: "delivered, no response data")
module.Fail(err)  // failure (Fail(nil) is a successful no-op)

r.Err()    // the wrapped error, or nil on success
r.Value()  // the wrapped payload, or nil on failure
r.IsErr()  // true when the result wraps an error

Why a typed result instead of any or error? The return must propagate up through blocking-I/O call chains — http-server hands a request to its downstream port and waits on the synchronous response flowing back. A dropped return means a lost response and a timed-out request. Result makes the return explicit at every call site.

System Ports Never Reach Handle

Handle is called for business ports only. Settings, control, reconcile, client and identity messages are dispatched through capability interfaces (OnSettings, OnControl, OnReconcile, ...) before anything reaches Handle. A switch port { case v1alpha1.SettingsPort: ... } inside Handle is dead code.

See System Ports and Settings and Configuration.

Type Assertions

Messages arrive already decoded into the port's Configuration type. Use the safe two-value assertion:

func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    input, ok := msg.(InputMessage)
    if !ok {
        return module.Fail(fmt.Errorf("unexpected message type: %T", msg))
    }
    return c.process(ctx, output, input)
}

Sending Output

Single Output

func (c *Transformer) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    in, ok := msg.(InMessage)
    if !ok {
        return module.Fail(fmt.Errorf("invalid message"))
    }
    // Chain the Result: the downstream response propagates back to the caller
    return output(ctx, "out", OutMessage{Context: in.Context})
}

Multiple Outputs

Check .Err() on each emit before continuing:

func (c *Splitter) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    input := msg.(Input)

    if res := output(ctx, "output_a", input.PartA); res.IsErr() {
        return res
    }
    if res := output(ctx, "output_b", input.PartB); res.IsErr() {
        return res
    }
    return module.Ok(nil)
}

Conditional Output

func (c *Router) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    input := msg.(Message)

    switch input.Type {
    case "urgent":
        return output(ctx, "priority", input)
    case "normal":
        return output(ctx, "standard", input)
    default:
        return output(ctx, "other", input)
    }
}

No Output

func (c *Logger) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    input := msg.(LogMessage)
    log.Info().Str("level", input.Level).Msg(input.Message)
    return module.Ok(nil) // no downstream emit
}

Blocking vs Non-Blocking

Default: Blocking

By default, output() blocks until the downstream flow completes, and its Result carries whatever flowed back:

func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    // Blocks until the entire downstream flow finishes
    res := output(ctx, "output", msg)
    if err := res.Err(); err != nil {
        log.Error().Err(err).Msg("downstream failed")
    }
    return res
}

Time -------------------------------------------------------------->

Component A          Component B          Component C
    |                    |                    |
    | output(B)          |                    |
    | ===================+==================+ |
    |                    | Handle()         | |
    |                    | =================+=+========+
    |                    |                  | | Handle()
    |                    | Result           | | Result
    | Result             |=================+| |========+
    | ===================+                  | |
    v                    v                    v

Background Emits

For fire-and-forget work from goroutines (tickers, watchers, cron loops), do not stash the output handler passed to Handle — implement EmitterAware or embed module.Base and use the injected long-lived handler:

type Component struct {
    module.Base
}

func (c *Component) startLoop() {
    go func() {
        for range time.Tick(time.Second) {
            if err := c.Emit(context.Background(), "out", payload).Err(); err != nil {
                log.Warn().Err(err).Msg("downstream error")
            }
        }
    }()
}

The injected handler stays valid for the runner's lifetime; calling it after the component is destroyed is a safe no-op.

Error Handling

Return Failures

func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    input := msg.(Input)

    result, err := c.process(input)
    if err != nil {
        // The error propagates back through the flow
        return module.Fail(fmt.Errorf("processing failed: %w", err))
    }
    return output(ctx, "output", result)
}

Mark Retryability

The SDK's error contract makes retryability part of the error itself. Wrap transient failures with module.Retryable; permanent ones can be marked explicitly with module.Permanent (plain errors default to not-retryable):

if resp.StatusCode >= 500 {
    return module.Fail(module.Retryable(fmt.Errorf("upstream 5xx: %s", body)))
}
if resp.StatusCode == 400 {
    return module.Fail(module.Permanent(fmt.Errorf("bad request: %s", body)))
}

The scheduler and the retry component both consult module.ShouldRetry(err); only errors explicitly marked retryable are re-attempted.

Error Port Pattern

Emit the canonical module.ErrorMessage shape via module.NewError, which derives Retryable from the error:

func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    input := msg.(Input)

    result, err := c.process(input)
    if err != nil {
        // Route to the error port instead of failing the hop.
        // NewError carries {context, error, retryable}.
        return output(ctx, "error", module.NewError(input, err))
    }
    return output(ctx, "success", result)
}

Prefer this over hand-rolled error structs — the retry component and the platform understand ErrorMessage by construction.

Context Usage

Cancellation

func (c *LongProcessor) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    items := msg.(Input).Items

    for _, item := range items {
        select {
        case <-ctx.Done():
            return module.Fail(ctx.Err()) // graceful shutdown
        default:
            c.processItem(item)
        }
    }
    return module.Ok(nil)
}

Leader Check

Multi-replica modules gate side-effecting work on leadership (mostly relevant inside OnControl/OnReconcile, but available anywhere a context flows):

import "github.com/tiny-systems/module/pkg/utils"

if !utils.IsLeader(ctx) {
    return nil
}

Timeout

func (c *TimedProcessor) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
    defer cancel()

    result, err := c.processWithContext(ctx, msg)
    if err != nil {
        if errors.Is(ctx.Err(), context.DeadlineExceeded) {
            return module.Fail(fmt.Errorf("processing timed out"))
        }
        return module.Fail(err)
    }
    return output(ctx, "output", result)
}

State Management

Settings

Settings arrive through SettingsHandler.OnSettings, not Handle:

type Processor struct {
    settings Settings
}

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

func (p *Processor) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    // settings were applied before any business message arrives
    return p.processWithSettings(ctx, output, msg, p.settings)
}

Accumulating State

type Aggregator struct {
    items []Item
    mu    sync.Mutex
}

func (a *Aggregator) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    switch port {
    case "item":
        a.mu.Lock()
        a.items = append(a.items, msg.(Item))
        a.mu.Unlock()
        return module.Ok(nil)

    case "flush":
        a.mu.Lock()
        items := a.items
        a.items = nil
        a.mu.Unlock()

        return output(ctx, "batch", Batch{Items: items})
    }
    return module.Fail(fmt.Errorf("unknown port %q", port))
}

In-memory state does not survive pod restarts. For durable state use the State backend (via module.Stateful or module.Base) — see Component Patterns.

Common Patterns

Pass-Through with Modification

func (c *Enricher) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    input := msg.(Message)
    input.Timestamp = time.Now()
    return output(ctx, "output", input)
}

Fan-Out

func (c *Broadcaster) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    for _, outPort := range []string{"out1", "out2", "out3"} {
        if res := output(ctx, outPort, msg); res.IsErr() {
            return res
        }
    }
    return module.Ok(nil)
}

Fan-In (via multiple input ports)

func (c *Merger) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    switch port {
    case "input_a", "input_b", "input_c":
        return output(ctx, "merged", msg)
    }
    return module.Fail(fmt.Errorf("unknown port %q", port))
}

Next Steps