scoped

package module
v0.0.0-...-e5b6f20 Latest Latest
Warning

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

 Go to latest Published: Apr 28, 2026 License: MIT Imports: 8 Imported by: 0

README

scoped

Go Reference CI Go Report Card

Structured concurrency primitives for Go — run goroutines with clear lifecycles, coordinated cancellation, and composable error handling.

Install

go get github.com/baxromumarov/scoped

Core Concepts

Run — Scoped lifecycle

Run creates a scope, executes your function, and waits for all spawned tasks:

err := scoped.Run(ctx, func(sp scoped.Spawner) {
    sp.Go("fetch", func(ctx context.Context) error {
        return fetch(ctx)
    })
    sp.Spawn("process", func(ctx context.Context, sub scoped.Spawner) error {
        sub.Go("step-1", step1)
        return nil
    })
})

Use Go for simple tasks and Spawn when the task needs to spawn sub-tasks.

New / Wait — Manual lifecycle

For cases where spawning happens outside a callback:

sc, sp := scoped.New(ctx, scoped.WithPolicy(scoped.Collect))
sp.Go("task", func(ctx context.Context) error {
    return doWork(ctx)
})
err := sc.Wait()

WaitTimeout adds a deadline — it returns context.DeadlineExceeded if tasks don't finish in time without cancelling the scope:

err := sc.WaitTimeout(5 * time.Second)
Error Policies
  • FailFast (default) — First error cancels all siblings. Wait() returns that error.
  • Collect — All errors are collected. Wait() returns all via errors.Join.
scoped.Run(ctx, fn, scoped.WithPolicy(scoped.Collect))

Use WithMaxErrors(n) to cap stored errors in Collect mode, preventing unbounded memory growth. Check sc.DroppedErrors() for the overflow count.

Error Introspection

All task errors are wrapped in *TaskError for attribution:

err := scoped.Run(ctx, fn, scoped.WithPolicy(scoped.Collect))

for _, te := range scoped.AllTaskErrors(err) {
    fmt.Printf("task %q failed: %v\n", te.Task.Name, te.Err)
}

// Or check individual errors:
if info, ok := scoped.TaskOf(err); ok {
    fmt.Println("failed task:", info.Name)
}
cause := scoped.CauseOf(err) // unwrap TaskError to get the root cause
Bounded Concurrency

Limit the number of concurrent goroutines:

scoped.Run(ctx, fn, scoped.WithLimit(10))
Panic Recovery

By default, panics are re-raised in Wait(). Use WithPanicAsError() to convert them to *PanicError errors with full stack traces:

scoped.Run(ctx, fn, scoped.WithPanicAsError())
Panic Contract

scoped intentionally panics for programmer misuse (invalid arguments or invalid lifecycle usage), for example:

  • Calling Spawn on a closed spawner.
  • Calling stream Next() concurrently.
  • Passing invalid option values (negative limits, nil required callbacks, etc.).

These panics are API contract checks, not runtime data-path errors. Task-function panics are handled separately:

  • Default behavior: panic is captured and re-raised at Wait().
  • WithPanicAsError(): panic is converted to *PanicError and returned as an error.

Helpers

ForEachSlice — Parallel iteration
err := scoped.ForEachSlice(ctx, urls, func(ctx context.Context, u string) error {
    return fetch(ctx, u)
}, scoped.WithLimit(10))
MapSlice — Parallel map with results

Results are returned in input order. Use Collect policy for partial results:

results, err := scoped.MapSlice(ctx, items, func(ctx context.Context, item T) (R, error) {
    return transform(ctx, item)
}, scoped.WithLimit(5))

for _, r := range results {
    if r.Err != nil {
        // handle per-item error
    }
    // use r.Value
}
SpawnResult — Typed async result
r := scoped.SpawnResult(sp, "compute", func(ctx context.Context) (int, error) {
    return expensiveCalc(ctx)
})
val, err := r.Wait()
SpawnTimeout — Per-task deadline
scoped.SpawnTimeout(sp, "slow-op", 5*time.Second,
    func(ctx context.Context, _ scoped.Spawner) error {
        return slowOperation(ctx) // cancelled after 5s
    },
)
SpawnRetry — Retry with backoff

Retries up to n times with exponential backoff. Stops immediately on context cancellation:

scoped.SpawnRetry(sp, "flaky-api", 3, 100*time.Millisecond,
    func(ctx context.Context, _ scoped.Spawner) error {
        return callFlakyAPI(ctx)
    },
)
// Backoff: 100ms, 200ms, 400ms
Race — First successful result

Run multiple tasks concurrently, return the first successful result, cancel the rest:

val, err := scoped.Race(ctx,
    func(ctx context.Context) (string, error) { return fetchFromA(ctx) },
    func(ctx context.Context) (string, error) { return fetchFromB(ctx) },
    func(ctx context.Context) (string, error) { return fetchFromC(ctx) },
)
SpawnScope — Sub-scopes

Run a group of tasks with an independent error policy inside a parent scope:

scoped.Run(ctx, func(sp scoped.Spawner) {
    scoped.SpawnScope(sp, "batch", func(sub scoped.Spawner) {
        for _, item := range items {
            sub.Go(item.Name, item.Process)
        }
    }, scoped.WithPolicy(scoped.Collect))
})

Semaphore

Standalone weighted semaphore for use outside scopes:

sem := scoped.NewSemaphore(5)

if err := sem.Acquire(ctx); err != nil {
    return err // context cancelled
}
defer sem.Release()

// ... do bounded work

Worker Pool

Fixed-size worker pool with queue:

pool := scoped.NewPool(ctx, 4, scoped.WithQueueSize(100))

pool.Submit(func() error {
    return processJob(job)
})

if ok := pool.TrySubmit(fn); !ok {
    // queue full or pool closed
}

err := pool.Close() // waits for in-flight tasks, returns joined errors

Observability

Lifecycle hooks
scoped.Run(ctx, fn,
    scoped.WithOnEvent(func(e scoped.TaskEvent) {
        log.Printf("[%s] task=%s err=%v dur=%s",
            e.Kind, e.Task.Name, e.Err, e.Duration)
    }),
)

Event kinds: EventStarted, EventDone, EventErrored, EventPanicked, EventCancelled.

Legacy per-phase hooks are also available via WithOnStart and WithOnDone.

Periodic metrics
scoped.Run(ctx, fn,
    scoped.WithOnMetrics(time.Second, func(m scoped.Metrics) {
        fmt.Printf("active=%d completed=%d errored=%d longest=%s\n",
            m.ActiveTasks, m.Completed, m.Errored, m.LongestActive)
    }),
)
Stall detection

Detect tasks running longer than a threshold (purely observational — does not cancel):

