303. Pipes

A mesh with only one component is rarely useful. The true power of FMesh emerges when multiple components are connected using pipes. By connecting components, you enable them to exchange signals and tackle complex problems in small, manageable steps. This idea is not unique to FMesh—it draws inspiration from concepts in object-oriented and functional programming. However, FMesh implements it in a simpler, more transparent way.

Pipes are intentionally simple: they merely connect two ports. In fact, "pipe" is an abstraction — there is no pipe object in the API. Calling PipeTo just registers the destination on the source port. Each pipe connects exactly one output port to exactly one input port.

This design doesn't limit your ability to create one-to-many or many-to-one connections. Such configurations are achieved using multiple pipes, each maintaining the simplicity of a single output-to-input link.

One to one connection

To demonstrate how pipes work, let’s start with a simple example:

// Connect output port "o1" of component "c1" to input port "i1" of component "c2"
if err := c1.OutputByName("o1").PipeTo(c2.InputByName("i1")); err != nil {
    // handle error
}

Semantics: All signals put to the output port o1 of c1 will be transferred to the input port i1 of c2.

Important

On flush, signals are forwarded by pointer (not deep-copied) to destination ports. Output ports are cleared after flushing. Because fan-out shares the same *Signal pointers across destinations, keep signals immutable — use WithLabel, WithLabels, MapPayload, etc. rather than mutating shared state.

One-to-Many Connections

To create one to many connection we create multiple pipes from the same source port:

// Create three pipes from "o1"
_ = c1.OutputByName("o1").PipeTo(
    c2.InputByName("i1"),
    c3.InputByName("i2"),
    c4.InputByName("i3"),
)

Semantics: All signals put on o1 will be copied by reference to all three destination ports (i1,i2 and i3).

Important

Signals are always copied by reference. To avoid unexpected behavior:

• Keep signals immutable whenever possible.
• Design components as pure functions to minimize side effects.
• If necessary, create a deep copy of the signal before modifying it.

Understanding how pointers work in Go is crucial for working with FMesh pipes effectively

Many-to-One Connections

Creating a many-to-one connection is equally straightforward using the same API:

// Connect multiple source ports to the same destination port
_ = c1.OutputByName("o1").PipeTo(sink.InputByName("i1"))
_ = c2.OutputByName("o1").PipeTo(sink.InputByName("i1"))
_ = c3.OutputByName("o1").PipeTo(sink.InputByName("i1"))

Semantics: This configuration creates three pipes from the output ports of different components (c1, c2, and c3) to the same input port (i1) of the sink component. All signals from the respective source ports will appear on i1

You can find runnable fan-in/fan-out examples in the examples repository.

Cyclic Connections

In some scenarios, you may need to create a cyclic connection, where a component’s output port is connected to one of its own input ports. This is fully supported and a common pattern in FMesh. Just as functional programming favors recursion over loops, cyclic pipes enable components to "self-activate."

Example:

// Connect the output port "o1" to the input port "i1" within the same component
_ = c1.OutputByName("o1").PipeTo(c1.InputByName("i1"))

Semantics: The component will reactivate itself in the next cycle, provided there is at least one signal on any of its input ports. This allows the component to control when to activate and how many cycles to execute before stopping

For a practical example, check out the Fibonacci example in the examples repository, which demonstrates how cyclic pipes can be used to implement recursive logic.

That’s all you need to know about pipes in FMesh. Pipes are the backbone of communication between components, enabling you to build modular, efficient, and highly flexible systems. With the fundamental building blocks of ports and pipes, you can create a wide range of powerful and versatile patterns, including:

• Fan-Out (e.g. broadcast,observer, pub-sub): Distribute signals to multiple components.
• Fan-In (e.g. merge,aggregator): Combine signals from multiple components.
• Pipeline: Pass signals sequentially through multiple processing stages.
• Load Balancer: Distribute signals evenly across multiple components.
• Filter: Allow or block signals based on specific conditions.
• State Machine: Manage state transitions based on incoming signals.
• Circuit Breaker: Monitor signals and temporarily halt processing when a failure threshold is reached.
• Rate Limiter: Throttle the flow of signals to ensure components are not overwhelmed.
• Retry Logic: Automatically reprocess failed signals after a specified condition.
• Batch processing: Process signals in groups for efficiency.

These are just a few examples — the examples repository contains runnable implementations of several of these patterns. The simplicity and flexibility of FMesh make it possible to design and implement countless other patterns tailored to your specific needs.

LoopbackPipe

For cyclic connections within a single component, use component.LoopbackPipe:

_ = c.LoopbackPipe("o1", "i1")

This is equivalent to c.OutputByName("o1").PipeTo(c.InputByName("i1")) but validates port names and returns a descriptive error if either port is missing.

Conditional forwarding

When bypassing or routing signals inside an activation function without explicit piping, use port forwarding helpers:

// Forward everything
_ = port.ForwardSignals(c.InputByName("in"), c.OutputByName("out"))

// Forward only high-priority signals
_ = port.ForwardWithFilter(c.InputByName("in"), c.OutputByName("high"),
    signal.LabelEquals("priority", "high"))

// Transform payloads while forwarding
_ = port.ForwardWithMap(c.InputByName("in"), c.OutputByName("out"),
    func(s *signal.Signal) *signal.Signal {
        return s.MapPayload(func(p any) any { return strings.ToUpper(p.(string)) })
    })

Forwarding copies signals to the destination without clearing the source port.