flx

package module
v0.1.6 Latest Latest
Warning

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

 Go to latest Published: Apr 7, 2026 License: MIT Imports: 14 Imported by: 0

README

# flx

flx 是一个以泛型 Stream[T] 为核心的流式处理与动态并发控制库。

它基于 go-zero 的 fx 并发语义重建，但不兼容 fx API 。整体目标很直接：

  • 用 Stream[T] 提供类型安全的流式处理
  • 保留固定并发、无限并发、动态并发和强制缩容能力
  • 用显式 context.Context 替代隐式 context option
  • 让 API 更贴近现代 Go

适用场景

  • 对一批数据做过滤、映射、分组、聚合
  • 需要在 pipeline 中控制并发度
  • 需要在运行中动态调整 worker 数量
  • 需要在缩容时取消多余 worker
  • 需要在流式处理之外复用 retry / timeout / parallel 工具

版本要求

  • Go 1.26.1+

当前模块路径是 github.com/ezra-sullivan/flx，示例代码直接使用：

import "github.com/ezra-sullivan/flx"

如果后续发布到新的模块路径，示例中的导入路径需要同步替换。

快速示例

package main

import (
	"fmt"

	"github.com/ezra-sullivan/flx"
)

func main() {
	out := flx.Map(
		flx.Values(1, 2, 3, 4, 5).
			Filter(func(v int) bool { return v%2 == 0 }),
		func(v int) int { return v * 10 },
	)

	out.ForEach(func(v int) {
		fmt.Println(v)
	})
}

输出：

20
40

实战示例

仓库里还带了一个可直接运行的 HTTP 图片处理 example：

go run ./examples/http_image_pipeline

这个示例会：

  • 从 https://picsum.photos/v2/list 分页拉取图片
  • 下载大图并保存到 examples/http_image_pipeline/.output/original/（默认 5 张）
  • 把大图缩成小图并保存到 examples/http_image_pipeline/.output/processed/
  • 下载阶段会用 flx.DoWithRetryCtx(...) 自动重试，默认 3 次
  • 默认只做下载和处理，不做上传；如需上传，在 examples/http_image_pipeline/main.go 里填 UploadEndpoint

两种实现都在仓库里：

  • Stage 组合版：examples/http_image_pipeline/stage_pipeline.go
  • 原生 flx API 版：examples/http_image_pipeline/native_pipeline.go

完整说明见 doc/examples/http-image-pipeline.md。

API 形状

flx 采用“同类型操作保留方法，跨类型操作使用包级泛型函数”的混合设计。

例如：

  • 同类型操作：Filter、Sort、Head、Skip
  • 跨类型操作：flx.Map、flx.FlatMap、flx.MapContext
  • stage 语义封装：flx.Stage、flx.StageErr、flx.FlatStage、flx.FlatStageErr、flx.Tap

如果一段 pipeline 在若干 stage 之间保持同一个 Stream[T] 类型，可以继续用方法链把这段写得更像“流经多个 stage”，例如 flx.Stage(...).Through(...).Through(...)。

s := flx.Values(1, 2, 3).Filter(func(v int) bool { return v > 1 })
out := flx.Map(s, func(v int) string { return fmt.Sprintf("v=%d", v) })

这样设计不是为了追求风格特别，而是因为 Go 1.26 仍然不支持“额外类型参数的方法”。

核心能力

Stream 构造与中间操作

  • Values / From / FromChan / Concat
  • Filter / Buffer / Sort / Reverse
  • Head / Tail / Skip
  • Map / MapErr
  • FlatMap / FlatMapErr
  • MapContext / FlatMapContext
  • Stage / StageErr / FlatStage / FlatStageErr / Tap
  • DistinctBy / GroupBy / Chunk
  • DistinctByCount / GroupByCount
  • DistinctByWindow / GroupByWindow
  • Reduce

终结与查询

  • Done / DoneErr
  • ForEach / ForEachErr
  • Parallel / ParallelErr
  • Count / Collect
  • First / Last
  • AllMatch / AnyMatch / NoneMatch
  • Max / Min
  • Err

