rill

package module
v0.0.0-...-74e9280 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 5, 2026 License: Apache-2.0 Imports: 6 Imported by: 0

README

Rill - Composable Concurrency for Go

Go Report Card GoDoc License

Rill is a lightweight toolkit that brings composable concurrency to Go, making it easier to build concurrent programs from simple, reusable parts. It reduces boilerplate while preserving Go's natural channel-based model and backpressure behavior.

简体中文: Rill 是一个轻量级的 Go 并发编程工具包，使您能够从简单、可复用的组件构建并发程序。它在保留 Go 原生通道模型和背压行为的同时，减少了样板代码。

Table of Contents

Features

English 中文 Description
Make common tasks easier 简化常见任务 Provides a cleaner and safer way of solving common concurrency problems
Composable and clean 可组合且简洁 Functions take Go channels as inputs and return new, transformed channels as outputs
Centralized error handling 集中式错误处理 Errors are automatically propagated through a pipeline
Stream processing 流处理 Handle potentially infinite streams, processing items as they arrive
Advanced operations 高级操作 Batching, ordered fan-in, map-reduce, stream splitting, merging, and more
Custom extensions 自定义扩展 Easy to write custom functions compatible with the library
Lightweight 轻量级 Small, type-safe, channel-based API with zero external dependencies
Order preservation 顺序保持 Choose between unordered (faster) or ordered (consistent) processing

Go Version Requirements

Requirement Version Reason
Minimum Go 1.18 Uses generics (any type)
Recommended Go 1.21+ Better performance and tooling
Declared Go 1.20 As specified in go.mod

Note: This library uses Go generics extensively. Go 1.17 and earlier versions are not supported.

简体中文: 需要 Go 1.18 或更高版本（使用泛型特性），推荐使用 Go 1.21+。

Installation

go get -u github.com/bycall/rill-go

Quick Start

Here's a practical example demonstrating a concurrent data processing pipeline:

这是一个并发数据处理管道的实际示例：

package main

import (
	"fmt"
	"time"

	"github.com/bycall/rill-go"
)

func main() {
	// Convert a slice to a stream (data source)
	ids := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Process each item concurrently with 3 workers
	processed := rill.Map(ids, 3, func(id int) (int, error) {
		time.Sleep(10 * time.Millisecond)
		return id * 2, nil
	})

	// Filter results - keep only numbers greater than 10
	filtered := rill.Filter(processed, 2, func(n int) (bool, error) {
		return n > 10, nil
	})

	// Consume and print results with 2 workers
	err := rill.ForEach(filtered, 2, func(n int) error {
		fmt.Printf("Processed: %d\n", n)
		return nil
	})

	fmt.Println("Error:", err)
}

Usage Examples

Batching

Processing items in batches can significantly improve performance when working with external services or databases:

import "time"

ids := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)
// Batch by size 3 with 100ms timeout
batches := rill.Batch(ids, 3, 100*time.Millisecond)

err := rill.ForEach(batches, 1, func(batch []int) error {
	fmt.Printf("Processing batch: %v\n", batch)
	return nil
})
Filtering
numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)
evens := rill.Filter(numbers, 2, func(n int) (bool, error) {
	return n%2 == 0, nil
})
Order Preservation

Use ordered versions when you need to maintain input order:

ids := rill.FromSlice([]int{1, 2, 3, 4, 5}, nil)
processed := rill.OrderedMap(ids, 3, func(id int) (int, error) {
	return id * 2, nil
})
Error Handling

Errors are automatically propagated through the pipeline:

stream := rill.FromSlice([]int{1, 2, 3, 4, 5}, nil)
processed := rill.Map(stream, 2, func(n int) (int, error) {
	if n == 3 {
		return 0, fmt.Errorf("error at %d", n)
	}
	return n * 2, nil
})

err := rill.Err(processed)

Core Concepts

Streams and Try Containers

