README
¶
Caravana
Caravana is a Go library for building concurrent pipelines using channels and workers, with retry support and full observability through events.
The library is simple, type-safe, and unopinionated: it runs the pipeline — you decide how to observe, measure, and stop it.
Example
package main
import (
"fmt"
"sync"
"sync/atomic"
"github.com/andmart/caravana"
)
func main() {
const total = 5
var done atomic.Uint64
var wg sync.WaitGroup
wg.Add(1)
// -----------------------------
// Stage 1: multiply by 2
// -----------------------------
stage1 := caravana.NewTaskHolder(
func(v int) (*int, bool, error) {
r := v * 2
return &r, false, nil
},
)
// -----------------------------
// Stage 2: add 1
// -----------------------------
var stage2 *caravana.TaskHolder[int, int]
stage2 = caravana.NewTaskHolder(
func(v int) (*int, bool, error) {
fmt.Println(v + 1)
return nil, false, nil
},
caravana.WithOnEvent(func(e caravana.Event[int, int]) {
if e.Type == caravana.Processed {
if done.Add(1) == total {
stage2.Stop()
stage1.Stop()
wg.Done()
}
}
}),
)
// wire and start pipeline
c := &caravana.Caravana{}
c.Link(stage1, stage2)
c.Start()
// send data
for i := 1; i <= total; i++ {
stage1.Send(i)
}
wg.Wait()
}
✨ Features
- ⚡ Concurrent workers per stage
- 🔁 Retry support per task
- 🔗 Pipeline composition via channels
- 📡 Event-driven observability
- 🧠 Fully generic (
TaskHolder[P, T]) - 🌿 Fan-out: one stage can emit to multiple downstream stages
- 🧼 No built-in logging or metrics
📦 Installation
go get github.com/andmart/caravana
🧩 Concept
A pipeline is composed of TaskHolders, where each one:
- reads from an input channel
- executes a task
- optionally emits to one or more output channels
- triggers events
input → TaskHolder → TaskHolder → TaskHolder
↘ TaskHolder (fan-out)
🏗️ Constructors
NewTaskHolder
NewTaskHolder[P, T](task func(P) (*T, bool, error), opts ...Option[P, T]) *TaskHolder[P, T]
Creates a TaskHolder with an auto-managed input channel. Closes the input channel automatically on Stop(). Use this when you don't need to share the input channel externally.
NewTaskHolderFrom
NewTaskHolderFrom[P, T](in chan P, task func(P) (*T, bool, error), opts ...Option[P, T]) *TaskHolder[P, T]
Creates a TaskHolder from an existing channel. The caller retains ownership of the channel; it is not closed on Stop(). Use this when you manage the channel lifecycle yourself or share it across multiple holders.
🔗 Caravana (pipeline orchestrator)
Caravana wires and manages a set of stages as a unit.
c := &caravana.Caravana{}
c.Link(stage1, stage2, stage3) // connect sequentially
c.Start() // start all stages
c.Stop() // stop all stages
Link
func (c *Caravana) Link(stages ...stage)
Connects stages in sequence: each stage's output is wired to the next stage's input. Can be called multiple times to create fan-out topologies:
c.Link(stage1, stage2) // stage1 → stage2
c.Link(stage1, stage3) // stage1 → stage2 and stage1 → stage3
⚙️ Options
TaskHolder behavior is configured using functional options.
WithOutput
WithOutput(chan T)
Adds an output channel. Can be called multiple times to fan-out to several downstream channels.
WithWorkers
WithWorkers(n int)
Number of concurrent workers.
WithInterval
WithInterval(d time.Duration)
Delay before retry.
WithOnEvent
WithOnEvent(func(Event[P,T]))
Main observability mechanism.
WithMaxRetries
WithMaxRetries(n int)
Maximum number of retry cycles allowed per item. When the task keeps returning retry=true and this limit is reached, an Error event is fired with ErrMaxRetriesExceeded and processing of that item stops. The default (0) means unlimited retries.
caravana.WithMaxRetries[int, int](3)
Detecting the limit in an event callback:
caravana.WithOnEvent(func(e caravana.Event[int, int]) {
if e.Type == caravana.Error && errors.Is(e.Err, caravana.ErrMaxRetriesExceeded) {
fmt.Println("gave up after max retries")
}
})
Retry count semantics: with
maxRetries=N, the loop stops on the N-th retry-requested cycle. N−1Retryevents fire beforeErrMaxRetriesExceededis emitted.
WithCloseChannelsOnStop
WithCloseChannelsOnStop(bool)
Controls whether the input and output channels are closed when Stop() is called. Defaults to true for NewTaskHolder and false for NewTaskHolderFrom.
📡 Events
const (
Received EventType = iota
Processed
Emitted
Retry
Error
)
🧠 Receiving Events
caravana.WithOnEvent(func(e caravana.Event[int,int]) {
switch e.Type {
case caravana.Received:
fmt.Println("received", *e.In)
case caravana.Processed:
fmt.Println("processed")
case caravana.Emitted:
fmt.Println("emitted", *e.Out)
case caravana.Error:
fmt.Println("error:", e.Err)
}
})
Event also implements String() for convenient logging:
fmt.Println(e.String())
// Event{Stage: stage1, Type: Processed, In: 42, Out: <nil>, Err: <nil>, IsRetry: false}
More in examples.
📄 License
MIT
Documentation
¶
Index ¶
- Variables
- type Caravana
- type Event
- type EventType
- type OnEvent
- type Option
- func WithCloseChannelsOnStop[P any, T any](shouldClose bool) Option[P, T]
- func WithInterval[P any, T any](d time.Duration) Option[P, T]
- func WithMaxRetries[P any, T any](n int) Option[P, T]
- func WithName[P any, T any](name string) Option[P, T]
- func WithOnEvent[P any, T any](cb OnEvent[P, T]) Option[P, T]
- func WithOutput[P any, T any](ch chan T) Option[P, T]
- func WithWorkers[P any, T any](n int) Option[P, T]
- type TaskHolder
Constants ¶
This section is empty.
Variables ¶
var ErrMaxRetriesExceeded = errors.New("max retries exceeded")
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option[P any, T any] func(*TaskHolder[P, T])
Option represents a configuration function used to modify a TaskHolder during construction.
Options are applied by NewTaskHolder to configure optional behavior such as the task name, retry interval, channels, worker count, or logger.
This pattern allows the TaskHolder to be configured in a flexible and extensible way without requiring a large constructor with many parameters.
func WithCloseChannelsOnStop ¶
func WithOutput ¶
type TaskHolder ¶
A TaskHolder reads values of type P from the input channel, executes the configured Task, and optionally emits results of type T to the output channel. Errors returned by the Task are forwarded to the error channel.
Task execution is performed by a configurable number of worker goroutines. Each worker processes items independently, allowing concurrent execution of tasks while preserving a simple pipeline model.
If a Task requests a retry, it will be executed again after the configured interval before completing.
Fields:
name Optional identifier used for logging or debugging. interval Delay applied between retry attempts when a task requests retry. task The function executed for each input value. in Channel from which input values are consumed. out Channel where successful task results are emitted when non-nil. err Channel where task errors are forwarded when non-nil. workers Number of worker goroutines processing the input stream. onEvent Callback for receive events.
func NewTaskHolder ¶
func NewTaskHolderFrom ¶
func (*TaskHolder[P, T]) Start ¶
func (th *TaskHolder[P, T]) Start()
func (*TaskHolder[P, T]) Stop ¶
func (th *TaskHolder[P, T]) Stop()
Source Files
Directories