并发与错误控制

  • WithWorkers
  • WithUnlimitedWorkers
  • WithDynamicWorkers
  • WithForcedDynamicWorkers
  • WithInterruptibleWorkers（兼容别名）
  • WithErrorStrategy
  • ErrorStrategyFailFast
  • ErrorStrategyCollect
  • ErrorStrategyLogAndContinue

独立工具函数

  • Parallel / ParallelErr / ParallelWithErrorStrategy
  • DoWithRetry / DoWithRetryCtx
  • DoWithTimeout / DoWithTimeoutCtx

内存边界

  • DistinctBy 会维护当前流的全局已见 key 集合；如果唯一 key 持续增长，内存也会持续增长。
  • GroupBy 会先缓存整条输入流，再按 key 输出分组；它不适合超大数据集或无界流。
  • 这两个 API 更适合有明确边界的批处理数据。
  • DistinctByCount / GroupByCount 采用按输入数量切分的 tumbling window，适合需要更硬内存边界的去重/分组场景。
  • DistinctByWindow / GroupByWindow 采用 processing-time tumbling window，并显式接受 context.Context；它们更适合做按时归档或周期性输出。

动态并发

优雅缩容

缩容时不打断已运行 worker，只影响后续竞争 slot 的 worker：

ctrl := flx.NewConcurrencyController(4)

out := flx.Map(
	flx.Values(1, 2, 3, 4, 5),
	func(v int) int { return v * 10 },
	flx.WithDynamicWorkers(ctrl),
)

ctrl.SetWorkers(8)
ctrl.SetWorkers(2)

out.Done()

强制缩容

缩容时取消多余 worker，要求使用 MapContext* 或 FlatMapContext*：

ctx := context.Background()
ctrl := flx.NewConcurrencyController(4)

out := flx.FlatMapContext(
	ctx,
	flx.Values("a", "b", "c"),
	func(ctx context.Context, v string, pipe chan<- string) {
		select {
		case <-ctx.Done():
			return
		default:
		}

		flx.SendContext(ctx, pipe, v+"!")
	},
	flx.WithForcedDynamicWorkers(ctrl),
)

ctrl.SetWorkers(1)
out.Done()

WithInterruptibleWorkers 仍然可用，但新代码建议统一写成 WithForcedDynamicWorkers。

MapContext / MapContextErr 的单结果发送现在也会响应 ctx.Done()；如果你在 FlatMapContext* 里自己向下游发送值，仍然建议显式使用 SendContext。

错误处理建议

默认错误策略是 ErrorStrategyFailFast。如果 worker 返回错误或 panic：

  • fail-fast：尽快取消当前操作，并在终结阶段暴露错误
  • collect：继续执行，最后合并错误
  • log-and-continue：记录日志并继续

业务代码里，如果你需要稳定的最终错误边界，优先使用会完整消费 source 的 *Err 终结操作，例如 DoneErr / CollectErr：

out := flx.MapErr(flx.Values("1", "x", "3"), strconv.Atoi)
items, err := out.CollectErr()

补充说明：

  • From 的生产函数如果 panic，会进入 stream 错误状态
  • DoneErr / CollectErr 等完整消费型 *Err 终结操作可以显式拿到这类错误
  • FirstErr / AllMatchErr / AnyMatchErr / NoneMatchErr 现在是真正短路：命中结果后立即返回，并在后台 drain 上游
  • 这四个短路 *Err API 返回的是当前错误快照；返回后才发生的 fail-fast error 不保证包含在返回值里
  • First / AllMatch / AnyMatch / NoneMatch 这类短路终结操作也遵循 fail-fast 语义；如果上游已经记录 fail-fast 错误，它们会像其他非 *Err 终结操作一样 panic

flx 默认把错误建模为 stream 状态，而不是官方提供一个 value + error 的 item 容器。 如果你从 fx 迁移过来，之前会在流里传 struct{ Value T; Err error } 这类结果对象，这种写法在 flx 里仍然可以保留，但它应该被视为业务数据建模：适合“部分成功、最后统一收集失败项”的场景，而不是替代 MapErr / CollectErr / DoneErr 这条主错误通道。