In Rill, a stream is a channel of Try[A] containers. A Try[A] container holds either a value of type A or an error:

type Try[A any] struct {
	Value A
	Error error
}
Non-blocking vs Blocking Functions
  • Non-blocking functions (Map, Filter, Batch) take a stream as input and return a new stream immediately
  • Blocking functions (ForEach, Reduce, ToSlice) consume a stream and return a final result or error
Concurrency Control

Most functions accept a n parameter that controls the number of concurrent workers:

// 3 concurrent workers
result := rill.Map(input, 3, mapperFunc)

API Reference

Stream Creation
  • FromSlice(slice []A, err error) Stream[A] - Convert a slice to a stream
  • FromChan(values <-chan A, err error) Stream[A] - Convert a channel to a stream
  • FromChans(values <-chan A, errs <-chan error) Stream[A] - Convert two channels to a stream
  • Generate(f func(send func(A), sendErr func(error))) Stream[A] - Create a stream from a generator
Transformations
  • Map(in Stream[A], n int, f func(A) (B, error)) Stream[B] - Transform each item
  • OrderedMap(in Stream[A], n int, f func(A) (B, error)) Stream[B] - Transform preserving order
  • Filter(in Stream[A], n int, f func(A) (bool, error)) Stream[A] - Filter items
  • OrderedFilter(in Stream[A], n int, f func(A) (bool, error)) Stream[A] - Filter preserving order
  • FilterMap(in Stream[A], n int, f func(A) (B, bool, error)) Stream[B] - Filter and transform
  • FlatMap(in Stream[A], n int, f func(A) Stream[B]) Stream[B] - Transform and flatten
  • OrderedFlatMap(in Stream[A], n int, f func(A) Stream[B]) Stream[B] - FlatMap preserving order
  • Catch(in Stream[A], n int, f func(error) error) Stream[A] - Catch and handle errors
  • OrderedCatch(in Stream[A], n int, f func(error) error) Stream[A] - Catch errors preserving order
Terminal Operations
  • ForEach(in Stream[A], n int, f func(A) error) error - Apply a function to each item
  • Err(in Stream[A]) error - Return the first error
  • First(in Stream[A]) (value A, found bool, err error) - Get the first item
  • Any(in Stream[A], n int, f func(A) (bool, error)) (bool, error) - Check if any item satisfies the predicate
  • All(in Stream[A], n int, f func(A) (bool, error)) (bool, error) - Check if all items satisfy the predicate
  • Count(in Stream[A]) (int, error) - Count items
  • ToSlice(in Stream[A]) ([]A, error) - Collect items into a slice
Batching
  • Batch(in Stream[A], size int, timeout time.Duration) Stream[[]A] - Group items into batches
  • Unbatch(in Stream[[]A]) Stream[A] - Flatten batches into individual items
Merging & Concatenation
  • Merge(streams ...Stream[A]) Stream[A] - Merge multiple streams
  • Concat(streams ...Stream[A]) Stream[A] - Concatenate multiple streams
  • Zip(in1 Stream[A], in2 Stream[B]) Stream[[2]any] - Zip two streams together
Additional Operations
  • Take(in Stream[A], n int) Stream[A] - Take first n items
  • Skip(in Stream[A], n int) Stream[A] - Skip first n items
  • Distinct(in Stream[A]) Stream[A] - Remove duplicates
  • DistinctBy(in Stream[A], keyFunc func(A) K) Stream[A] - Remove duplicates by key
  • Repeat(value A, count int) Stream[A] - Repeat a value
  • Range(start, end int) Stream[int] - Create a range stream
  • Tap(in Stream[A], f func(A)) Stream[A] - Apply a function for side effects
Utility Functions
  • Wrap(value A, err error) Try[A] - Wrap a value and error
  • ToChans(in Stream[A]) (<-chan A, <-chan error) - Split a stream into channels
  • Discard(in Stream[A]) - Discard remaining items

Contributing

