routex

func CompleteOnStop ¶

func CompleteOnStop() AggregateOption

CompleteOnStop emits incomplete groups at shutdown instead of dropping them.

func CompleteSize ¶

func CompleteSize(n int) AggregateOption

CompleteSize completes a group when it has n exchanges.

func CompleteTimeout ¶

func CompleteTimeout(d time.Duration) AggregateOption

CompleteTimeout completes a group after d without a new exchange. The completed aggregate continues through the rest of the route from a background goroutine, so this only works under Engine.Run.

func CompleteWhen ¶

func CompleteWhen(pred func(acc *Exchange) bool) AggregateOption

CompleteWhen completes a group when pred(accumulator) returns true.

func AppendBodies ¶

func AppendBodies[T any]() AggregateStrategy

AppendBodies is a strategy collecting the bodies (each a T) of a group into a []T on the accumulator, in arrival order.

func Otherwise ¶

func Otherwise(steps ...Step) Branch

Otherwise creates the default Choice branch.

func When ¶

func When(match func(*Exchange) bool, steps ...Step) Branch

When creates a Choice branch taken when match returns true.

func WhenBody ¶

func WhenBody[T any](match func(T) bool, steps ...Step) Branch

WhenBody creates a Choice branch that tests the body as T. A body of another type never matches.

func New ¶

func New(opts ...Option) *Engine

New returns an Engine with the given options.

func NewExchange ¶

func NewExchange(body any) *Exchange

NewExchange returns an exchange with a random ID and the given body.

func CollectBodies ¶

func CollectBodies[E any]() ForkOption

CollectBodies is a Join that replaces the original body with a []E holding each part's body in order. A part whose body is not an E is an error.

func ContinueOnError ¶

func ContinueOnError() ForkOption

ContinueOnError keeps processing the remaining fragments after one fails. Failed parts are excluded from the join and all errors are returned together at the end. The default is to stop at the first error.

func Join ¶

func Join(fn func(orig *Exchange, parts []*Exchange) error) ForkOption

Join combines the finished parts back into the original exchange. Parts are in original order; stopped and failed parts are omitted. Without Join the original exchange continues unchanged.

func Parallel ¶

func Parallel(n int) ForkOption

Parallel runs up to n fragments or branches concurrently (n <= 1 is sequential, the default). Parts share the body value of the original: with Parallel the sub-step must not mutate shared state.

func NewMemoryRepository ¶

func NewMemoryRepository(capacity int) *MemoryRepository

NewMemoryRepository returns a repository remembering up to capacity keys (a capacity below 1 is treated as 1).

func WithLogger ¶

func WithLogger(l *slog.Logger) Option

WithLogger sets the logger used for errors nobody can receive (default slog.Default()).

func From ¶

func From(uri string, opts ...RouteOption) *Route

From starts a route consuming the endpoint uri, for example "seda:orders".

func WithDeadLetter ¶

func WithDeadLetter(step Step) RouteOption

WithDeadLetter sets the dead-letter channel: when the route fails for an exchange, the exchange (with HeaderError set and Err populated) is handed to step. If step succeeds the failure counts as handled and the consumer sees no error. The step runs even during shutdown.

package main

import (
	"context"
	"fmt"

	"github.com/JiaBao-do/routex"
)

func main() {
	e := routex.New()
	dlq := routex.Process(func(_ context.Context, ex *routex.Exchange) error {
		fmt.Println("parked:", ex.Body, "-", ex.Err())
		return nil
	})
	_ = e.AddRoute(routex.From("direct:in", routex.WithDeadLetter(dlq)).Then(
		routex.Transform(func(_ context.Context, s string) (int, error) { return 0, fmt.Errorf("cannot parse %q", s) })))
	ctx, cancel := context.WithCancel(context.Background())
	wait, _ := e.Start(ctx)
	_, err := e.Request(ctx, "direct:in", "abc")
	fmt.Println("caller sees:", err)
	cancel()
	_ = wait()
}

Output:
parked: abc - transform: cannot parse "abc"
caller sees: <nil>

func WithID ¶

func WithID(id string) RouteOption

WithID names the route (default: the from URI). Ids must be unique per engine.