与 fx 的主要差异

  • fx.Just -> flx.Values
  • fx.Range -> flx.FromChan
  • stream.Map(...) -> flx.Map(stream, ...)
  • stream.Walk(...) -> flx.FlatMap(stream, ...)
  • stream.WalkCtx... -> flx.MapContext... / flx.FlatMapContext...
  • stream.Merge() -> stream.Collect() / stream.CollectErr()
  • fx.WithDynamicWorkersCtx -> flx.WithForcedDynamicWorkers
  • fx.SendCtx -> flx.SendContext
  • flx 不提供官方 Result[T] / ItemError[T]；需要逐项结果时请自定义业务结构体

如果你从 fx 迁移，建议结合本 README、doc/quickstart.md 和 doc/guide.md 中的 API 对照与语义说明逐步调整。

文档导航

Documentation

Overview

Package flx provides generic stream processing with dynamic concurrency control, explicit context-aware transforms, and reusable retry, timeout, and parallel execution helpers.

Module path: github.com/ezra-sullivan/flx

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrWorkerLimitReduced reports that a worker was canceled because a forced
	// dynamic controller shrank below the number of active workers.
	ErrWorkerLimitReduced = control.ErrWorkerLimitReduced
	// ErrInvalidErrorStrategy reports that an operation received an unsupported
	// worker error handling mode.
	ErrInvalidErrorStrategy = config.ErrInvalidErrorStrategy
	// ErrNilController reports that a dynamic-worker option received a nil
	// concurrency controller.
	ErrNilController = config.ErrNilController
	// ErrInterruptibleWorkersRequireContextTransform reports that forced dynamic
	// workers were requested for a transform that does not accept a context.
	ErrInterruptibleWorkersRequireContextTransform = config.ErrInterruptibleWorkersRequireContextTransform
)
View Source 
var (
	// ErrCanceled is an alias for context.Canceled.
	ErrCanceled = context.Canceled
	// ErrTimeout is an alias for context.DeadlineExceeded.
	ErrTimeout = context.DeadlineExceeded
	// ErrNilContext reports that a required parent context was nil.
	ErrNilContext = errors.New("flx: nil context")
	// ErrNegativeTimeout reports that a timeout duration was negative.
	ErrNegativeTimeout = errors.New("flx: timeout must not be negative")
	// ErrInvalidRetryTimes reports that a retry count was zero or negative.
	ErrInvalidRetryTimes = errors.New("flx: retry times must be greater than 0")
	// ErrNegativeRetryInterval reports that a retry interval was negative.
	ErrNegativeRetryInterval = errors.New("flx: retry interval must not be negative")
	// ErrNegativeRetryTimeout reports that a total retry timeout was negative.
	ErrNegativeRetryTimeout = errors.New("flx: retry timeout must not be negative")
	// ErrNegativeAttemptTimeout reports that a per-attempt timeout was negative.
	ErrNegativeAttemptTimeout = errors.New("flx: attempt timeout must not be negative")
	// ErrAttemptTimeoutRequiresRetryCtx reports that attempt timeouts only work
	// with the context-aware retry API.
	ErrAttemptTimeoutRequiresRetryCtx = errors.New("flx: WithAttemptTimeout requires DoWithRetryCtx")
	// ErrRetryAttemptTimeout reports that one retry attempt exceeded its own
	// attempt timeout.
	ErrRetryAttemptTimeout = errors.New("flx: retry attempt timeout")
)
View Source 
var (
	// ErrInvalidWindowCount reports that a count-based window size was less than
	// one.
	ErrInvalidWindowCount = errors.New("flx: window count must be greater than 0")
	// ErrInvalidWindowDuration reports that a time-based window duration was not
	// positive.
	ErrInvalidWindowDuration = errors.New("flx: window duration must be positive")
)

Functions

func DoWithRetry 

func DoWithRetry(fn func() error, opts ...RetryOption) error

DoWithRetry executes fn until it succeeds or the retry budget is exhausted.

func DoWithRetryCtx 