Contributions are welcome! Before submitting a pull request, please consider:

  • Focus on generic, widely applicable solutions
  • Keep the API surface clean and focused
  • Avoid adding external dependencies
  • Add tests for any new features
  • For major changes, please open an issue first to discuss the approach

Acknowledgments

This library is inspired by and builds upon the excellent work in destel/rill. Special thanks to the original author!

License

MIT License - see LICENSE file for details.

Documentation

Overview

Package rill provides a toolkit for composable concurrency in Go, making it easier to build concurrent programs from simple, reusable parts.

Rill reduces boilerplate while preserving Go's natural channel-based model and backpressure behavior. It provides functions for transforming, filtering, batching, and reducing streams of data with full control over concurrency.

Key features:

  • Map, Filter, FlatMap with configurable concurrency
  • Ordered versions of all transformation functions
  • Batch processing with size and timeout controls
  • Reduce and MapReduce for aggregating results
  • Error handling through the Try container type

Basic usage: 

// Convert a slice to a stream
ids := rill.FromSlice([]int{1, 2, 3, 4, 5}, nil)

// Process concurrently with Map
users := rill.Map(ids, 3, func(id int) (*User, error) {
    return fetchUser(id)
})

// Consume the stream
err := rill.ForEach(users, 2, func(u *User) error {
    return saveUser(u)
})

Index

Constants

View Source 
const (
	DefaultBufferSize = 256
	DefaultMaxWorkers = 16
	DefaultLRUSize    = 10000
)

Variables

This section is empty.

Functions

func All 

func All[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error)

func Any 

func Any[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error)

func AnyWithContext 

func AnyWithContext[A any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, A) (bool, error)) (bool, error)

func Batch 

func Batch[A any](in <-chan Try[A], size int, timeout time.Duration) <-chan Try[[]A]

func BatchWithContext 

func BatchWithContext[A any](ctx context.Context, in <-chan Try[A], size int, timeout time.Duration) <-chan Try[[]A]

func Catch 

func Catch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A]

func CatchWithContext 

func CatchWithContext[A any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, error) error) <-chan Try[A]

func CatchWithFallback 

func CatchWithFallback[A any](in <-chan Try[A], n int, fallbackValue A, f func(error) error) <-chan Try[A]

CatchWithFallback catches errors and replaces them with a fallback value. Unlike Catch which can silently drop items, this always outputs either the original value or the fallback.

func Concat 

func Concat[A any](ins ...<-chan Try[A]) <-chan Try[A]

func ConcatWithContext 

func ConcatWithContext[A any](ctx context.Context, ins ...<-chan Try[A]) <-chan Try[A]

func Count 

func Count[A any](in <-chan Try[A]) (int, error)

func CountWindow 

func CountWindow[A any](in <-chan Try[A], size int) <-chan []Try[A]

func Debounce 

func Debounce[A any](ctx context.Context, in <-chan Try[A], delay time.Duration) <-chan Try[A]

func Discard 

func Discard[A any](in <-chan A)

func Distinct 

func Distinct[A comparable](in <-chan Try[A]) <-chan Try[A]

func DistinctBy 

func DistinctBy[A any, K comparable](in <-chan Try[A], keyFunc func(A) K) <-chan Try[A]

func DistinctByWithLimit 

func DistinctByWithLimit[A any, K comparable](in <-chan Try[A], keyFunc func(A) K, maxSize int) <-chan Try[A]

func DistinctWithLimit 

func DistinctWithLimit[A comparable](in <-chan Try[A], maxSize int) <-chan Try[A]

func Drain 

func Drain[A any](ctx context.Context, in <-chan Try[A]) error

func Err 

func Err[A any](in <-chan Try[A]) error

func FanIn 

func FanIn[A any](ctx context.Context, ins ...<-chan Try[A]) <-chan Try[A]

func FanOut 

func FanOut[A any](ctx context.Context, in <-chan Try[A], n int) []<-chan Try[A]

func Filter 