scoped.Run(ctx, fn,
    scoped.WithStallDetector(5*time.Second, func(rt scoped.RunningTask) {
        log.Printf("STALLED: task=%q running for %s", rt.Name, rt.Elapsed)
    }),
)
Scope snapshots

Get a point-in-time view of all running tasks:

sc, sp := scoped.New(ctx, scoped.WithTaskTracking())
// ... spawn tasks ...
snap := sc.Snapshot()
for _, rt := range snap.RunningTasks {
    fmt.Printf("  %s running for %s\n", rt.Name, rt.Elapsed)
}
fmt.Printf("longest active: %s\n", snap.LongestActive)
Pool monitoring
pool := scoped.NewPool(ctx, 4,
    scoped.WithPoolMetrics(time.Second, func(s scoped.PoolStats) {
        fmt.Printf("submitted=%d inflight=%d queue=%d\n",
            s.Submitted, s.InFlight, s.QueueDepth)
    }),
)

// Or poll on demand:
stats := pool.Stats()
Stream monitoring

Streams track items, errors, and throughput automatically:

s := scoped.FromSlice(items)
// ... consume stream ...
stats := s.Stats()
fmt.Printf("read=%d errors=%d throughput=%.0f items/sec\n",
    stats.ItemsRead, stats.Errors, stats.Throughput)

For per-item event hooks:

observed := scoped.Observe(stream, func(e scoped.StreamEvent[int]) {
    if e.Err != nil {
        log.Printf("stream error at seq %d: %v", e.Seq, e.Err)
    }
})
Live flow visualizer (real-time UI)

Use viz to inspect spawned goroutine hierarchy and channel data flow in real time:

tr := viz.New()
go func() { _ = tr.Serve(":8080") }()

err := tr.Run(ctx, func(sp scoped.Spawner) {
    jobs := viz.NewChannel[int](tr, "jobs", 16)
    // use jobs.Send / jobs.Recv inside traced tasks
    // ...
})

Open http://localhost:8080 to see:

  • parent/child task hierarchy
  • task lifecycle transitions (running, done, errored, cancelled, panicked)
  • live channel send/receive flow graph
  • real-time transfer timeline

See examples/flowviz/main.go for a runnable starter demo and examples/flowviz_full/main.go for a full-featured end-to-end demonstration.

Streams

Pull-based, composable data pipelines with lazy evaluation.

Creating streams
s := scoped.FromSlice([]int{1, 2, 3, 4, 5})
s := scoped.FromChan(ch)
s := scoped.NewStream(func(ctx context.Context) (int, error) {
    // custom iterator — return io.EOF when done
})
s := scoped.Empty[int]()              // immediate EOF
s := scoped.Repeat("hello", 5)       // emit "hello" 5 times (-1 = infinite)
s := scoped.Generate(1, func(v int) int { return v * 2 }) // 1, 2, 4, 8, ...
Chaining operations
results, err := scoped.FromSlice(items).
    Filter(func(v int) bool { return v > 0 }).
    Skip(10).
    Take(100).
    TakeWhile(func(v int) bool { return v < 500 }).
    DropWhile(func(v int) bool { return v < 50 }).
    Peek(func(v int) { log.Println(v) }).
    ToSlice(ctx)
Type-changing transforms

Go does not support generic methods on generic types, so cross-type operations are top-level functions:

mapped := scoped.Map(stream, func(ctx context.Context, v int) (string, error) {
    return strconv.Itoa(v), nil
})

batched := scoped.Batch(stream, 10)          // *Stream[[]int]
reduced, _ := scoped.Reduce(ctx, stream, 0,  // fold
    func(acc, v int) int { return acc + v },
)
flat := scoped.FlatMap(stream, func(ctx context.Context, v int) *scoped.Stream[string] {
    return scoped.FromSlice(strings.Split(fmt.Sprint(v), ""))
})
unique := scoped.Distinct(stream)            // requires comparable

scan := scoped.Scan(stream, 0,               // running fold
    func(acc, v int) int { return acc + v },
)

zipped := scoped.Zip(streamA, streamB)       // *Stream[Pair[A, B]]
ParallelMap — Concurrent stream transformation
out := scoped.ParallelMap(ctx, sp, src, scoped.StreamOptions{
    MaxWorkers: 4,
    Ordered:    true,
    MaxPending: 16, // backpressure buffer for ordered mode
}, func(ctx context.Context, v int) (string, error) {
    return transform(ctx, v)
})

results, err := out.ToSlice(ctx)
Terminal operations
Method Description
ToSlice(ctx) Collect all items into a slice
ForEach(ctx, fn) Apply a function to each item
Count(ctx) Count items in the stream
First(ctx) Return the first item
Last(ctx) Consume all, return the last item
Any(ctx, fn) True if any item matches predicate
All(ctx, fn) True if all items match predicate
ToChanScope(sp) Bridge to a channel within a scope
Stop() Release resources (safe to call multiple times)

Channel Utilities (chanx)

The chanx subpackage provides context-aware channel operations:

go get github.com/baxromumarov/scoped/chanx
Send and Receive
Function Description
Send / Recv Context-aware send and receive
TrySend / TryRecv Non-blocking send and receive
SendTimeout / RecvTimeout Send and receive with a deadline
SendBatch / RecvBatch Batch send and receive
Fan-in, Fan-out, and Broadcasting
Function Description
Merge Combine multiple channels into one
FanOut Distribute to N workers (round-robin)
Tee Broadcast to N unbuffered consumers (all must read)
Broadcast Broadcast with buffered outputs (tolerates slow consumers)
Transformation and Filtering
Function Description
Map Transform values through a pipeline
Filter Pass only values matching a predicate
Take Forward first n items then close
Skip Drop first n items then forward rest
Scan Running accumulation of input values
Partition Split by predicate into two channels (both must be read concurrently)
Rate Limiting and Batching
Function Description
Throttle Token-bucket rate limiting
Buffer Batch by size or timeout
BufferWithReason Batch with flush reason (FlushSize, FlushTimeout, FlushClose)
Timing
Function Description
Debounce Emit last value after a quiet period
Window Time-based grouping (Tumbling or Sliding mode)
Combining and Selection
Function Description
Zip Pair values from two channels into Pair[A, B]
First Race: first value from any channel
Lifecycle
Function Description
OrDone Wrap a channel with context cancellation
Drain Discard remaining values to unblock producers
Closable Idempotent-close channel wrapper (panics become ErrClosed)

Benchmarks

Comparison against raw goroutines, golang.org/x/sync/errgroup, and sourcegraph/conc (AMD Ryzen 7, Go 1.26):