func DoWithRetryCtx(ctx context.Context, fn func(context.Context, int) error, opts ...RetryOption) error

DoWithRetryCtx executes fn until it succeeds or the retry budget is exhausted, passing the current attempt context and zero-based attempt index.

func DoWithTimeout 

func DoWithTimeout(fn func() error, timeout time.Duration, opts ...TimeoutOption) error

DoWithTimeout runs fn with a derived timeout context and returns its result.

func DoWithTimeoutCtx 

func DoWithTimeoutCtx(fn func(context.Context) error, timeout time.Duration, opts ...TimeoutOption) error

DoWithTimeoutCtx runs fn with a derived timeout context and passes that context into the callback.

func Parallel 

func Parallel(fns ...func())

Parallel runs each function in its own goroutine and panics if the chosen fail-fast strategy records an error.

func ParallelErr 

func ParallelErr(fns ...func() error) error

ParallelErr runs each function in its own goroutine and returns the joined worker errors without applying stream fail-fast semantics.

func ParallelWithErrorStrategy 

func ParallelWithErrorStrategy(strategy ErrorStrategy, fns ...func()) error

ParallelWithErrorStrategy runs each function in its own goroutine and applies strategy to worker panics and returned errors.

func Reduce 

func Reduce[T, R any](s Stream[T], fn func(<-chan T) (R, error)) (R, error)

Reduce hands s's source channel to fn, drains any remaining items after fn returns, and joins fn's returned error with the stream error state.

func SendContext 

func SendContext[T any](ctx context.Context, pipe chan<- T, item T) bool

SendContext sends item to pipe unless ctx has already been canceled.

Types

type ConcurrencyController 

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

ConcurrencyController coordinates resizable worker limits for dynamic stream operations. It is safe for concurrent use.

func NewConcurrencyController 

func NewConcurrencyController(workers int) *ConcurrencyController

NewConcurrencyController returns a controller whose worker limit starts at workers, clamped to at least one slot.

func (*ConcurrencyController) ActiveWorkers 

func (c *ConcurrencyController) ActiveWorkers() int

ActiveWorkers returns the number of workers currently holding semaphore slots.

func (*ConcurrencyController) SetWorkers 

func (c *ConcurrencyController) SetWorkers(n int)

SetWorkers updates the target worker count. When the limit shrinks, the newest registered interruptible workers are canceled until the active count fits within the new limit.

func (*ConcurrencyController) Workers 

func (c *ConcurrencyController) Workers() int

Workers returns the configured worker limit.

type DynamicSemaphore 

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

DynamicSemaphore is a resizable semaphore used by dynamic worker pipelines. It is safe for concurrent use.

func NewDynamicSemaphore 

func NewDynamicSemaphore(n int) *DynamicSemaphore

NewDynamicSemaphore returns a semaphore with capacity n, clamped to at least one slot.

func (*DynamicSemaphore) Acquire 

func (s *DynamicSemaphore) Acquire()

Acquire blocks until a slot is available.

func (*DynamicSemaphore) AcquireCtx 

func (s *DynamicSemaphore) AcquireCtx(ctx context.Context) error

AcquireCtx blocks until a slot is available or ctx is canceled.

func (*DynamicSemaphore) Cap 

func (s *DynamicSemaphore) Cap() int

Cap returns the configured semaphore capacity.

func (*DynamicSemaphore) Current 

func (s *DynamicSemaphore) Current() int

Current returns the number of slots currently held.

func (*DynamicSemaphore) Release 

func (s *DynamicSemaphore) Release()

Release frees one acquired slot and wakes the next waiter if capacity is available.

func (*DynamicSemaphore) Resize 

func (s *DynamicSemaphore) Resize(n int)

Resize updates the semaphore capacity and wakes queued waiters when the new capacity allows additional work to start.

type ErrorStrategy 

type ErrorStrategy uint8

ErrorStrategy controls how concurrent operations react to worker failures. 