func Aggregate ¶

func Aggregate(key func(*Exchange) (string, error), strategy AggregateStrategy, opts ...AggregateOption) Step

Aggregate is the aggregator. Exchanges are grouped by key; strategy folds each into its group; when a group completes (CompleteSize, CompleteWhen or CompleteTimeout) the aggregate continues down the route and the individual exchanges are stopped. With no completion option a group never completes.

Groups live in memory until they complete: always set a size or timeout and bound the key space, otherwise memory grows without limit. Aggregate must be a direct step of a route or of a nested Pipeline.

package main

import (
	"context"
	"fmt"

	"github.com/JiaBao-do/routex"
)

func main() {
	step := routex.Aggregate(
		func(ex *routex.Exchange) (string, error) { return "all", nil },
		routex.AppendBodies[int](),
		routex.CompleteSize(3),
	)
	for i := 1; i <= 3; i++ {
		ex := routex.NewExchange(i)
		_ = step.Process(context.Background(), ex)
		if !ex.Stopped() {
			fmt.Println(ex.Body)
		}
	}
}

Output:
[1 2 3]

func Choice ¶

func Choice(branches ...Branch) Step

Choice is the content-based router: the first branch whose condition matches processes the exchange; an Otherwise branch matches always. If nothing matches the exchange passes through unchanged.

package main

import (
	"context"
	"fmt"

	"github.com/JiaBao-do/routex"
)

func main() {
	step := routex.Choice(
		routex.WhenBody(func(n int) bool { return n < 0 }, routex.SetHeader("kind", "negative")),
		routex.Otherwise(routex.SetHeader("kind", "other")),
	)
	ex := routex.NewExchange(-5)
	_ = step.Process(context.Background(), ex)
	kind, _ := ex.Header("kind")
	fmt.Println(kind)
}

Output:
negative

func Enrich ¶

func Enrich[In, Extra, Out any](fetch func(ctx context.Context, in In) (Extra, error), merge func(in In, extra Extra) Out) Step

Enrich is the content enricher: fetch obtains extra data for the body (a lookup, an HTTP call, ...) and merge combines both into the new body.

func Filter ¶

func Filter[T any](keep func(T) bool) Step

Filter is the message filter: exchanges whose body (as T) fails keep is dropped, that is stopped without error. A body that is not a T is an error.

package main

import (
	"context"
	"fmt"

	"github.com/JiaBao-do/routex"
)

func main() {
	step := routex.Filter(func(n int) bool { return n > 0 })
	for _, n := range []int{-1, 1} {
		ex := routex.NewExchange(n)
		_ = step.Process(context.Background(), ex)
		fmt.Println(n, "dropped:", ex.Stopped())
	}
}

Output:
-1 dropped: true
1 dropped: false

func Idempotent ¶

func Idempotent(key func(*Exchange) (string, error), repo IdempotentRepository, sub Step) Step

Idempotent is the idempotent consumer. key derives a message key; if repo has seen it the exchange is stopped as a duplicate, otherwise sub processes it. When sub fails the key is removed again so a redelivery is not lost.

package main

import (
	"context"
	"fmt"

	"github.com/JiaBao-do/routex"
)

func main() {
	repo := routex.NewMemoryRepository(100)
	step := routex.Idempotent(
		func(ex *routex.Exchange) (string, error) { return ex.Body.(string), nil },
		repo, routex.Process(func(_ context.Context, ex *routex.Exchange) error {
			fmt.Println("processing", ex.Body)
			return nil
		}))
	for _, id := range []string{"a", "a", "b"} {
		_ = step.Process(context.Background(), routex.NewExchange(id))
	}
}

Output:
processing a
processing b

func Multicast ¶

func Multicast(branches []Step, opts ...ForkOption) Step

Multicast sends a copy of the exchange to every branch. Combine the branch results with Join; without it the original exchange continues unchanged.

func Pipeline ¶

func Pipeline(steps ...Step) Step

Pipeline returns a Step that runs steps in order and stops at the first error or when the exchange is stopped. Only steps placed directly in a route or in a nested Pipeline can own background work such as Aggregate timeouts.

func Process ¶