Overhead per task spawn
Implementation 10 tasks 100 tasks 1000 tasks
Raw goroutine+WG 1.7 us 17.5 us 167 us
errgroup 1.8 us 16.7 us 181 us
conc 1.8 us 17.0 us 172 us
scoped 3.2 us 24.7 us 227 us
ForEach (10 items, light work)
Implementation ns/op allocs/op
scoped ForEach 7,552 29
conc Iterator 8,402 14
raw goroutines 297,209 1,002
errgroup 337,836 2,004
Hot-path operations (zero allocations)
Operation ns/op
Semaphore Acquire/Release ~30 ns
Pool Submit ~120 ns
chanx.TrySend ~9 ns
chanx.TryRecv ~13 ns
chanx.Send (ctx-aware) ~23 ns

Run benchmarks locally:

make bench

License

Licensed under MIT. See LICENSE.

Release Process

Release checklist and tagging steps are in RELEASE.md.

Documentation

Overview

Package scoped provides structured concurrency primitives for Go.

Structured concurrency ensures that concurrent tasks have well-defined lifecycles: they are spawned and joined within a clear scope, preventing goroutine leaks, orphaned tasks, and unpredictable control flow.

Running Tasks

The primary entry point is Run, which creates a scope, executes a function that spawns tasks via Spawner, and waits for all tasks to complete before returning: 

err := scoped.Run(ctx, func(sp scoped.Spawner) {
    sp.Go("fetch", func(ctx context.Context) error {
        return fetch(ctx)
    })
    sp.Spawn("process", func(ctx context.Context, sub scoped.Spawner) error {
        sub.Go("step-1", step1)
        return nil
    })
})

Use Spawner.Go for simple tasks and Spawner.Spawn when the task needs to spawn sub-tasks of its own.

For manual lifecycle control, New returns a Scope and root Spawner separately. The caller must call Scope.Wait to finalize. Scope.WaitTimeout adds a deadline to finalization.

Error Policies

Error policies control how the scope reacts to task failures:

All task errors are wrapped in *TaskError for attribution. Use IsTaskError, TaskOf, CauseOf, and AllTaskErrors to inspect them.

Helpers

Convenience functions for common patterns:

  • ForEachSlice: apply a function to every item in a slice concurrently.
  • MapSlice: transform every item concurrently, preserving order.
  • SpawnResult: spawn a task that returns a typed value via Result.
  • SpawnTimeout: spawn a task with a per-task deadline.
  • SpawnRetry: spawn a task with exponential-backoff retries.
  • SpawnScope: spawn a sub-scope as a single task, allowing hierarchical error handling with independent policies.
  • Race: run multiple tasks concurrently and return the first successful result, cancelling the rest.

Bounded Concurrency

Use WithLimit to restrict the number of goroutines executing concurrently within a scope. Tasks beyond the limit wait for a slot, respecting context cancellation while waiting.

For standalone use outside scopes, Semaphore provides a weighted semaphore with Semaphore.Acquire, Semaphore.TryAcquire, and Semaphore.Release.

Worker Pool

Pool provides a reusable fixed-size worker pool. Tasks are submitted via Pool.Submit (blocking) or Pool.TrySubmit (non-blocking) and processed by a fixed number of goroutines. Call Pool.Close to drain the queue and collect errors.

Panic Recovery

By default, a panic in any task is captured with its full stack trace and re-raised in Scope.Wait. Use WithPanicAsError to convert panics to *PanicError values and return them as regular errors instead.

Panic Contract

This package intentionally panics for programmer misuse (invalid options, nil required callbacks, invalid lifecycle calls such as spawning after shutdown, or concurrent Stream.Next calls). These panics enforce API invariants and are considered contract violations, not recoverable runtime data errors.

Observability

Register hooks for task lifecycle events:

Pool exposes Pool.Stats returning a PoolStats snapshot, and WithPoolMetrics for periodic pool metrics callbacks.

Stream tracks items read, errors, and timing automatically. Stream.Stats returns a StreamStats snapshot including throughput. Observe wraps a stream with a per-item StreamEvent callback.

Streams

Stream provides a pull-based, composable data pipeline. Create streams with NewStream, FromSlice, FromSliceRef, FromChan, Empty, Repeat, or Generate. Chains of Stream.Filter, Stream.Take, Stream.Skip, Stream.Peek, Stream.TakeWhile, Stream.DropWhile, Map, Batch, FlatMap, Distinct, Scan, and Zip are evaluated lazily. Reduce folds a stream into a single value.

ParallelMap processes items concurrently with optional ordering and backpressure via StreamOptions.MaxPending.

Terminal methods (Stream.ToSlice, Stream.ForEach, Stream.Count) return partial results alongside any error, following io.Reader conventions. Stream.ToChanScope bridges a stream to a channel within a scope. Stream errors are aggregated via errors.Join.

Streams are single-consumer; concurrent Stream.Next calls panic.

Spawner Lifetime

Each task function receives a child Spawner that is valid only for the duration of the task. Storing the child Spawner and calling Spawn on it after the task returns will panic. This is by design: structured concurrency requires that all child tasks are scoped to their parent.

Channel Utilities

The github.com/baxromumarov/scoped/chanx subpackage provides context-aware channel operations (Send, Recv, TrySend, TryRecv, SendTimeout, RecvTimeout, SendBatch, RecvBatch), fan-in/fan-out patterns (Merge, Tee, FanOut, Broadcast), transformation pipelines (Map, Filter), rate limiting (Throttle), batching (Buffer, BufferWithReason), timing (Debounce, Window), combining (Zip, First), and an idempotent-close channel wrapper (Closable).

Index

Examples

Constants

This section is empty.

Variables

View Source 
var ErrNilPoolTask = errors.New("scoped: task must not be nil")

ErrNilPoolTask is returned by Pool.Submit when a nil task is submitted.

View Source 
var ErrPoolClosed = errors.New("scoped: pool is closed")

ErrPoolClosed is returned by Pool.Submit when the pool has been closed.

View Source 
var (
	// ErrStreamGap is returned when an ordered stream terminates with missing items.
	ErrStreamGap = fmt.Errorf("stream terminated with missing results (gap)")
)

Functions

func CauseOf 

func CauseOf(err error) error

CauseOf unwraps the first *TaskError in err's chain and returns its underlying cause. If err is not a TaskError, it is returned as-is. Returns nil if err is nil.

func ForEachSlice 

func ForEachSlice[T any](
	ctx context.Context,
	items []T,
	fn func(ctx context.Context, item T) error,
	opts ...Option,
) error

ForEachSlice executes fn for each item in the slice concurrently, using the provided options to control concurrency and error policy.

This is a convenience wrapper around Run and [Scope.Spawn]. 

err := scoped.ForEachSlice(ctx, URLs, func(ctx context.Context, u string) error {
    return fetch(ctx, u)
}, scoped.WithLimit(10))
Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	items := []string{"a", "b", "c"}
	err := scoped.ForEachSlice(context.Background(), items, func(ctx context.Context, s string) error {
		// process each item concurrently
		return nil
	}, scoped.WithLimit(2))
	if err != nil {
		fmt.Println("error:", err)
	}
	fmt.Println("done")
}