func Filter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[A]

func FilterMap 

func FilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <-chan Try[B]

func FilterMapWithContext 

func FilterMapWithContext[A, B any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, A) (B, bool, error)) <-chan Try[B]

func FilterWithContext 

func FilterWithContext[A any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, A) (bool, error)) <-chan Try[A]

func First 

func First[A any](in <-chan Try[A]) (value A, found bool, err error)

func FlatMap 

func FlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan Try[B]

func FlatMapWithContext 

func FlatMapWithContext[A, B any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, A) <-chan Try[B]) <-chan Try[B]

func ForEach 

func ForEach[A any](in <-chan Try[A], n int, f func(A) error) error

func ForEachWithContext 

func ForEachWithContext[A any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, A) error) error

func FromChan 

func FromChan[A any](values <-chan A, err error) <-chan Try[A]

func FromChans 

func FromChans[A any](values <-chan A, errs <-chan error) <-chan Try[A]

func FromSlice 

func FromSlice[A any](slice []A, err error) <-chan Try[A]

func Generate 

func Generate[A any](f func(send func(A), sendErr func(error))) <-chan Try[A]

func Map 

func Map[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B]

func MapReduce 

func MapReduce[A any, K comparable, V any](in <-chan Try[A], nm int, mapper func(A) (K, V, error), nr int, reducer func(V, V) (V, error)) (map[K]V, error)

func MapReduceWithContext 

func MapReduceWithContext[A any, K comparable, V any](ctx context.Context, in <-chan Try[A], nm int, mapper func(context.Context, A) (K, V, error), nr int, reducer func(context.Context, V, V) (V, error)) (map[K]V, error)

func MapWithContext 

func MapWithContext[A, B any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, A) (B, error)) <-chan Try[B]

func Merge 

func Merge[A any](ins ...<-chan Try[A]) <-chan Try[A]

func MergeWithContext 

func MergeWithContext[A any](ctx context.Context, ins ...<-chan Try[A]) <-chan Try[A]

func Must 

func Must[A any](result Try[A], err error) A

func MustOk 

func MustOk[A any](result Try[A], err error) (A, error)

func OrElse 

func OrElse[A any](in <-chan Try[A], fallback <-chan Try[A]) <-chan Try[A]

OrElse provides a fallback stream when the input stream encounters errors. If an error occurs, it switches to consuming from the fallback stream.

func OrderedFilter 

func OrderedFilter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[A]

func OrderedFilterWithContext 

func OrderedFilterWithContext[A any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, A) (bool, error)) <-chan Try[A]

func OrderedFlatMap 

func OrderedFlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan Try[B]

func OrderedFlatMapWithContext 

func OrderedFlatMapWithContext[A, B any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, A) <-chan Try[B]) <-chan Try[B]

func OrderedMap 

func OrderedMap[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B]

func OrderedMapWithContext 

func OrderedMapWithContext[A, B any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, A) (B, error)) <-chan Try[B]

func ParallelMap 

func ParallelMap[A, B any](ctx context.Context, items []A, workers int, fn func(context.Context, A) (B, error)) ([]B, error)

func ParallelProcess 

func ParallelProcess[A any](ctx context.Context, items []A, workers int, fn func(context.Context, A) error) error

func Range 

func Range(start, end int) <-chan Try[int]

func Reduce 

func Reduce[A any](in <-chan Try[A], n int, f func(A, A) (A, error)) (result A, hasResult bool, err error)

func ReduceWithContext 

func ReduceWithContext[A any](ctx context.Context, in <-chan Try[A], n int, f func(context.Context, A, A) (A, error)) (result A, hasResult bool, err error)

func Repeat 

func Repeat[A any](value A, count int) <-chan Try[A]

func Retry 

func Retry[A any](ctx context.Context, in <-chan Try[A], opts *RetryOptions, f func(context.Context, A) (A, error)) <-chan Try[A]

func SetTracer 