const (
	// ErrorStrategyFailFast cancels the operation and makes non-Err terminals
	// panic once the failure is observed.
	ErrorStrategyFailFast ErrorStrategy = iota
	// ErrorStrategyCollect records worker errors and returns them from Err
	// terminals without canceling sibling workers.
	ErrorStrategyCollect
	// ErrorStrategyLogAndContinue logs worker errors and allows the operation to
	// continue without recording them in stream state.
	ErrorStrategyLogAndContinue
)

func (ErrorStrategy) String 

func (s ErrorStrategy) String() string

String returns a human-readable representation of s.

type Group added in v0.1.2 

type Group[K comparable, T any] = streaming.Group[K, T]

Group holds one grouping key plus the items assigned to that key.

type Option 

type Option func(*config.Options)

Option mutates the execution settings for one transform or terminal call.

func WithDynamicWorkers 

func WithDynamicWorkers(controller *ConcurrencyController) Option

WithDynamicWorkers enables graceful dynamic resizing for the current operation. Shrinking does not interrupt workers that already hold a slot.

func WithErrorStrategy 

func WithErrorStrategy(strategy ErrorStrategy) Option

WithErrorStrategy configures how worker panics and errors are handled for the current operation.

func WithForcedDynamicWorkers 

func WithForcedDynamicWorkers(controller *ConcurrencyController) Option

WithForcedDynamicWorkers enables forced dynamic resizing for the current operation. Shrinking cancels excess workers via context and only works with MapContext or FlatMapContext variants.

func WithInterruptibleWorkers deprecated 

func WithInterruptibleWorkers(controller *ConcurrencyController) Option

WithInterruptibleWorkers is a compatibility alias for WithForcedDynamicWorkers.

Deprecated: use WithForcedDynamicWorkers.

func WithUnlimitedWorkers 

func WithUnlimitedWorkers() Option

WithUnlimitedWorkers spawns one worker per item for the current operation.

func WithWorkers 

func WithWorkers(workers int) Option

WithWorkers limits the current operation to a fixed number of workers.

type RetryOption 

type RetryOption func(*retryOptions)

RetryOption mutates the behavior of one retry call.

func WithAttemptTimeout 

func WithAttemptTimeout(timeout time.Duration) RetryOption

WithAttemptTimeout sets a timeout for each individual retry attempt.

func WithIgnoreErrors 

func WithIgnoreErrors(ignoreErrors []error) RetryOption

WithIgnoreErrors treats matching errors as successful completion.

func WithInterval 

func WithInterval(interval time.Duration) RetryOption

WithInterval sets the delay between failed attempts.

func WithRetry 

func WithRetry(times int) RetryOption

WithRetry sets the maximum number of attempts, including the first one.

func WithTimeout 

func WithTimeout(timeout time.Duration) RetryOption

WithTimeout sets an overall timeout for the full retry loop.

type Stream 

type Stream[T any] struct {
	// contains filtered or unexported fields
}

Stream is a lazy sequence of values backed by a channel plus shared error state that records upstream worker failures.

func Chunk 

func Chunk[T any](s Stream[T], n int) Stream[[]T]

Chunk groups items into slices of size n, emitting a final short chunk when the source ends.

func Concat 

func Concat[T any](s Stream[T], others ...Stream[T]) Stream[T]

Concat merges s with others and returns a stream that emits items from all inputs as they arrive.

func DistinctBy 

func DistinctBy[T any, K comparable](s Stream[T], fn func(T) K) Stream[T]

DistinctBy keeps the first item for each key produced by fn.

func DistinctByCount added in v0.1.2 

func DistinctByCount[T any, K comparable](s Stream[T], n int, fn func(T) K) Stream[T]

DistinctByCount keeps the first item for each key within windows of n input items, resetting the seen-key set after every window.

func DistinctByWindow added in v0.1.2 

func DistinctByWindow[T any, K comparable](ctx context.Context, s Stream[T], every time.Duration, fn func(T) K) Stream[T]

DistinctByWindow keeps the first item for each key within a time window that starts when the first item in that window arrives.

func FlatMap 

func FlatMap[T, U any](s Stream[T], fn func(T, chan<- U), opts ...Option) Stream[U]

FlatMap calls fn for each item and lets fn emit zero or more output values.

func FlatMapContext 

