flx

package module
v0.1.9 Latest Latest
Warning

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

 Go to latest Published: Apr 10, 2026 License: MIT Imports: 13 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"

worker 并发控制相关 API 现在推荐从下面的公开子包导入：

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

pipeline coordinator 与观测 API 推荐从下面两个公开子包导入：

import "github.com/ezra-sullivan/flx/pipeline/coordinator"
import "github.com/ezra-sullivan/flx/pipeline/observe"

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

快速示例

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

实战示例

仓库里现在把图片处理场景拆成三条主 example，再加一个独立 retry example：

  • 主线 examples 走真实 picsum.photos 列图和下载
  • imgproc_retry 单独保持本地可控，方便稳定演示重试行为

1. 图片处理流水线

go run ./examples/imgproc_pipeline

这条 example 聚焦业务 pipeline 本身：

  • Picsum 分页列图
  • 下载、缩放、水印、本地保存
  • Stage 组合版作为默认写法
  • 原生 flx API 版作为对照写法

可显式切到 native 对照版：

go run ./examples/imgproc_pipeline -mode native

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

2. 图片处理 Observe 示例

go run ./examples/imgproc_observe

这条 example 专门讲 observe：

  • 使用同一条 Picsum 图片处理业务线
  • 只展示 stage/link snapshot
  • 不接 PipelineCoordinator

完整说明见 doc/examples/imgproc-observe.md。

3. 图片处理 Coordinator 示例

go run ./examples/imgproc_coordinator

这条 example 专门讲 coordinator：

  • 使用同一条 Picsum 图片处理业务线
  • 给 resize stage 配 budget
  • 展示 snapshot、decision、resource observer 和外部 Tick() loop

完整说明见 doc/examples/imgproc-coordinator.md。

4. 图片处理重试示例

go run ./examples/imgproc_retry

这个例子固定演示：

  • 本地可控 transport 上的下载重试
  • 某些下载在第 2 次或第 3 次尝试成功
  • 某些下载在 3 次预算内全部失败
  • WithOnRetry warning 是怎么跟最终 item 结果对应起来的

完整说明见 doc/examples/imgproc-retry.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

并发与错误控制

  • control.WithWorkers
  • control.WithUnlimitedWorkers
  • control.WithDynamicWorkers
  • control.WithForcedDynamicWorkers
  • control.WithInterruptibleWorkers（兼容别名）
  • control.WithErrorStrategy
  • control.ErrorStrategyFailFast
  • control.ErrorStrategyCollect
  • control.ErrorStrategyContinue
  • control.ErrorStrategyLogAndContinue（兼容别名，已废弃）

独立工具函数

  • Parallel / ParallelErr / ParallelWithErrorStrategy
  • DoWithRetry / DoWithRetryCtx / retry observer via WithOnRetry
  • DoWithTimeout / DoWithTimeoutCtx / timeout late-panic observer via WithTimeoutLatePanicObserver

内存边界

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

动态并发

优雅缩容

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

需要额外导入：github.com/ezra-sullivan/flx/pipeline/control

ctrl := control.NewConcurrencyController(4)

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

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

out.Done()

强制缩容

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

需要额外导入：github.com/ezra-sullivan/flx/pipeline/control

ctx := context.Background()
ctrl := control.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+"!")
	},
	control.WithForcedDynamicWorkers(ctrl),
)

ctrl.SetWorkers(1)
out.Done()

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

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

Pipeline Coordinator

pipeline/coordinator 用来把动态 worker 从“手动 SetWorkers(...)”提升到“基于 stage/link/resource 信号的显式 Tick() 决策”。

最小用法通常是：

  • 给动态 stage 挂 coordinator.WithCoordinator(...)
  • 给 stage 命名 coordinator.WithStageName(...)
  • 给可调 stage 配预算 coordinator.WithStageBudget(...)
  • 在外部控制循环里周期性调用 PipelineCoordinator.Snapshot() 和 PipelineCoordinator.Tick()

当前适用边界：

  • 一个 PipelineCoordinator 实例按“一次 pipeline run”使用；它会在实例生命周期内保留最近一次看到的 stage/link/control 状态
  • Links 视图当前按 fromStage 聚合，只保留每个 stage 的一条 outbound link snapshot；线性链路场景是当前主目标，fan-out 还没有单独建模

PipelineCoordinatorPolicy 当前几个核心字段：

  • ScaleUpStep 每次扩容最多增加多少 worker。
  • ScaleDownStep 每次缩容最多减少多少 worker。
  • ScaleUpCooldown 同一 stage 两次扩容之间的最小时间间隔。
  • ScaleDownCooldown 同一 stage 两次缩容之间的最小时间间隔。
  • ScaleUpHysteresis 扩容信号需要连续满足多少个 Tick() 才会真正扩容。
  • ScaleDownHysteresis 缩容信号需要连续满足多少个 Tick() 才会真正缩容。