Output:
done

func IsTaskError 

func IsTaskError(err error) bool

IsTaskError reports whether err (or any error in its chain) is a *TaskError.

func New 

func New(parent context.Context, opts ...Option) (*Scope, Spawner)

New creates a Scope and root Spawner for manual lifecycle control. The caller must call Scope.Wait to finalize the scope and collect errors.

Prefer Run for most use cases; use New when you need to pass the Spawner across function boundaries or integrate with existing lifecycle management.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	sc, sp := scoped.New(context.Background())
	sp.Spawn("work", func(ctx context.Context, _ scoped.Spawner) error {
		fmt.Println("doing work")
		return nil
	})
	err := sc.Wait()
	if err != nil {
		fmt.Println("error:", err)
	}
}

Output:
doing work

func Race 

func Race[T any](
	ctx context.Context,
	tasks ...func(context.Context) (T, error),
) (T, error)

Race runs all tasks concurrently and returns the result of the first task to succeed (return nil error). The contexts of remaining tasks are cancelled immediately upon the first success.

If all tasks fail, Race returns the zero value and the last error observed. If ctx is cancelled before any task succeeds, Race returns ctx.Err().

If tasks is empty, Race returns (zero, nil).

Race panics if any element of tasks is nil.

func Reduce 

func Reduce[T, R any](ctx context.Context, s *Stream[T], initial R, fn func(R, T) R) (R, error)

Reduce folds the stream into a single value using the given accumulator function. On error, the partial accumulation so far is returned alongside the error.

func Run 

func Run(parent context.Context, fn func(sp Spawner), opts ...Option) (err error)

Run creates a Scope, invokes fn with its root Spawner, then waits for every spawned task to complete. It returns the aggregated error according to the configured Policy (default FailFast).

Run is the primary entry point for structured concurrency. The scope is automatically finalized when fn returns, so no explicit cleanup is needed.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	err := scoped.Run(context.Background(), func(sp scoped.Spawner) {
		sp.Spawn("greet", func(ctx context.Context, _ scoped.Spawner) error {
			fmt.Println("hello from task")
			return nil
		})
	})
	if err != nil {
		fmt.Println("error:", err)
	}
}

Output:
hello from task
Example (Collect) 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	err := scoped.Run(context.Background(), func(sp scoped.Spawner) {
		for i := range 3 {
			sp.Spawn(fmt.Sprintf("task-%d", i), func(ctx context.Context, _ scoped.Spawner) error {
				if i == 1 {
					return fmt.Errorf("task %d failed", i)
				}
				return nil
			})
		}
	}, scoped.WithPolicy(scoped.Collect))

	if err != nil {
		fmt.Println("got errors")
	}
}

Output:
got errors

func SpawnRetry 

func SpawnRetry(
	sp Spawner,
	name string,
	n int,
	backoff time.Duration,
	fn TaskFunc,
)

SpawnRetry spawns a task that retries on failure up to n times with exponential backoff starting from the given base duration. The backoff doubles on each retry: base, base*2, base*4, ... Retries stop immediately if the context is cancelled.

SpawnRetry panics if n < 0 or backoff <= 0.

func SpawnScope 

func SpawnScope(sp Spawner, name string, fn func(sp Spawner), opts ...Option)

SpawnScope spawns a sub-scope as a single task within the parent scope. The sub-scope has its own error policy and options, allowing hierarchical error handling. The sub-scope's aggregated error (if any) is propagated to the parent scope as the task's error.

This enables patterns like running a batch of tasks with Collect policy inside a parent scope using FailFast: 

scoped.Run(
	ctx,
	func(sp scoped.Spawner) {
    scoped.SpawnScope(sp, "batch", func(sub scoped.Spawner) {
        for _, item := range items {
            sub.Spawn(item.Name, item.Process)
        }
    }, scoped.WithPolicy(scoped.Collect))
})

func SpawnTimeout 

func SpawnTimeout(sp Spawner, name string, d time.Duration, fn TaskFunc)

SpawnTimeout spawns a task with a per-task deadline. If the task does not complete within d, its context is cancelled with context.DeadlineExceeded.

The timeout only affects this task's context; it does not cancel the scope.

Types

type EventKind 

type EventKind int

EventKind identifies the lifecycle stage of a TaskEvent. 

const (
	// EventStarted is emitted when a task begins executing.
	EventStarted EventKind = iota
	// EventDone is emitted when a task completes successfully (Err is nil).
	EventDone
	// EventErrored is emitted when a task returns a non-nil error.
	EventErrored
	// EventPanicked is emitted when a task panics.
	EventPanicked
	// EventCancelled is emitted when a task's context was cancelled
	// before or during execution.
	EventCancelled
)

func (EventKind) String 

func (k EventKind) String() string

type ItemResult 

type ItemResult[R any] struct {
	Value R
	Err   error
}

ItemResult holds the outcome of processing an individual item in MapSlice. In Collect mode, function-level failures are captured in Err and successful items still have Value populated.

func MapSlice 

func MapSlice[T, R any](
	ctx context.Context,
	items []T,
	fn func(ctx context.Context, item T) (R, error),
	opts ...Option,
) (
	[]ItemResult[R],
	error,
)

MapSlice executes fn for each item concurrently and collects the results in the same order as the input slice. It uses FailFast policy by default; pass Collect Policy to gather partial results and errors that happened during execution.

Result semantics:

  • FailFast: the first function error fails the whole call and MapSlice returns nil results with that error.

  • Collect: function errors are captured per item in ItemResult.Err, and MapSlice still returns the partial results slice.

  • The outer returned error is reserved for scope/infrastructure failures (for example context cancellation, panic-as-error, etc.).

    prices, err := scoped.MapSlice(ctx, products, func(ctx context.Context, p Product) (float64, error) { return fetchPrice(ctx, p) }, scoped.WithLimit(5))

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	items := []int{1, 2, 3}
	results, err := scoped.MapSlice(context.Background(), items, func(ctx context.Context, v int) (int, error) {
		return v * 10, nil
	})
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	for _, r := range results {
		fmt.Println(r.Value)
	}
}

Output:
10
20
30

type Metrics 

type Metrics struct {
	TotalSpawned  int64
	ActiveTasks   int64
	Completed     int64
	Errored       int64
	Panicked      int64
	Cancelled     int64
	LongestActive time.Duration // zero if task tracking is not enabled
}

Metrics provides aggregated counters for a scope's lifecycle.

type Option 

type Option func(*config)

Option configures a Scope.

func WithLimit 

func WithLimit(n int) Option

WithLimit sets the maximum number of goroutines that can execute concurrently within the scope. Tasks beyond the limit block until a slot becomes available or the context is canceled.

A limit of zero (the default) means unlimited concurrency. WithLimit panics if n is negative.