func FlatMapContext[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T, chan<- U), opts ...Option) Stream[U]

FlatMapContext is FlatMap with a caller-provided context passed into each worker.

func FlatMapContextErr 

func FlatMapContextErr[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T, chan<- U) error, opts ...Option) Stream[U]

FlatMapContextErr is FlatMapErr with a caller-provided context passed into each worker.

func FlatMapErr 

func FlatMapErr[T, U any](s Stream[T], fn func(T, chan<- U) error, opts ...Option) Stream[U]

FlatMapErr calls fn for each item and records any returned worker error in the stream state.

func FlatStage added in v0.1.5 

func FlatStage[I, O any](
	ctx context.Context,
	in Stream[I],
	fn func(context.Context, I, chan<- O),
	opts ...Option,
) Stream[O]

FlatStage applies fn to each item in in and lets fn emit zero or more output values. It is a thin semantic wrapper around FlatMapContext for stage-oriented pipelines.

func FlatStageErr added in v0.1.5 

func FlatStageErr[I, O any](
	ctx context.Context,
	in Stream[I],
	fn func(context.Context, I, chan<- O) error,
	opts ...Option,
) Stream[O]

FlatStageErr applies fn to each item in in, lets fn emit zero or more output values, and records returned worker errors in the stream state. It is a thin semantic wrapper around FlatMapContextErr.

func From 

func From[T any](generate func(chan<- T)) Stream[T]

From adapts a producer callback into a stream. Panics from generate are captured in the stream state and surfaced by terminal operations.

func FromChan 

func FromChan[T any](source <-chan T) Stream[T]

FromChan wraps source as a Stream without changing its production semantics.

func GroupBy 

func GroupBy[T any, K comparable](s Stream[T], fn func(T) K) Stream[[]T]

GroupBy drains s, groups items by fn, and emits groups in first-seen key order.

func GroupByCount added in v0.1.2 

func GroupByCount[T any, K comparable](s Stream[T], n int, fn func(T) K) Stream[Group[K, T]]

GroupByCount groups items by fn within windows of n input items and emits one Group per key in first-seen order for each window.

func GroupByWindow added in v0.1.2 

func GroupByWindow[T any, K comparable](ctx context.Context, s Stream[T], every time.Duration, fn func(T) K) Stream[Group[K, T]]

GroupByWindow groups items by fn within a time window that starts when the first item in that window arrives and flushes on timer tick or source close.

func Map 

func Map[T, U any](s Stream[T], fn func(T) U, opts ...Option) Stream[U]

Map applies fn to each item in s and emits the mapped values.

func MapContext 

func MapContext[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T) U, opts ...Option) Stream[U]

MapContext is Map with a caller-provided context passed into each worker.

func MapContextErr 

func MapContextErr[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T) (U, error), opts ...Option) Stream[U]

MapContextErr is MapErr with a caller-provided context passed into each worker.

func MapErr 

func MapErr[T, U any](s Stream[T], fn func(T) (U, error), opts ...Option) Stream[U]

MapErr applies fn to each item in s and records any returned error in the stream state.

func Stage added in v0.1.5 

func Stage[I, O any](
	ctx context.Context,
	in Stream[I],
	fn func(context.Context, I) O,
	opts ...Option,
) Stream[O]

Stage applies fn to each item in in and emits the mapped values. It is a thin semantic wrapper around MapContext for pipelines that want explicit stage-shaped call sites without introducing a second runtime model.

func StageErr added in v0.1.5 

func StageErr[I, O any](
	ctx context.Context,
	in Stream[I],
	fn func(context.Context, I) (O, error),
	opts ...Option,
) Stream[O]

StageErr applies fn to each item in in and records returned worker errors in the stream state. It is a thin semantic wrapper around MapContextErr for stage-oriented pipelines that still want stream-level error handling.

func Tap added in v0.1.5 

func Tap[T any](
	ctx context.Context,
	in Stream[T],
	fn func(context.Context, T) error,
	opts ...Option,
) Stream[T]

Tap runs fn for each item in in and re-emits the original item when fn succeeds. If fn returns an error, Tap records that error in the stream state and does not forward the failed item.