func SetTracer(t Tracer)

func Skip 

func Skip[A any](in <-chan Try[A], n int) <-chan Try[A]

func Take 

func Take[A any](in <-chan Try[A], n int) <-chan Try[A]

func Tap 

func Tap[A any](in <-chan Try[A], f func(A)) <-chan Try[A]

func Throttle 

func Throttle[A any](ctx context.Context, in <-chan Try[A], rate int, burst int) <-chan Try[A]

func TimeWindow 

func TimeWindow[A any](in <-chan Try[A], size int, timeout time.Duration) <-chan []Try[A]

func Timeout 

func Timeout[A any](in <-chan Try[A], timeout time.Duration) <-chan Try[A]

Timeout returns a stream that will timeout if no items are received within the specified duration. Unlike the original buggy implementation, this correctly manages the context lifecycle.

func ToChans 

func ToChans[A any](in <-chan Try[A]) (<-chan A, <-chan error)

func ToSlice 

func ToSlice[A any](in <-chan Try[A]) ([]A, error)

func Unbatch 

func Unbatch[A any](in <-chan Try[[]A]) <-chan Try[A]

func Window 

func Window[A any](in <-chan Try[A], cfg WindowConfig) <-chan []Try[A]

func WithBackoff 

func WithBackoff(d time.Duration) func(*RetryOptions)

func WithBackpressure 

func WithBackpressure(enabled bool) func(*StreamConfig)

func WithBufferSize 

func WithBufferSize(size int) func(*StreamConfig)

func WithContext 

func WithContext[A any](ctx context.Context, in <-chan Try[A]) (<-chan Try[A], <-chan error)

func WithMaxAttempts 

func WithMaxAttempts(n int) func(*RetryOptions)

func WithMaxBackoff 

func WithMaxBackoff(d time.Duration) func(*RetryOptions)

func WithMaxWorkers 

func WithMaxWorkers(n int) func(*StreamConfig)

func Zip 

func Zip[A, B any](ctx context.Context, inA <-chan Try[A], inB <-chan Try[B]) <-chan Try[[2]any]

Types

type LRUCache 

type LRUCache struct {
	// contains filtered or unexported fields
}

func NewLRUCache 

func NewLRUCache(maxSize int) *LRUCache

func (*LRUCache) Add 

func (l *LRUCache) Add(key interface{}) bool

Add returns true if the key is NEW (added), false if it already existed (duplicate). This is the correct semantics for deduplication.

func (*LRUCache) Contains 

func (l *LRUCache) Contains(key interface{}) bool

func (*LRUCache) Get 

func (l *LRUCache) Get(key interface{}) (interface{}, bool)

Get returns the value and true if found, otherwise returns nil and false. This allows callers to distinguish between "key not found" and "value is nil".

func (*LRUCache) Len 

func (l *LRUCache) Len() int

type Metrics 

type Metrics struct {
	// contains filtered or unexported fields
}

func GetMetrics 

func GetMetrics() *Metrics

func (*Metrics) DecrementActiveWorkers 

func (m *Metrics) DecrementActiveWorkers()

func (*Metrics) IncrementActiveWorkers 

func (m *Metrics) IncrementActiveWorkers() int32

func (*Metrics) RecordBlockedGoroutine 

func (m *Metrics) RecordBlockedGoroutine()

func (*Metrics) RecordChannelReceive 

func (m *Metrics) RecordChannelReceive()

func (*Metrics) RecordChannelSend 

func (m *Metrics) RecordChannelSend()

func (*Metrics) RecordError 

func (m *Metrics) RecordError()

func (*Metrics) RecordItem 

func (m *Metrics) RecordItem()

func (*Metrics) RecordProcessingTime 

func (m *Metrics) RecordProcessingTime(d time.Duration)

func (*Metrics) Reset 

func (m *Metrics) Reset()

func (*Metrics) Snapshot 

func (m *Metrics) Snapshot() MetricsSnapshot

