Documentation
¶
Overview ¶
Package pipe is an orch recipe: linear Push → node₀ → … → nodeₙ → Result.
Each Node owns an internal schedx.Scheduler (ingress; default parallel=1, FIFO, maxPending=1) and a chanx subject (egress). Jobs on a node fan out in parallel for the same value (concurrency = len(jobs)); all must succeed before the value is forwarded. The first job error fails fast (siblings canceled) and does not forward.
Lifecycle:
pipe.New(ctx, pipe.NewNode(...), …) → Push* → Result → Close → Done
Pipeline.Result carries values that finished the last node successfully. Pipeline.Done closes when the pipeline is shut down (Close or parent ctx cancel). Callers need not use package chanx.
Jobs are schedx.Job / schedx.JobFunc; this package does not re-export them.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Node ¶
type Node[T any] struct { // contains filtered or unexported fields }
Node describes one stage: ingress via Scheduler, egress after all jobs succeed. Nodes are inert until wired by New.
type NodeOption ¶
NodeOption configures a Node.
func WithJobs ¶
func WithJobs[T any](jobs ...schedx.Job[T]) NodeOption[T]
WithJobs appends jobs to the node's fan-out group. When multiple jobs are configured, they run concurrently for each value (fan-out) and fail fast on the first error. May be called multiple times; the final list is scheduled together. len(jobs)==1 is a plain stage with no fan-out overhead.
func WithName ¶
func WithName[T any](name string) NodeOption[T]
WithName sets a debug label; it does not affect scheduling.
type Pipeline ¶
type Pipeline[T any] interface { Push(ctx context.Context, v T) error // Result receives values that finished the last node successfully. // The channel is closed after the pipeline shuts down. Result() <-chan T // Done closes when the pipeline is shut down (Close or parent context cancel). Done() <-chan struct{} Close() error }
Pipeline wires Push → node₀ → … → nodeₙ → Result.
func New ¶
New builds a linear pipeline and starts all nodes under ctx. Requires at least one node. Parent ctx cancel or Pipeline.Close stops it.
Example ¶
ExampleNew shows a linear pipe: single-job node → parallel multi-job node → single-job node.
package main
import (
"context"
"fmt"
"github.com/xoctopus/concx/pkg/orch/pipe"
"github.com/xoctopus/concx/pkg/schedx"
)
func main() {
ctx := context.Background()
p := pipe.New(ctx,
pipe.NewNode(
pipe.WithName[string]("validate"),
pipe.WithJobs(schedx.JobFunc[string](func(_ context.Context, v string) error {
fmt.Println("validate:", v)
return nil
})),
),
pipe.NewNode(
pipe.WithName[string]("enrich"),
pipe.WithJobs(
schedx.JobFunc[string](func(_ context.Context, v string) error {
fmt.Println("enrich-a:", v)
return nil
}),
schedx.JobFunc[string](func(_ context.Context, v string) error {
fmt.Println("enrich-b:", v)
return nil
}),
),
),
pipe.NewNode(
pipe.WithName[string]("persist"),
pipe.WithJobs(schedx.JobFunc[string](func(_ context.Context, v string) error {
fmt.Println("persist:", v)
return nil
})),
),
)
defer func() { _ = p.Close() }()
_ = p.Push(ctx, "item-1")
fmt.Println("result:", <-p.Result())
}
Output: validate: item-1 enrich-a: item-1 enrich-b: item-1 persist: item-1 result: item-1
Source Files