Nodes

A node wraps any callable. Its input types are inferred from the function's parameter list; its output types from the return type.

Node types

Type	Thread model	Use case
Node<Func>	Dedicated thread per node	Default — simplest, most isolated
PoolNode<Func>	Shared ThreadPool	Many nodes, resource-bounded execution
InterruptNode<Func>	Event-driven, no thread	Camera frame ready, timer tick, socket
FanoutNode<T, N>	Dedicated thread	Broadcast one item to N outputs
RouterNode<T, N>	Dedicated thread	Route one item to one of N outputs
FilterNode<T>	Dedicated thread	Pass items matching a predicate

Creating nodes

All node types are created via factory functions that infer types from the callable:

// Free function — simplest case
auto node = make_node<my_func>();

// Stateful functor (operator() is the function)
MyProcessor proc;
auto node = make_node(proc);

// Pool node — shares a ThreadPool with other nodes
auto pool = std::make_shared<ThreadPool>(4);
auto node = make_pool_node<my_func>(pool);

// Interrupt node — triggered externally
auto sched = std::make_shared<ThreadPool>(2);
auto node  = make_interrupt_node<produce_frame>(sched, out<"frame">{});
camera_sdk.on_frame_ready(node.get_trigger());

Channel capacity

Each node's input FIFO has a configurable capacity (default 5):

auto node = make_node<my_func>(/*capacity=*/20);
auto node = make_pool_node<my_func>(pool, /*capacity=*/20);

When an upstream push would exceed capacity, ChannelOverflowError is thrown and the item is dropped. See Error Handling & Events to observe and react to this.

Source nodes

A node with no inputs is a source. It self-submits immediately on start() and re-submits after each execution:

static int produce() {
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    return ++counter;
}
auto src = make_node<produce>();

Tip

Source nodes must sleep or yield to avoid overflowing their output channel. The channel capacity provides the only bound.

Sink nodes

A node with a void return is a sink — it consumes items without producing output:

static void print_it(int x) { std::cout << x << '\n'; }
auto snk = make_node<print_it>();

Error handler

When a node's function throws an unhandled exception, the default behaviour is to stop the node (disabling its channels so the shutdown cascades downstream). Install a handler to override:

    // Return true  → skip this invocation, keep the node running.
    // Return false → stop the node (downstream drains then also stops).
    proc.set_error_handler([](std::string_view name, std::exception_ptr ep) {
        try { std::rethrow_exception(ep); }
        catch (const std::exception& e) {
            std::cerr << "[" << name << "] skipping item — " << e.what() << '\n';
        }
        return true;
    });

See Error Handling & Events for the full picture.