Anvil

Anvil is a small Go library of generic building blocks: a TTL cache, a worker pool, channel pipes and a Result[T] type. It has no dependencies and requires Go 1.24 or newer.

go get go.xchunk.org/anvil

import (
    "go.xchunk.org/anvil"
    "go.xchunk.org/anvil/pipes"
)

Cache

An in-memory key-value store whose items expire ttl after they were set. It is safe for concurrent use.

NewCache returns a cache whose items live for ttl. A non-positive ttl makes every item expire immediately.

Get, Set, SetWithTTL, Invalidate

• Get reports false if the key is missing or the item has expired. Concurrent Gets don't block each other.
• Set stores a value and restarts its ttl.
• SetWithTTL is like Set but the item lives for the given ttl. A non-positive value makes the item expire immediately.
• Invalidate removes the item, if any.

c := anvil.NewCache[string, int](time.Minute)
c.Set("answer", 42)

v, ok := c.Get("answer")
fmt.Println(v, ok) // 42 true

c.Invalidate("answer")
_, ok = c.Get("answer")
fmt.Println(ok) // false

GetOrSet

Returns the item stored under key. If there is none, it calls load, stores the result with the cache's default ttl and returns it.

Concurrent calls for the same missing key share a single load: one caller runs it and the others wait for its result.

• An error from load is returned to all waiting callers and is not cached.
• If load panics, the panic propagates in the calling goroutine and waiting callers get an error.

c := anvil.NewCache[string, int](time.Minute)

load := func() (int, error) {
    fmt.Println("loading")
    return 7, nil
}
a, _ := c.GetOrSet("k", load) // loads
b, _ := c.GetOrSet("k", load) // served from the cache
fmt.Println(a, b)
// Output:
// loading
// 7 7

Cleanup, RunCleanup

Cleanup removes all expired items and returns how many were removed. Since Set already sweeps periodically, you only need it to reclaim memory sooner, or when the cache is no longer written to.

RunCleanup calls Cleanup every interval until ctx is done. It blocks, so run it in its own goroutine. It panics if interval is not positive.

go c.RunCleanup(ctx, 30*time.Second)

WorkerPool

Runs tasks on a fixed number of goroutines fed from a bounded queue. Create it with NewWorkerPool, call Start once, submit tasks with Submit or SubmitCtx and finish with Shutdown.

• NewWorkerPool creates size workers with room for queueSize pending tasks. It panics if size is not positive. Workers don't run until Start is called.
• Start launches the workers. Only the first call has an effect.
• Task.Exec receives the context passed to Start.
• Task.Result receives exactly one Response when the task completes. It may be nil if the outcome isn't needed.
• A panic in Exec and a cancelled pool context are also reported through Response.Err.

Submit, SubmitCtx

Submit enqueues a task, blocking while the queue is full. It returns ErrPoolClosed if the pool has been shut down. SubmitCtx is the same but gives up with ctx.Err() if ctx is done before the task could be enqueued.

wp := anvil.NewWorkerPool[int](2, 4)
wp.Start(context.Background())
defer wp.Shutdown()

result := make(chan anvil.Response[int], 1)
wp.Submit(anvil.Task[int]{
    Result: result,
    Exec:   func(ctx context.Context) (int, error) { return 6 * 7, nil },
})

res := <-result
fmt.Println(res.Value, res.Err) // 42 <nil>

Shutdown

Stops accepting tasks, lets queued tasks finish and waits for the workers to exit. It is safe to call more than once. Workers keep draining the queue after cancellation, so every submitted task still gets a response (with the context's error).

pipes

The pipes package connects channels through a processing step. A TransformPipe[In, Out] reads values from an input channel, passes them through a middleware and writes the results to an output channel. Pipe[T] is a TransformPipe[T, T] with a simpler set of options.

Without a middleware a Pipe forwards values unchanged. In async modes a middleware may be called from several goroutines at once.

Start

Processes values until in is closed (returns nil) or ctx is done (returns ctx.Err()), then waits for work still in flight and closes out. It blocks, so run it in its own goroutine.

• By default values are processed one at a time, in order.
• In async modes (WithAsync, WithWorkerPool, WithConcurrency) calls overlap and results reach out in completion order, not input order.
• Returns early with the pool's error if a pool set with WithWorkerPool has been shut down.
• Returns ErrMiddlewareRequired if the pipe has no middleware and In differs from Out.

in, out := make(chan string), make(chan int)
p := pipes.NewTransformPipe(in, out,
    pipes.WithTransformMiddlewareErr(func(_ context.Context, s string) (int, error) {
        return strconv.Atoi(s)
    }),
    pipes.WithTransformConcurrency[string, int](4),
    pipes.WithTransformErrorHandler[string, int](func(err error) { log.Println(err) }),
)
go p.Start(ctx)

Options

Every option has a Pipe form and a WithTransform… form for TransformPipe.

Pause, Resume, Read, Write

• Write sends to the input, Read receives from the output (ok is false once out is closed). Both block while the pipe is paused.
• Push and Pull do the same but ignore Pause.
• Pause / Resume only affect Read and Write; processing done by Start is not affected.

Result experimental

A Result[T] holds either a successful value or an error.

• Of wraps the usual (value, error) pair, so Of(strconv.Atoi(s)) works directly. If err is non-nil the value is discarded.
• Must returns the value or panics with the error — meant for initialization that can't reasonably fail, like Must(regexp.Compile(p)).

Methods

Chaining

These are functions rather than methods because Go methods can't declare their own type parameters. If r holds an error, f is not called and the error is passed through. AndThen lets fallible steps be chained.

r := anvil.Of(strconv.Atoi("12"))
doubled := anvil.Map(r, func(n int) int { return n * 2 })

v, err := doubled.Value()
fmt.Println(v, err) // 24 <nil>

fmt.Println(anvil.Of(strconv.Atoi("x")).UnwrapOr(-1)) // -1

FSM deprecated

Versioning

The module follows semantic versioning. Deprecated and experimental APIs are marked as such in their doc comments. More runnable examples live in example_test.go and pipes/example_test.go; see also go doc and pkg.go.dev.