func WithMaxErrors 

func WithMaxErrors(n int) Option

WithMaxErrors sets the maximum number of errors stored in Collect mode. When the limit is reached, subsequent errors are still counted but not stored, preventing unbounded memory growth in high-volume scenarios.

A value of zero (the default) means unlimited error collection. This option has no effect in FailFast mode. WithMaxErrors panics if n is negative.

func WithOnDone 

func WithOnDone(fn func(TaskInfo, error, time.Duration)) Option

WithOnDone registers a hook invoked when each task finishes. The hook receives the task's error (nil on success) and wall-clock duration. The hook runs inside the task's goroutine after the task function returns.

func WithOnEvent 

func WithOnEvent(fn func(TaskEvent)) Option

WithOnEvent registers a unified lifecycle hook that receives a TaskEvent for every task state change: started, done, errored, panicked, and cancelled. The hook runs inside the task's goroutine.

WithOnEvent can be used alongside WithOnStart and WithOnDone; all registered hooks will fire.

func WithOnMetrics 

func WithOnMetrics(interval time.Duration, fn func(Metrics)) Option

WithOnMetrics registers a periodic metrics callback that fires every interval. The callback receives a snapshot of current scope counters.

Panics if interval <= 0 or fn is nil.

func WithOnStart 

func WithOnStart(fn func(TaskInfo)) Option

WithOnStart registers a hook invoked when each task begins executing. The hook runs inside the task's goroutine before the task function.

func WithPanicAsError 

func WithPanicAsError() Option

WithPanicAsError converts panics in child tasks to *PanicError values returned as regular errors, instead of re-raising them in Scope.Wait.

func WithPolicy 

func WithPolicy(p Policy) Option

WithPolicy sets the error handling policy for the scope. It panics if p is not a known Policy value.

func WithStallDetector 

func WithStallDetector(threshold time.Duration, fn func(RunningTask)) Option

WithStallDetector registers a periodic check for tasks running longer than threshold. The callback receives each stalled task with its elapsed duration. The check runs every threshold/2 interval (minimum 10ms).

This is purely observational — unlike SpawnTimeout, it does not cancel stalled tasks. Implicitly enables task tracking (see WithTaskTracking).

Panics if threshold <= 0 or fn is nil.

func WithTaskTracking 

func WithTaskTracking() Option

WithTaskTracking enables per-task tracking so that Scope.Snapshot includes RunningTask entries and ScopeSnapshot.LongestActive duration. This has a small overhead (mutex acquisition per task start/end).

type Pair 

type Pair[A, B any] struct {
	First  A
	Second B
}

Pair holds two values paired from two streams. It is used by Zip.

type PanicError 

type PanicError struct {
	// Value is the original value passed to panic().
	Value any

	// Stack is the goroutine stack trace at the point of panic.
	Stack string
}

PanicError wraps a recovered panic value together with the goroutine stack trace captured at the point of the panic.

When WithPanicAsError is set, panics in tasks are converted to *PanicError and returned as regular errors. Otherwise, the *PanicError is re-raised via panic in Scope.Wait.

func (*PanicError) Error 

func (e *PanicError) Error() string

Error returns a human-readable representation of the panic, including the value and the full stack trace.

func (*PanicError) Unwrap 

func (e *PanicError) Unwrap() error

Unwrap returns the panic value as an error, if it implements the error interface, or nil otherwise.

type Policy 

type Policy int

Policy determines how a Scope handles errors from child tasks. 

const (
	// FailFast cancels all sibling tasks when the first error occurs.
	// [Scope.Wait] returns the first error encountered.
	FailFast Policy = iota

	// Collect gathers all errors without cancelling siblings.
	// [Scope.Wait] returns all errors joined via [errors.Join].
	Collect
)

type Pool 

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

Pool is a reusable worker pool. Tasks are submitted via Submit and processed by a fixed number of worker goroutines.

func NewPool 

func NewPool(
	ctx context.Context,
	n int,
	opts ...PoolOption,
) *Pool

NewPool creates a pool with n worker goroutines. Workers start immediately and process tasks until Pool.Close is called. Panics if n <= 0.

func (*Pool) Close 

func (p *Pool) Close() error

Close stops accepting new tasks and waits for in-flight tasks to finish. Returns the joined errors from all failed tasks. Safe to call multiple times; subsequent calls return the same result.

func (*Pool) Stats 

func (p *Pool) Stats() PoolStats

Stats returns a point-in-time snapshot of pool activity. Safe to call concurrently.

func (*Pool) Submit 

func (p *Pool) Submit(fn func() error) (err error)

Submit submits a task to the pool. It blocks if the queue is full. Returns ErrNilPoolTask if fn is nil. Returns ErrPoolClosed if the pool has been closed. Returns ctx.Err() if the pool's context is cancelled.

func (*Pool) TrySubmit 

func (p *Pool) TrySubmit(fn func() error) (submitted bool)

TrySubmit attempts to submit without blocking. Returns false if fn is nil, the queue is full, the pool is closed, or the pool's context has been cancelled.

type PoolOption 

type PoolOption func(*poolConfig)

PoolOption configures a Pool.

func WithPoolMetrics 

func WithPoolMetrics(interval time.Duration, fn func(PoolStats)) PoolOption

WithPoolMetrics registers a periodic pool metrics callback that fires every interval. The callback receives a snapshot of current pool counters.

Panics if interval <= 0 or fn is nil.

func WithQueueSize 

func WithQueueSize(size int) PoolOption

WithQueueSize sets the task queue buffer size. Default is n * 2.

type PoolStats 

type PoolStats struct {
	Submitted  int64 // total tasks submitted
	Completed  int64 // tasks finished (success + error)
	Errored    int64 // tasks that returned non-nil error
	InFlight   int64 // tasks currently executing
	QueueDepth int   // tasks waiting in the queue
	Workers    int   // worker count (fixed at creation)
}

PoolStats provides a point-in-time snapshot of pool activity.

type Result 

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

Result holds the outcome of an asynchronous task that produces a typed value. Create one via SpawnResult.

func SpawnResult 

func SpawnResult[T any](
	sp Spawner,
	name string,
	fn func(ctx context.Context) (T, error),
) *Result[T]

SpawnResult spawns a named task that returns a typed value and wraps the outcome in a Result. The task runs within the given Scope, inheriting its lifecycle and error policy. 

 Example:
	r := scoped.SpawnResult(s, "compute", func(ctx context.Context) (int, error) {
    	return expensiveCalc(ctx)
	})
	val, err := r.Wait()
Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	err := scoped.Run(context.Background(), func(sp scoped.Spawner) {
		r := scoped.SpawnResult(sp, "compute", func(ctx context.Context) (int, error) {
			return 42, nil
		})
		val, err := r.Wait()
		if err != nil {
			fmt.Println("error:", err)
			return
		}
		fmt.Println("result:", val)
	})
	if err != nil {
		fmt.Println("scope error:", err)
	}
}