默认语义是保守兼容的：

  • ScaleUpStep / ScaleDownStep 零值按 1
  • ScaleUpCooldown / ScaleDownCooldown 零值表示不额外等待
  • ScaleUpHysteresis / ScaleDownHysteresis 零值按 1

当前 Tick() 的策略重点：

  • 下游 incoming link backlog 优先触发下游扩容
  • stage 自身 backlog 次之
  • warning 级资源压力会 brake scale-up
  • critical 级资源压力只会对满足 activeWorkers < currentWorkers 的空闲 stage 施加 shrink bias
  • budget_min 属于硬纠偏，不受 cooldown / hysteresis 限制

资源采样契约补充：

  • Snapshot() 和 Tick() 都会各自独立轮询 resource observers
  • 同一个 coordinator 实例会串行化这些轮询，避免同一个 observer 被 Snapshot() / Tick() 并发重入
  • 但相邻的 Snapshot() / Tick() 不保证使用同一份 resource sample；如果把同一个 observer 共享给多个 coordinator，仍需 observer 自己保证同步

示例：

pipelineCoordinator := coordinator.NewPipelineCoordinator(
	coordinator.PipelineCoordinatorPolicy{
		ScaleUpStep:         1,
		ScaleDownStep:       1,
		ScaleUpCooldown:     500 * time.Millisecond,
		ScaleDownCooldown:   750 * time.Millisecond,
		ScaleUpHysteresis:   1,
		ScaleDownHysteresis: 2,
	},
	coordinator.WithResourceObserver(myResourceObserver),
)

完整接法可直接看 doc/examples/imgproc-coordinator.md。

错误处理建议

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

  • fail-fast：尽快取消当前操作，并在终结阶段暴露错误
  • collect：继续执行，最后合并错误
  • continue：继续执行，但不记录到 stream state

control.ErrorStrategyLogAndContinue 仍保留为兼容别名，但不再内建输出日志；如需日志，请在业务代码里自行记录。

业务代码里，如果你需要稳定的最终错误边界，优先使用会完整消费 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 这条主错误通道。

与 go-zero 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 -> control.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 (
	// 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 control.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 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 RetryEvent added in v0.1.9 

type RetryEvent struct {
	Attempt     int
	MaxAttempts int
	Retry       int
	MaxRetries  int
	NextAttempt int
	NextDelay   time.Duration
	Err         error
}

RetryEvent describes one failed attempt that will be retried. Attempt fields describe overall attempts including the first try, while Retry fields describe the upcoming retry ordinal only.

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 WithOnRetry added in v0.1.9 

func WithOnRetry(fn func(RetryEvent)) RetryOption

WithOnRetry registers one callback that is invoked after a failed attempt when another retry attempt is still scheduled. Panics in the callback are recovered and ignored so observability hooks do not change retry behavior.

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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 ...control.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 TimeoutLatePanicEvent added in v0.1.9 

type TimeoutLatePanicEvent struct {
	Panic any
	Stack string
}

TimeoutLatePanicEvent reports a panic that happened after DoWithTimeout or DoWithTimeoutCtx had already returned, so the panic could not be rethrown to the finished caller anymore.

type TimeoutOption 

type TimeoutOption func(*timeoutOptions)

TimeoutOption mutates the behavior of one timeout call.

func WithContext 

func WithContext(ctx context.Context) TimeoutOption

WithContext sets the parent context for a timeout call.

func WithTimeoutLatePanicObserver added in v0.1.9 

func WithTimeoutLatePanicObserver(fn func(TimeoutLatePanicEvent)) TimeoutOption

WithTimeoutLatePanicObserver registers one callback that is invoked when a timeout callback panics after DoWithTimeout or DoWithTimeoutCtx has already returned. Panics in the callback are recovered and ignored so observability hooks do not change timeout behavior.

Source Files

View all Source files

Directories

Path Synopsis
examples
imgproc_coordinator command
imgproc_observe command
imgproc_pipeline command
imgproc_retry command
internal/imgproc
internal
collections
config
control
coordinator
link
metrics
resource
runtime
state
streaming
pipeline
control
Package control exposes worker and concurrency control primitives used by flx stream, stage, and parallel execution APIs.
 Package control exposes worker and concurrency control primitives used by flx stream, stage, and parallel execution APIs.
coordinator
Package coordinator exposes stage identity, snapshot, and Tick-facing pipeline control options.
 Package coordinator exposes stage identity, snapshot, and Tick-facing pipeline control options.
observe
Package observe exposes helpers for inspecting pipeline stage, link, and resource health.
 Package observe exposes helpers for inspecting pipeline stage, link, and resource health.

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.