plugin

Documentation

Index

• func BatchLoopback(name string, matcher message.Matcher, ...) message.Plugin
• func GroupLoopback[K comparable](name string, matcher message.Matcher, ...) message.Plugin
• func Loopback(name string, matcher message.Matcher) message.Plugin
• func ProcessLoopback(name string, matcher message.Matcher, ...) message.Plugin
• type BatchLoopbackConfig
• type GroupLoopbackConfig

Functions

func BatchLoopback

func BatchLoopback(
	name string,
	matcher message.Matcher,
	handle func([]*message.Message) []*message.Message,
	config BatchLoopbackConfig,
) message.Plugin

BatchLoopback batches matching output messages before transformation. Batches are sent when MaxSize is reached or MaxDuration elapses. The handle function returns zero or more messages; return nil to drop. Uses graceful shutdown coordination.

Tracks fan-in: when N messages are batched into M outputs (where M < N), the tracker is adjusted to ensure proper graceful shutdown.

func GroupLoopback

func GroupLoopback[K comparable](
	name string,
	matcher message.Matcher,
	handle func(key K, msgs []*message.Message) []*message.Message,
	keyFunc func(*message.Message) K,
	config GroupLoopbackConfig,
) message.Plugin

GroupLoopback groups matching output messages by key before transformation. Messages with the same key are batched together until config limits are reached. The handle function receives the key and grouped messages; return nil to drop. Uses graceful shutdown coordination.

Tracks fan-in: when N grouped messages produce M outputs (where M < N), the tracker is adjusted to ensure proper graceful shutdown.

func Loopback

func Loopback(name string, matcher message.Matcher) message.Plugin

Loopback routes matching output messages back to the engine for re-processing. Uses graceful shutdown coordination - loopback outputs are closed after the pipeline drains to break cycles.

func ProcessLoopback

func ProcessLoopback(
	name string,
	matcher message.Matcher,
	handle func(*message.Message) []*message.Message,
) message.Plugin

ProcessLoopback routes matching output messages back after transformation. The handle function returns zero or more messages; return nil to drop. Uses graceful shutdown coordination.

Tracks fan-in: if handle returns fewer messages than input (e.g., drops), the tracker is adjusted to ensure proper graceful shutdown.

Types

type BatchLoopbackConfig

type BatchLoopbackConfig struct {
	MaxSize     int           // Flush when batch reaches this size (default: 100).
	MaxDuration time.Duration // Flush after this duration since first item (default: 1s).
}

BatchLoopbackConfig configures the BatchLoopback plugin.

type GroupLoopbackConfig

type GroupLoopbackConfig struct {
	MaxSize             int           // Flush group when it reaches this size (default: 100).
	MaxDuration         time.Duration // Flush group after this duration since first item (default: 1s).
	MaxConcurrentGroups int           // Max active groups (default: 0 meaning unlimited).
}

GroupLoopbackConfig configures the GroupLoopback plugin.