Output:
result: 42

func (*Result[T]) Done 

func (r *Result[T]) Done() <-chan ResultValue[T]

Done returns a channel that receives exactly one ResultValue and then closes.

func (*Result[T]) Wait 

func (r *Result[T]) Wait() (T, error)

Wait blocks until the task completes and returns its value and error.

type ResultValue 

type ResultValue[T any] struct {
	Val T
	Err error
}

ResultValue holds the value and error from a completed Result task.

type RunningTask 

type RunningTask struct {
	Name    string
	Started time.Time
	Elapsed time.Duration // populated at snapshot time
}

RunningTask describes a task currently executing within a scope.

type Scope 

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

Scope wraps the internal scope state and exposes lifecycle and observability methods. Create one via New; finalize with Scope.Wait.

func (*Scope) ActiveTasks 

func (sc *Scope) ActiveTasks() int64

ActiveTasks returns the number of tasks currently executing within the scope.

func (*Scope) Cancel 

func (sc *Scope) Cancel(err error)

Cancel cancels the scope's context with the given cause, signaling all tasks to stop. Subsequent calls have no additional effect on the context.

func (*Scope) Context 

func (sc *Scope) Context() context.Context

Context returns the scope's context, which is cancelled when the scope finalizes or is explicitly cancelled via Scope.Cancel.

func (*Scope) DroppedErrors 

func (sc *Scope) DroppedErrors() int

DroppedErrors returns the number of errors that were not stored because the WithMaxErrors limit was reached. This is only meaningful in Collect mode.

func (*Scope) Snapshot 

func (sc *Scope) Snapshot() ScopeSnapshot

Snapshot returns a point-in-time view of the scope's state. If task tracking is not enabled (via WithStallDetector or WithTaskTracking), RunningTasks will be nil and LongestActive will be zero.

func (*Scope) TotalSpawned 

func (sc *Scope) TotalSpawned() int64

TotalSpawned returns the total number of tasks that have been spawned within the scope, including those that have already completed.

func (*Scope) Wait 

func (sc *Scope) Wait() error

Wait closes the root Spawner, waits for all spawned tasks to complete, and returns the aggregated error. If a task panicked and WithPanicAsError was not set, Wait re-panics with the captured *PanicError.

Wait is idempotent; subsequent calls return the same result.

func (*Scope) WaitTimeout 

func (sc *Scope) WaitTimeout(d time.Duration) error

WaitTimeout is like [Wait] but returns context.DeadlineExceeded if the timeout expires before all tasks finish. The scope is NOT cancelled on timeout — call Scope.Cancel explicitly if needed.

WaitTimeout is safe to call alongside [Wait]; finalization runs at most once.

type ScopeSnapshot 

type ScopeSnapshot struct {
	Metrics       Metrics
	RunningTasks  []RunningTask
	LongestActive time.Duration
}

ScopeSnapshot is a point-in-time view of a scope's state.

type Semaphore 

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

Semaphore is a weighted semaphore for bounding concurrency. It is context-aware: Acquire unblocks if the context is cancelled.

func NewSemaphore 

func NewSemaphore(n int) *Semaphore

NewSemaphore creates a semaphore with the given capacity. Panics if n <= 0.

func (*Semaphore) Acquire 

func (s *Semaphore) Acquire(ctx context.Context) error

Acquire blocks until a slot is available or ctx is cancelled. Returns ctx.Err() on cancellation, nil on success.

func (*Semaphore) Available 

func (s *Semaphore) Available() int

Available returns the number of available slots. The value may be stale in concurrent contexts.

func (*Semaphore) Release 

func (s *Semaphore) Release()

Release releases a slot. Panics if more slots are released than acquired.

func (*Semaphore) TryAcquire 

func (s *Semaphore) TryAcquire() bool

TryAcquire attempts to acquire a slot without blocking. Returns true if acquired, false otherwise.

type Spawner 

type Spawner interface {
	// Go starts a new concurrent task with the given name.
	// This is the simplified form for tasks that don't need to spawn sub-tasks.
	//
	//	sp.Go("fetch", func(ctx context.Context) error {
	//	    return fetch(ctx, url)
	//	})
	Go(name string, fn func(ctx context.Context) error)

	// Spawn starts a new concurrent task with the given name.
	//
	// The task function receives a child [Spawner] allowing it to create sub-tasks.
	// The child Spawner is only valid for the lifetime of the task function; storing
	// it and calling Spawn after the task returns will panic.
	//
	// Use [Spawner.Go] for tasks that don't need nested spawning.
	Spawn(name string, fn TaskFunc)
}

Spawner allows spawning concurrent tasks into a scope.

type Stream 

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

Stream represents a structured, pull-based data stream.

Streams are single-consumer. Next() and other terminal methods must not be called concurrently. Concurrent calls to Next() will panic.

func Batch 

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

Batch groups items into slices of size n.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	s := scoped.FromSlice([]int{1, 2, 3, 4, 5})
	batches := scoped.Batch(s, 2)
	res, err := batches.ToSlice(context.Background())
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Println(res)
}

Output:
[[1 2] [3 4] [5]]

func Distinct 

func Distinct[T comparable](s *Stream[T]) *Stream[T]

Distinct returns a stream that suppresses duplicate items. Items are compared by value equality. T must be comparable.

func Empty 

func Empty[T any]() *Stream[T]

Empty returns a stream that immediately signals io.EOF. It never yields any items.

func FlatMap 

func FlatMap[A, B any](s *Stream[A], fn func(context.Context, A) *Stream[B]) *Stream[B]

FlatMap transforms each item in the source stream into a sub-stream and concatenates all sub-streams sequentially. Each sub-stream is fully consumed before moving to the next source item.

Nil sub-streams returned by fn are skipped.

func FromChan 

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

FromChan creates a stream from a channel.

func FromSlice 

func FromSlice[T any](items []T) *Stream[T]

FromSlice creates a stream from a slice. The slice is copied internally; the caller may safely modify the original after this call.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	s := scoped.FromSlice([]int{1, 2, 3, 4, 5})
	res, err := s.Filter(func(v int) bool {
		return v%2 == 0
	}).ToSlice(context.Background())
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Println(res)
}

Output:
[2 4]

func FromSliceRef 

func FromSliceRef[T any](items []T) *Stream[T]

FromSliceRef creates a stream that reads directly from the provided slice without copying. The caller must not modify the slice after this call.

Use this instead of FromSlice in performance-critical paths where the allocation cost of copying matters and ownership can be guaranteed.

func FromSliceUnsafe deprecated 

func FromSliceUnsafe[T any](items []T) *Stream[T]

FromSliceUnsafe is a deprecated alias for FromSliceRef.

