Array Iterator Component

A complete example of an array processing component, modeled on the array_split component from common-module.

Overview

This component iterates over an array, emitting each item individually, followed by a completion message. It demonstrates:

Sequential, blocking iteration (backpressure by design)
Context propagation through every item
Error propagation via module.Result
A completion port

Complete Implementation

package iterator

import (
    "context"
    "fmt"

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

const (
    ComponentName = "array_iterator"

    InPort   = "in"
    OutPort  = "out"
    DonePort = "done"
)

type Context any

type ItemContext any

// InMessage carries the array plus a context passed through with each item
type InMessage struct {
    Context Context       `json:"context" configurable:"true" title:"Context" description:"Message to be sent further with each item"`
    Array   []ItemContext `json:"array" required:"true" configurable:"true" title:"Array" description:"Array of items to be split"`
}

// OutMessage is emitted once per array element
type OutMessage struct {
    Context Context     `json:"context" title:"Context"`
    Item    ItemContext `json:"item" title:"Item"`
    Index   int         `json:"index" title:"Index"`
    Total   int         `json:"total" title:"Total Items"`
    IsFirst bool        `json:"isFirst" title:"Is First Item"`
    IsLast  bool        `json:"isLast" title:"Is Last Item"`
}

// DoneMessage is emitted after the last item completed
type DoneMessage struct {
    Context        Context `json:"context" title:"Context"`
    TotalProcessed int     `json:"totalProcessed" title:"Total Processed"`
}

type Component struct{}

func (t *Component) Instance() module.Component {
    return &Component{}
}

func (t *Component) GetInfo() module.ComponentInfo {
    return module.ComponentInfo{
        Name:        ComponentName,
        Description: "Array Iterator",
        Info:        "Array iterator. Emits one message per array element on Out, each containing {context, item, index}. Elements are processed sequentially — the next item is sent after the previous Out completes. Emits {context, totalProcessed} on Done when finished.",
        Tags:        []string{"SDK", "ARRAY"},
    }
}

// Handle iterates the array. Each handler call BLOCKS until the downstream
// subtree completes, and returns a Result we must check — an error stops
// the iteration and propagates upstream.
func (t *Component) Handle(ctx context.Context, handler module.Handler, port string, msg any) module.Result {
    if port != InPort {
        return module.Fail(fmt.Errorf("unknown port: %s", port))
    }

    in, ok := msg.(InMessage)
    if !ok {
        return module.Fail(fmt.Errorf("invalid message"))
    }

    total := len(in.Array)
    for i, item := range in.Array {
        r := handler(ctx, OutPort, OutMessage{
            Context: in.Context,
            Item:    item,
            Index:   i,
            Total:   total,
            IsFirst: i == 0,
            IsLast:  i == total-1,
        })
        if r.IsErr() {
            // Propagate the downstream failure — the sender sees which
            // item broke the run.
            return r
        }
    }

    return handler(ctx, DonePort, DoneMessage{
        Context:        in.Context,
        TotalProcessed: total,
    })
}

func (t *Component) Ports() []module.Port {
    return []module.Port{
        {
            Name:          InPort,
            Label:         "In",
            Configuration: InMessage{},
            Position:      module.Left,
        },
        {
            Name:          OutPort,
            Label:         "Out",
            Source:        true,
            Configuration: OutMessage{},
            Position:      module.Right,
        },
        {
            Name:          DonePort,
            Label:         "Done",
            Source:        true,
            Configuration: DoneMessage{},
            Position:      module.Bottom,
        },
    }
}

var _ module.Component = (*Component)(nil)

func init() {
    registry.Register((&Component{}).Instance())
}

The shipped array_split component is the minimal version of this pattern — just {context, item} per element, no index or done port. Reach for it before writing your own.

Usage Example

Edge Configuration into `in`

{
  "context": { "source": "fruit_list" },
  "array": "{{$.context.items}}"
}

Input

{
  "array": ["apple", "banana", "cherry"],
  "context": { "source": "fruit_list" }
}

Outputs (3 messages on `out`, then 1 on `done`)

// Message 1
{ "context": {"source": "fruit_list"}, "item": "apple",  "index": 0, "total": 3, "isFirst": true,  "isLast": false }

// Message 2
{ "context": {"source": "fruit_list"}, "item": "banana", "index": 1, "total": 3, "isFirst": false, "isLast": false }

// Message 3
{ "context": {"source": "fruit_list"}, "item": "cherry", "index": 2, "total": 3, "isFirst": false, "isLast": true }

// Done
{ "context": {"source": "fruit_list"}, "totalProcessed": 3 }

A downstream edge reads item fields as $.item... and the passthrough as $.context....

Visual Flow

+---------------------------------------------------------+
|                    Array Iterator                        |
|                                                          |
|  Input Array: [A, B, C]                                  |
|                                                          |
|  Item A  ---> out (blocks until downstream completes)    |
|  Item B  ---> out                                        |
|  Item C  ---> out                                        |
|                                                          |
|  After all items -------------------------------> done   |
+---------------------------------------------------------+

Key Patterns Demonstrated

1. Blocking Iteration = Backpressure

Each handler(...) call blocks until the downstream subtree for that item completes. The loop cannot outrun the consumer, and ordering is guaranteed:

r := handler(ctx, OutPort, OutMessage{...}) // blocks
if r.IsErr() {
    return r
}

Do not wrap handler calls in goroutines to "speed things up" — go handler(...) discards the Result, which loses both errors and any response flowing back to a blocking caller upstream (e.g. http_server).

2. Checking Every Result

module.Result carries the downstream outcome. r.IsErr() / r.Err() decide whether to continue; returning r unchanged preserves retryability markers (module.Retryable) set deeper in the chain.

3. Context Propagation

The context field rides along with every item so downstream nodes can correlate items with their originating request:

OutMessage{Context: in.Context, Item: item}

4. Completion Signal

A separate done source port lets a flow trigger aggregation after the last item — wire it to whatever should run once per array, not once per item.

Common Use Cases

1. Processing API Results

{ "array": "{{$.context.response.items}}", "context": { "requestId": "{{$.context.requestId}}" } }

Batching: emit slices of N items per message instead of single items
Filtering: skip items based on a configurable predicate
Continue-on-Error: collect failures and report them on done instead of stopping (make it a setting; default should stay fail-fast)
Progress Control Port: publish progress to _control via module.Base.Emit for dashboard visibility