fbp.go

package pipeline

import (
	"context"
	"errors"
	"fmt"
	"sync"
)

// Port is a named typed connection point on a Component. Connections carry
// `any`; component implementations type-assert. Buffer sets the channel
// capacity (the backpressure bound — a slow downstream blocks the upstream once
// the buffer fills).
type Port struct {
	Name   string
	ch     chan any
	closed bool
	mu     sync.Mutex
}

// NewPort creates a port with the given bounded buffer.
func NewPort(name string, buffer int) *Port {
	if buffer < 0 {
		buffer = 0
	}
	return &Port{Name: name, ch: make(chan any, buffer)}
}

// Send pushes a value, blocking when the bound is reached (backpressure) until
// space is available or the context is cancelled.
func (p *Port) Send(ctx context.Context, v any) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case p.ch <- v:
		return nil
	}
}

// Recv pulls a value; ok=false when the port is closed and drained.
func (p *Port) Recv(ctx context.Context) (v any, ok bool, err error) {
	select {
	case <-ctx.Done():
		return nil, false, ctx.Err()
	case v, ok := <-p.ch:
		return v, ok, nil
	}
}

// Close closes the underlying channel (idempotent).
func (p *Port) Close() {
	p.mu.Lock()
	defer p.mu.Unlock()
	if !p.closed {
		close(p.ch)
		p.closed = true
	}
}

// Component is a unit in an FBP network: it reads from its input ports and
// writes to its output ports inside Run, returning when its work is complete.
type Component interface {
	Name() string
	Run(ctx context.Context) error
}

// Network wires components together and runs them concurrently. A slow consumer
// applies backpressure to its producer through the bounded ports.
type Network struct {
	components []Component
	conns      []connection
}

type connection struct {
	from *Port
	to   *Port
}

// NewNetwork creates an empty network.
func NewNetwork() *Network { return &Network{} }

// Add registers a component.
func (n *Network) Add(c Component) *Network {
	n.components = append(n.components, c)
	return n
}

// Connect declares that everything sent on `from` should be forwarded to `to`.
func (n *Network) Connect(from, to *Port) error {
	if from == nil || to == nil {
		return errors.New("pipeline: nil port in Connect")
	}
	n.conns = append(n.conns, connection{from: from, to: to})
	return nil
}

// Run starts every component plus a forwarder per connection, and blocks until
// all complete (or the first error / cancellation). Forwarders propagate
// backpressure: forwarding blocks when `to` is full.
func (n *Network) Run(ctx context.Context) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	var wg sync.WaitGroup
	errCh := make(chan error, len(n.components)+len(n.conns))

	// Connection forwarders.
	for _, c := range n.conns {
		wg.Add(1)
		go func(c connection) {
			defer wg.Done()
			for {
				v, ok, err := c.from.Recv(ctx)
				if err != nil {
					errCh <- err
					return
				}
				if !ok {
					c.to.Close()
					return
				}
				if err := c.to.Send(ctx, v); err != nil {
					errCh <- err
					return
				}
			}
		}(c)
	}

	// Components.
	for _, comp := range n.components {
		wg.Add(1)
		go func(comp Component) {
			defer wg.Done()
			if err := comp.Run(ctx); err != nil {
				errCh <- fmt.Errorf("component %s: %w", comp.Name(), err)
				cancel()
			}
		}(comp)
	}

	wg.Wait()
	close(errCh)
	for err := range errCh {
		if err != nil && !errors.Is(err, context.Canceled) {
			return err
		}
	}
	return ctx.Err()
}