Deprecated: Use FromSliceRef instead.

func Generate 

func Generate[T any](seed T, fn func(T) T) *Stream[T]

Generate returns an infinite stream starting from seed, applying fn to produce each subsequent value: seed, fn(seed), fn(fn(seed)), ...

The stream is infinite; use Stream.Take, Stream.TakeWhile, or context cancellation to bound it.

Panics if fn is nil.

func Map 

func Map[A, B any](s *Stream[A], fn func(context.Context, A) (B, error)) *Stream[B]

Map transforms a stream using a function. Note: This is a function and not a method because Go does not support generic methods on generic types.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	s := scoped.FromSlice([]int{1, 2, 3})
	doubled := scoped.Map(s, func(ctx context.Context, v int) (string, error) {
		return fmt.Sprintf("%d", v*2), nil
	})
	res, err := doubled.ToSlice(context.Background())
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Println(res)
}

Output:
[2 4 6]

func NewStream 

func NewStream[T any](next func(context.Context) (T, error)) *Stream[T]

NewStream creates a new stream from an iterator function.

func Observe 

func Observe[T any](s *Stream[T], fn func(StreamEvent[T])) *Stream[T]

Observe wraps a stream with an event callback that fires after every Next() call. The callback runs synchronously in the consumer's goroutine.

Panics if s or fn is nil.

func ParallelMap 

func ParallelMap[A, B any](
	ctx context.Context,
	sp Spawner,
	src *Stream[A],
	opts StreamOptions,
	fn func(context.Context, A) (B, error),
) *Stream[B]

ParallelMap transforms a stream concurrently.

Workers are managed internally and do NOT use sp.Spawn, so the results channel can be consumed inside the same Run callback without deadlock. The dispatcher goroutine IS spawned via sp.Spawn so it respects the scope lifecycle.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	err := scoped.Run(context.Background(), func(sp scoped.Spawner) {
		src := scoped.FromSlice([]int{1, 2, 3, 4, 5})
		pm := scoped.ParallelMap(context.Background(), sp, src,
			scoped.StreamOptions{MaxWorkers: 3, Ordered: true},
			func(ctx context.Context, v int) (int, error) {
				return v * 10, nil
			},
		)
		res, err := pm.ToSlice(context.Background())
		if err != nil {
			fmt.Println("error:", err)
			return
		}
		fmt.Println(res)
	})
	if err != nil {
		fmt.Println("scope error:", err)
	}
}

Output:
[10 20 30 40 50]
Example (Unordered) 
package main

import (
	"context"
	"fmt"
	"sort"

	"github.com/baxromumarov/scoped"
)

func main() {
	err := scoped.Run(context.Background(), func(sp scoped.Spawner) {
		src := scoped.FromSlice([]int{3, 1, 2})
		pm := scoped.ParallelMap(context.Background(), sp, src,
			scoped.StreamOptions{MaxWorkers: 3},
			func(ctx context.Context, v int) (int, error) {
				return v * 10, nil
			},
		)
		res, err := pm.ToSlice(context.Background())
		if err != nil {
			fmt.Println("error:", err)
			return
		}
		sort.Ints(res)
		fmt.Println(res)
	})
	if err != nil {
		fmt.Println("scope error:", err)
	}
}

Output:
[10 20 30]

func Repeat 

func Repeat[T any](val T, n int) *Stream[T]

Repeat returns a stream that emits val exactly n times. If n is negative, the stream repeats indefinitely until the context is cancelled or the consumer stops reading.

func Scan 

func Scan[T, R any](s *Stream[T], initial R, fn func(R, T) R) *Stream[R]

Scan returns a stream that applies fn cumulatively to each item, emitting each intermediate accumulation. The first emitted value is fn(initial, firstItem).

This is the streaming counterpart of Reduce: Reduce produces a single final value, while Scan produces a stream of running values.

Panics if s is nil or fn is nil.

func Zip 

func Zip[A, B any](a *Stream[A], b *Stream[B]) *Stream[Pair[A, B]]

Zip pairs items from two streams element-by-element. The resulting stream emits Pair values and stops as soon as either input stream is exhausted (EOF). When one stream ends, the other is stopped immediately.

Both streams are read sequentially (a first, then b) within each Next call — this is safe because streams are single-consumer.

Panics if a or b is nil.

func (*Stream[T]) All 

func (s *Stream[T]) All(ctx context.Context, fn func(T) bool) (bool, error)

All returns true if every item in the stream satisfies fn. It stops reading as soon as a non-matching item is found. Returns (true, nil) for an empty stream (vacuous truth).

Panics if fn is nil.

func (*Stream[T]) Any 

func (s *Stream[T]) Any(ctx context.Context, fn func(T) bool) (bool, error)

Any returns true if any item in the stream satisfies fn. It stops reading as soon as a match is found. Returns (false, nil) for an empty stream.

Panics if fn is nil.

func (*Stream[T]) Count 

func (s *Stream[T]) Count(ctx context.Context) (int, error)

Count counts the number of items in the stream.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	count, err := scoped.FromSlice([]int{1, 2, 3, 4, 5}).Count(context.Background())
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Println(count)
}

Output:
5

func (*Stream[T]) DropWhile 

func (s *Stream[T]) DropWhile(fn func(T) bool) *Stream[T]

DropWhile returns a stream that skips items while fn returns true, then emits all remaining items unconditionally.

Panics if fn is nil.

func (*Stream[T]) Err 

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

Err returns the final aggregated error after completion.

func (*Stream[T]) Filter 

func (s *Stream[T]) Filter(fn func(T) bool) *Stream[T]

Filter returns a stream that only emits items for which fn returns true.

func (*Stream[T]) First 

func (s *Stream[T]) First(ctx context.Context) (T, error)

First returns the first item in the stream and stops the stream. If the stream is empty, it returns the zero value and nil error.

func (*Stream[T]) ForEach 

func (s *Stream[T]) ForEach(ctx context.Context, fn func(T) error) error

ForEach applies a function to each item in the stream.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	err := scoped.FromSlice([]int{1, 2, 3}).ForEach(context.Background(), func(v int) error {
		fmt.Println(v)
		return nil
	})
	if err != nil {
		fmt.Println("error:", err)
	}
}

Output:
1
2
3

func (*Stream[T]) Last 

func (s *Stream[T]) Last(ctx context.Context) (T, error)

Last consumes the entire stream and returns the last item. If the stream is empty, it returns the zero value and nil error.

func (*Stream[T]) Next 

func (s *Stream[T]) Next(ctx context.Context) (T, error)

Next returns the next item in the stream. Returns io.EOF when the stream is exhausted.

Next panics if called concurrently from multiple goroutines.

func (*Stream[T]) Peek 

func (s *Stream[T]) Peek(fn func(T)) *Stream[T]

Peek allows inspecting items as they pass through the stream.