type MetricsSnapshot 

type MetricsSnapshot struct {
	ProcessedItems      int64
	ProcessedErrors     int64
	TotalProcessingTime time.Duration
	LastActivityTime    time.Time
	ActiveWorkers       int32
	PeakActiveWorkers   int32
	ChannelSends        int64
	ChannelReceives     int64
	BlockedGoroutines   int64
}

type OnceWithWait 

type OnceWithWait struct {
	// contains filtered or unexported fields
}

func (*OnceWithWait) Do 

func (o *OnceWithWait) Do(f func())

func (*OnceWithWait) Wait 

func (o *OnceWithWait) Wait()

func (*OnceWithWait) WasCalled 

func (o *OnceWithWait) WasCalled() bool

type PipelineBuilder 

type PipelineBuilder struct {
	// contains filtered or unexported fields
}

PipelineBuilder provides a fluent API for building data processing pipelines. Note: Due to Go generics limitations, this builder uses Try[any] internally. Callers should use type assertions on the output. Prefer standalone functions (Map, Filter, FlatMap, etc.) for compile-time type safety.

func NewPipeline 

func NewPipeline[A any](ctx context.Context, source Stream[A]) *PipelineBuilder

NewPipeline creates a new PipelineBuilder from a source stream.

func (*PipelineBuilder) Batch 

func (p *PipelineBuilder) Batch(size int) *PipelineBuilder

Batch groups elements into batches of the specified size.

func (*PipelineBuilder) Build 

func (p *PipelineBuilder) Build() Stream[any]

Build executes the pipeline and returns the output stream.

func (*PipelineBuilder) Filter 

func (p *PipelineBuilder) Filter(f func(context.Context, any) (bool, error)) *PipelineBuilder

Filter keeps elements that satisfy the predicate.

func (*PipelineBuilder) FilterMap 

func (p *PipelineBuilder) FilterMap(f func(context.Context, any) (any, bool, error)) *PipelineBuilder

FilterMap transforms and filters elements.

func (*PipelineBuilder) FlatMap 

func (p *PipelineBuilder) FlatMap(f func(context.Context, any) ([]any, error)) *PipelineBuilder

FlatMap transforms each element into multiple elements.

func (*PipelineBuilder) ForEach 

func (p *PipelineBuilder) ForEach(f func(context.Context, any) error) *PipelineBuilder

ForEach executes a function for each element and forwards results.

func (*PipelineBuilder) Map 

func (p *PipelineBuilder) Map(f func(context.Context, any) (any, error)) *PipelineBuilder

Map transforms each element using the provided function.

func (*PipelineBuilder) Run 

func (p *PipelineBuilder) Run() error

Run executes the pipeline and discards all results.

func (*PipelineBuilder) WithWorkers 

func (p *PipelineBuilder) WithWorkers(n int) *PipelineBuilder

type RetryOptions 

type RetryOptions struct {
	MaxAttempts int
	Backoff     time.Duration
	MaxBackoff  time.Duration
}

func DefaultRetryOptions 

func DefaultRetryOptions() *RetryOptions

type Span 

type Span interface {
	End()
	RecordError(err error)
	SetTag(key string, value any)
}

type Stream 

type Stream[A any] <-chan Try[A]

type StreamConfig 

type StreamConfig struct {
	BufferSize   int
	MaxWorkers   int
	Backpressure bool
}

func DefaultStreamConfig 

func DefaultStreamConfig() *StreamConfig

type Tracer 

type Tracer interface {
	StartSpan(name string) Span
}

type Try 

type Try[A any] struct {
	Value A
	Error error
}

func Wrap 

func Wrap[A any](value A, err error) Try[A]

func (Try[A]) IsError 

func (t Try[A]) IsError() bool

func (Try[A]) IsOk 

func (t Try[A]) IsOk() bool

type WindowConfig 

type WindowConfig struct {
	Size    int
	Slide   int
	Timeout time.Duration
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.