func Values 

func Values[T any](items ...T) Stream[T]

Values returns a stream that emits items in order and then closes.

func (Stream[T]) AllMatch 

func (s Stream[T]) AllMatch(predicate func(T) bool) bool

AllMatch reports whether every item satisfies predicate. It drains the remainder of the stream after the first mismatch so delayed fail-fast errors can still surface.

func (Stream[T]) AllMatchErr 

func (s Stream[T]) AllMatchErr(predicate func(T) bool) (bool, error)

AllMatchErr reports whether every item satisfies predicate and returns the current error state when it short-circuits.

func (Stream[T]) AnyMatch 

func (s Stream[T]) AnyMatch(predicate func(T) bool) bool

AnyMatch reports whether any item satisfies predicate. It drains the remainder of the stream after the first match so delayed fail-fast errors can still surface.

func (Stream[T]) AnyMatchErr 

func (s Stream[T]) AnyMatchErr(predicate func(T) bool) (bool, error)

AnyMatchErr reports whether any item satisfies predicate and returns the current error state when it short-circuits.

func (Stream[T]) Buffer 

func (s Stream[T]) Buffer(n int) Stream[T]

Buffer inserts a channel buffer of size n between s and the returned stream.

func (Stream[T]) Collect 

func (s Stream[T]) Collect() []T

Collect drains the stream into a slice and panics on fail-fast errors.

func (Stream[T]) CollectErr 

func (s Stream[T]) CollectErr() ([]T, error)

CollectErr drains the stream into a slice and returns the final error state.

func (Stream[T]) Concat 

func (s Stream[T]) Concat(others ...Stream[T]) Stream[T]

Concat merges s with others while preserving per-stream item order. Items from different input streams may interleave based on runtime scheduling.

func (Stream[T]) Count 

func (s Stream[T]) Count() int

Count drains the stream and returns the number of items it produced.

func (Stream[T]) CountErr 

func (s Stream[T]) CountErr() (int, error)

CountErr drains the stream, returns the item count, and returns the final error state.

func (Stream[T]) Done 

func (s Stream[T]) Done()

Done drains the stream and panics if a fail-fast error was recorded.

func (Stream[T]) DoneErr 

func (s Stream[T]) DoneErr() error

DoneErr drains the stream and returns the final error state.

func (Stream[T]) Err 

func (s Stream[T]) Err() error

Err returns the currently accumulated stream error without draining the stream.

func (Stream[T]) Filter 

func (s Stream[T]) Filter(fn func(T) bool, opts ...Option) Stream[T]

Filter keeps only the items for which fn returns true.

func (Stream[T]) First 

func (s Stream[T]) First() (T, bool)

First returns the first item and then drains the rest of the stream so any delayed fail-fast error is observed before the call returns.

func (Stream[T]) FirstErr 

func (s Stream[T]) FirstErr() (T, bool, error)

FirstErr returns the first item and the current error state, then drains the remaining source asynchronously.

func (Stream[T]) ForAll 

func (s Stream[T]) ForAll(fn func(<-chan T))

ForAll hands the raw source channel to fn, then drains any leftovers and applies fail-fast panic behavior.

func (Stream[T]) ForAllErr 

func (s Stream[T]) ForAllErr(fn func(<-chan T)) error

ForAllErr hands the raw source channel to fn, then drains any leftovers and returns the final error state.

func (Stream[T]) ForEach 

func (s Stream[T]) ForEach(fn func(T))

ForEach calls fn for every item in the stream and panics if a fail-fast error was recorded.

func (Stream[T]) ForEachErr 

func (s Stream[T]) ForEachErr(fn func(T)) error

ForEachErr calls fn for every item in the stream and returns the final error state.

func (Stream[T]) Head 

func (s Stream[T]) Head(n int64) Stream[T]

Head returns a stream containing at most the first n items from s. The upstream source is drained after the head is satisfied so producers can exit.

func (Stream[T]) Last 

func (s Stream[T]) Last() (T, bool)

Last drains the stream and returns the last item it observed.