func (*Stream[T]) Skip 

func (s *Stream[T]) Skip(n int) *Stream[T]

Skip skips the first n items in the stream.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	s := scoped.FromSlice([]int{1, 2, 3, 4, 5}).Skip(2)
	res, err := s.ToSlice(context.Background())
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Println(res)
}

Output:
[3 4 5]

func (*Stream[T]) Stats 

func (s *Stream[T]) Stats() StreamStats

Stats returns the current observability counters for the stream. Safe to call concurrently with Next() from a monitoring goroutine.

func (*Stream[T]) Stop 

func (s *Stream[T]) Stop()

Stop terminates the stream and releases associated resources. Safe to call multiple times and concurrently.

func (*Stream[T]) Take 

func (s *Stream[T]) Take(n int) *Stream[T]

Take limits the stream to n items.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/baxromumarov/scoped"
)

func main() {
	s := scoped.FromSlice([]int{1, 2, 3, 4, 5}).Take(3)
	res, err := s.ToSlice(context.Background())
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Println(res)
}

Output:
[1 2 3]

func (*Stream[T]) TakeWhile 

func (s *Stream[T]) TakeWhile(fn func(T) bool) *Stream[T]

TakeWhile returns a stream that emits items as long as fn returns true. Once fn returns false, the upstream is stopped and the stream signals EOF.

Panics if fn is nil.

func (*Stream[T]) ToChanScope 

func (s *Stream[T]) ToChanScope(sp Spawner) (<-chan T, <-chan error)

ToChanScope sends all items in the stream to a channel within a Scope. The goroutine is managed by the scope and will stop when the stream is exhausted or the scope is canceled.

func (*Stream[T]) ToSlice 

func (s *Stream[T]) ToSlice(ctx context.Context) (items []T, err error)

ToSlice collects all items in the stream into a slice. On error, any items collected before the error are returned alongside it, following the io.Reader convention.

type StreamEvent 

type StreamEvent[T any] struct {
	Item     T
	Err      error         // non-nil for errors (not EOF)
	Duration time.Duration // time spent in the underlying Next call
	EOF      bool          // true when stream is exhausted
	Seq      int64         // 0-based sequence number
}

StreamEvent describes an event within a stream's lifecycle. It is passed to the callback registered via Observe.

type StreamOptions 

type StreamOptions struct {
	MaxWorkers int
	BufferSize int
	Ordered    bool
	// MaxPending limits the number of out-of-order results buffered in ordered mode.
	// When reached, the dispatcher blocks until the consumer catches up.
	// 0 means default (MaxWorkers * 2). Ignored when Ordered is false.
	MaxPending int
}

StreamOptions configures parallel stream processing.

type StreamStats 

type StreamStats struct {
	ItemsRead  int64     // items successfully read
	Errors     int64     // non-EOF errors encountered
	StartTime  time.Time // zero if Next() was never called
	LastReadAt time.Time // zero if no items have been read
	Throughput float64   // items per second since StartTime (0 if no items)
}

StreamStats provides observability counters for a stream.

type TaskError 

type TaskError struct {
	Task TaskInfo
	Err  error
}

TaskError wraps an error together with the TaskInfo of the task that produced it. Scope error aggregation wraps every task failure in a TaskError so callers can attribute errors to specific tasks.

func AllTaskErrors 

func AllTaskErrors(err error) []*TaskError

AllTaskErrors recursively collects every *TaskError from err's chain, including errors wrapped via errors.Join. Returns nil if none are found.

Example 
package main

import (
	"context"
	"fmt"
	"sort"

	"github.com/baxromumarov/scoped"
)

func main() {
	err := scoped.Run(context.Background(), func(sp scoped.Spawner) {
		sp.Spawn("a", func(ctx context.Context, _ scoped.Spawner) error {
			return fmt.Errorf("error a")
		})
		sp.Spawn("b", func(ctx context.Context, _ scoped.Spawner) error {
			return fmt.Errorf("error b")
		})
	}, scoped.WithPolicy(scoped.Collect))

	taskErrs := scoped.AllTaskErrors(err)
	names := make([]string, len(taskErrs))
	for i, te := range taskErrs {
		names[i] = te.Task.Name
	}
	sort.Strings(names)
	fmt.Println("failed tasks:", names)
}

Output:
failed tasks: [a b]

func (*TaskError) Error 

func (e *TaskError) Error() string

func (*TaskError) Unwrap 

func (e *TaskError) Unwrap() error

type TaskEvent 

type TaskEvent struct {
	Kind     EventKind
	Task     TaskInfo
	Err      error         // non-nil for EventErrored and EventPanicked
	Duration time.Duration // wall-clock time; zero for EventStarted
}

TaskEvent describes a lifecycle event for a task. It is passed to the callback registered via WithOnEvent.

type TaskFunc 

type TaskFunc func(ctx context.Context, sp Spawner) error

TaskFunc is the signature for a task function running within a scope. It receives a context (cancelled when the scope ends) and a Spawner to spawn sub-tasks.

type TaskInfo 

type TaskInfo struct {
	Name string
}

TaskInfo provides metadata about a running task. It is passed to observability hooks registered via WithOnStart and WithOnDone.

func TaskOf 

func TaskOf(err error) (TaskInfo, bool)

TaskOf extracts the TaskInfo from the first *TaskError in err's chain. Returns false if no TaskError is found.

Source Files

View all Source files

Directories

Path Synopsis
Package chanx provides context-aware, goroutine-safe channel utilities.
 Package chanx provides context-aware, goroutine-safe channel utilities.
examples
advanced command
Package main demonstrates advanced scoped patterns: SpawnResult, SpawnTimeout, observability hooks, error introspection, context cancellation, MapSlice, ForEachSlice, and chanx channel utilities.
 Package main demonstrates advanced scoped patterns: SpawnResult, SpawnTimeout, observability hooks, error introspection, context cancellation, MapSlice, ForEachSlice, and chanx channel utilities.
basic command
Package main demonstrates core scoped concurrency primitives: Run, Spawn, error policies, concurrency limits, and panic recovery.
 Package main demonstrates core scoped concurrency primitives: Run, Spawn, error policies, concurrency limits, and panic recovery.
flowviz command
flowviz_full command
monitoring command
Package main demonstrates the monitoring and observability features of scoped: scope metrics, stall detection, task snapshots, pool stats, and stream monitoring.
 Package main demonstrates the monitoring and observability features of scoped: scope metrics, stall detection, task snapshots, pool stats, and stream monitoring.
streams command
Package main demonstrates the Stream API: constructors, transforms, terminal operations, batching, and parallel processing.
 Package main demonstrates the Stream API: constructors, transforms, terminal operations, batching, and parallel processing.
Package viz provides a real-time visualization tool for scoped programs.
 Package viz provides a real-time visualization tool for scoped programs.

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.