func Process(fn func(ctx context.Context, ex *Exchange) error) Step

Process adapts fn to a Step. It is the escape hatch for custom logic.

func RecipientList ¶

func RecipientList(resolve func(*Exchange) ([]string, error), opts ...ForkOption) Step

RecipientList sends a copy of the exchange to each endpoint URI returned by resolve, computed per exchange. Options are those of Multicast.

func Retry ¶

func Retry(policy RetryPolicy, sub Step) Step

Retry runs sub and retries it on error according to policy. Before every retry the exchange is restored to its state on entry (ID, body value, headers); mutations made in place to the body value itself are not undone. After the last attempt the final error is returned wrapped with the attempt count. The wait between attempts honours context cancellation.

package main

import (
	"context"
	"fmt"

	"github.com/JiaBao-do/routex"
)

func main() {
	calls := 0
	flaky := routex.Process(func(context.Context, *routex.Exchange) error {
		calls++
		if calls < 3 {
			return fmt.Errorf("attempt %d failed", calls)
		}
		return nil
	})
	err := routex.Retry(routex.RetryPolicy{MaxAttempts: 5}, flaky).Process(context.Background(), routex.NewExchange(nil))
	fmt.Println(calls, err)
}

Output:
3 <nil>

func SetHeader ¶

func SetHeader(key string, value any) Step

SetHeader returns a step that sets a header to a constant value.

func Split ¶

func Split[T, E any](fn func(T) ([]E, error), sub Step, opts ...ForkOption) Step

Split is the splitter: fn breaks the body (as T) into fragments and every fragment is processed by sub in its own exchange, with HeaderSplitIndex, HeaderSplitSize and HeaderParentID set. Use Join or CollectBodies to re-assemble the results into the original exchange.

package main

import (
	"context"
	"fmt"
	"strings"

	"github.com/JiaBao-do/routex"
)

func main() {
	step := routex.Split(
		func(s string) ([]string, error) { return strings.Fields(s), nil },
		routex.Transform(func(_ context.Context, w string) (string, error) { return strings.ToUpper(w), nil }),
		routex.CollectBodies[string](),
	)
	ex := routex.NewExchange("a quick fox")
	_ = step.Process(context.Background(), ex)
	fmt.Println(ex.Body)
}

Output:
[A QUICK FOX]

func SplitSeq ¶

func SplitSeq[T, E any](fn func(T) iter.Seq[E], sub Step, opts ...ForkOption) Step

SplitSeq is Split for a splitter function returning an iterator. The iterator is drained before any fragment is processed.

func Stop ¶

func Stop() Step

Stop returns a step that ends processing of the exchange.

func Throttle ¶

func Throttle(n int, per time.Duration) Step

Throttle is the throttler: it lets at most n exchanges per period pass, spacing them evenly (period/n apart) and blocking callers until their slot arrives. A blocked caller returns ctx.Err() if the context ends first. The returned step is safe for concurrent use and its state is per step value. Waiting callers are served in arrival order of their slot reservation.

func To ¶

func To(uri string) Step

To sends the exchange to the endpoint uri and continues with the (possibly replaced) exchange. The producer is resolved on first use through the running Engine, so an unknown scheme fails at runtime, not at build time.

func Transform ¶

func Transform[In, Out any](fn func(ctx context.Context, in In) (Out, error)) Step

Transform is the message translator: it reads the body as In, applies fn and stores the result as the new body. A body that is not an In yields an error wrapping ErrBodyType.

package main

import (
	"context"
	"fmt"

	"github.com/JiaBao-do/routex"
)

func main() {
	step := routex.Transform(func(_ context.Context, s string) (int, error) { return len(s), nil })
	ex := routex.NewExchange("hello")
	_ = step.Process(context.Background(), ex)
	fmt.Println(ex.Body)
}

Output:
5

func WireTap ¶

func WireTap(sub Step) Step

WireTap sends a copy of the exchange to sub in a separate goroutine and continues immediately. Errors from sub are logged, never propagated. Under Engine.Run the tap goroutines are awaited at shutdown and run to completion even after the context is cancelled. The body value is shared with the main flow: sub must not mutate it.