func (Stream[T]) LastErr 

func (s Stream[T]) LastErr() (T, bool, error)

LastErr drains the stream, returns the last item it observed, and returns the final error state.

func (Stream[T]) Max 

func (s Stream[T]) Max(less func(T, T) bool) (T, bool)

Max drains the stream and returns the greatest item according to less.

func (Stream[T]) MaxErr 

func (s Stream[T]) MaxErr(less func(T, T) bool) (T, bool, error)

MaxErr drains the stream, returns the greatest item according to less, and returns the final error state.

func (Stream[T]) Min 

func (s Stream[T]) Min(less func(T, T) bool) (T, bool)

Min drains the stream and returns the least item according to less.

func (Stream[T]) MinErr 

func (s Stream[T]) MinErr(less func(T, T) bool) (T, bool, error)

MinErr drains the stream, returns the least item according to less, and returns the final error state.

func (Stream[T]) NoneMatch 

func (s Stream[T]) NoneMatch(predicate func(T) bool) bool

NoneMatch reports whether no item satisfies predicate. It drains the remainder of the stream after the first match so delayed fail-fast errors can still surface.

func (Stream[T]) NoneMatchErr 

func (s Stream[T]) NoneMatchErr(predicate func(T) bool) (bool, error)

NoneMatchErr reports whether no item satisfies predicate and returns the current error state when it short-circuits.

func (Stream[T]) Parallel 

func (s Stream[T]) Parallel(fn func(T), opts ...Option)

Parallel applies fn to each item using the same worker machinery as the transform operators and panics on fail-fast errors.

func (Stream[T]) ParallelErr 

func (s Stream[T]) ParallelErr(fn func(T) error, opts ...Option) error

ParallelErr applies fn to each item using worker options and returns the final error state.

func (Stream[T]) Reverse 

func (s Stream[T]) Reverse() Stream[T]

Reverse drains s, reverses the collected items, and replays them as a new stream.

func (Stream[T]) Skip 

func (s Stream[T]) Skip(n int64) Stream[T]

Skip discards the first n items from s and emits the remainder.

func (Stream[T]) Sort 

func (s Stream[T]) Sort(less func(T, T) bool) Stream[T]

Sort drains s, sorts all items with less, and then replays them as a new stream.

func (Stream[T]) Tail 

func (s Stream[T]) Tail(n int64) Stream[T]

Tail returns a stream containing the last n items from s in original order.

func (Stream[T]) Tap added in v0.1.5 

func (s Stream[T]) Tap(
	ctx context.Context,
	fn func(context.Context, T) error,
	opts ...Option,
) Stream[T]

Tap runs fn for each item in s and re-emits the original item when fn succeeds.

func (Stream[T]) Through added in v0.1.5 

func (s Stream[T]) Through(
	ctx context.Context,
	fn func(context.Context, T) T,
	opts ...Option,
) Stream[T]

Through applies fn to each item in s and returns another Stream[T]. It exists to make same-type stage segments read fluently in a chain.

func (Stream[T]) ThroughErr added in v0.1.5 

func (s Stream[T]) ThroughErr(
	ctx context.Context,
	fn func(context.Context, T) (T, error),
	opts ...Option,
) Stream[T]

ThroughErr applies fn to each item in s, records returned worker errors in the stream state, and returns another Stream[T]. It exists to make same-type stage segments read fluently in a chain.

type TimeoutOption 

type TimeoutOption func() context.Context

TimeoutOption supplies the parent context for a timeout call.

func WithContext 

func WithContext(ctx context.Context) TimeoutOption

WithContext sets the parent context for a timeout call.

type WorkerError 

type WorkerError struct {
	Err error
}

WorkerError wraps a failure produced by one worker goroutine.

func (*WorkerError) Error 

func (e *WorkerError) Error() string

Error returns the wrapped worker error text.

func (*WorkerError) Unwrap 

func (e *WorkerError) Unwrap() error

Unwrap returns the underlying worker error.

Source Files

View all Source files

Directories

Path Synopsis
examples
http_image_pipeline command
internal
collections
config
control
runtime
state
streaming

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.