flow

func NewChannelSink ¶

func NewChannelSink[T any](ch chan<- T) ChannelSink[T]

NewChannelSink creates a new ChannelSink from the provided channel.

func Merge ¶

func Merge(sources ...Source) Flow

Merge combines multiple sources into a single Flow. It processes each Source sequentially, emitting all values from one source before moving to the next.

The function will process sources in order until either:

• All sources are exhausted
• The context is cancelled
• The consumer stops accepting values

func NewFromChannel ¶

func NewFromChannel[T any](ch <-chan T) Flow

NewFromChannel creates a Flow from a channel. It provides a way to convert any typed channel for use in flow-based processing pipelines.

The flow will continue reading from the channel until it is closed or the context is cancelled.

Example:

// Create a channel and flow
ch := make(chan int)
flow := NewFromChannel(ch)

// Send values in a separate goroutine
go func() {
    defer close(ch)
    for i := 1; i <= 5; i++ {
        ch <- i
    }
}()

// Process values from the flow
ctx := context.Background()
for item := range flow.Stream(ctx) {
    fmt.Println(item)
}

func NewFromItems ¶

func NewFromItems[T any](items ...T) Flow

NewFromItems creates a Flow from a variadic list of items. It provides a convenient way to create a flow from a known set of values of any type T.

Example:

// Create a Flow from integers
flow := NewFromItems(1, 2, 3, 4, 5)

// Use the flow
ctx := context.Background()
for item := range flow.Stream(ctx) {
    fmt.Println(item)
}

func NewFromIterable ¶

func NewFromIterable(it Iterable) Flow

NewFromIterable creates a Flow from an Iterable source. It handles the conversion of the iteration-based interface to a stream-based flow.

func NewFromRange ¶

func NewFromRange(start, end, step int) Flow

NewFromRange creates a Flow that emits a sequence of integers from start (inclusive) to end (exclusive), incrementing by the specified step value.

The function will stop emitting values when either:

• The sequence reaches or exceeds the end value
• The context is cancelled
• The consumer stops accepting values

func NewFromSource ¶

func NewFromSource(s Source) Flow

NewFromSource creates a Flow from an existing Source, allowing it to be used in flow-based processing pipelines.

Example:

// Create a simple source
source := SourceFunc(func(ctx context.Context) Stream {
    return func(yield func(any) bool) {
        for i := 1; i <= 3; i++ {
            if ctx.Err() != nil {
                return
            }
            if !yield(i) {
                return
            }
        }
    }
})

// Create a Flow from the source
flow := NewFromSource(source)

// Use the flow in a pipeline
ctx := context.Background()
for item := range flow.Stream(ctx) {
    fmt.Println(item)
}

func NewFromSourceFunc ¶

func NewFromSourceFunc(fn func(context.Context) Stream) Flow

NewFromSourceFunc creates a Flow from a function that produces a Stream. This is a convenience function that wraps a stream-producing function into a flow, making it easier to create flows without explicitly implementing the Source interface.

Example:

// Create a Flow that emits numbers 1 through 5
flow := NewFromSourceFunc(func(ctx context.Context) Stream {
    return func(yield func(any) bool) {
        for i := 1; i <= 5; i++ {
            // Check for context cancellation
            if ctx.Err() != nil {
                return
            }
            // Emit value, stop if consumer requests
            if !yield(i) {
                return
            }
        }
    }
})

// Use the flow
ctx := context.Background()
for item := range flow.Stream(ctx) {
    fmt.Println(item)
}

func NewFromTicker ¶

func NewFromTicker(interval time.Duration) Flow

NewFromTicker creates a Flow that emits timestamps at specified intervals. It continues to emit timestamps until either the context is cancelled or the consumer stops accepting values.

The timestamps are of type time.Time and represent the actual time when the interval elapsed. Note that the actual intervals may be slightly longer than specified due to system scheduling and processing delays.

func Discard ¶

func Discard() Sinker

Discard creates a Sinker that consumes and discards all items from a Stream. It respects context cancellation while efficiently draining the stream without allocating memory for the items.

The function is useful for:

• Draining streams when values are not needed
• Testing flow processing without storing results
• Completing flow processing when only side effects matter

func NewSliceSink ¶

func NewSliceSink[T any](s []T) *SliceSink[T]

NewSliceSink creates a new SliceSink with the provided initial slice. If nil is provided, a new empty slice will be created.

func Empty ¶

func Empty() Stream

Empty creates a Stream that produces no values. This can be useful as a default or placeholder source in situations where a no-op source is needed.

Example:

// Use Empty as a fallback flow
var flow Source
if someCondition {
    flow = NewFromItems(1, 2, 3)
} else {
    flow = Empty()
}

// The empty flow will produce no values
ctx := context.Background()
for item := range flow.Stream(ctx) {
    // This loop will not execute for Empty flow
}

func Chunk ¶

func Chunk[T any](n int) Transform

Chunk creates a new Transform that groups items into fixed-size chunks. The last chunk may contain fewer items if the stream length is not divisible by n. Each chunk is emitted as a slice of items.

Example:

// Group numbers into chunks of 2
flow := NewFromItems(1, 2, 3, 4, 5).
    Transform(Chunk[int](2))

result, err := Collect[[]int](ctx, flow)
if err != nil {
    log.Fatal(err)
}
fmt.Println(result) // [[1, 2], [3, 4], [5]]

func Filter ¶

func Filter[T any](fn func(context.Context, T) bool) Transform

Filter creates a new Transform that selectively includes items from the Stream based on the provided predicate function. Only items that cause the predicate to return true are included in the output stream.

Example:

// Keep only even numbers
flow := NewFromItems(1, 2, 3, 4, 5).
    Transform(Filter(func(ctx context.Context, n int) bool {
        return n%2 == 0
    }))

result, err := Collect[int](ctx, flow)
if err != nil {
    log.Fatal(err)
}
fmt.Println(result) // [2, 4]

func FlatMap ¶

func FlatMap[T any, U any](fn func(context.Context, T) ([]U, error)) Transform

FlatMap creates a new Transform component that transforms items using the provided mapping function. Each input item is transformed into a slice of output items, which are then sent individually downstream.

func Flatten ¶

func Flatten[T []U, U any]() Transform

Flatten creates a new Transform component that receives a batch of items and flattens it, sending each item from the upstream batch individually downstream. It is the semantic equivalent of:

FlatMap(func(in []T) []T { return in })

For example, to handle an upstream csv in indexed row order, call:

// flatten a 2-D array of strings
Flatten[[][]string]() // [][]string - Flatten -> []string

func Join ¶

func Join(pipes ...Transform) Transform

Join combines multiple transformers into a single Transform by composing them in order. Each transformers's output becomes the input to the next transformer in the sequence.

Join returns Pass if no transformers are provided.

func Keep ¶

func Keep(n int) Transform

Keep creates a new Transform that limits the stream to at most n items from the input stream. Once n items have been processed, any remaining input items are ignored.

Example:

// Keep first 3 items
f := NewFromItems(1, 2, 3, 4, 5).
    Transform(Keep(3))

results, err := Collect[int](ctx, f)
if err != nil {
    log.Fatal(err)
}
fmt.Println(results) // [1, 2, 3]

func KeepDistinct ¶

func KeepDistinct[T any](opts ...func(*KeepDistinctOptions[T])) Transform

KeepDistinct creates a new Transform that filters out duplicate elements from the Stream. It accepts optional configuration functions to customize the Flow's behavior. By default, it uses the string representation of elements as the uniqueness key.

Example:

// Remove duplicate numbers
flow := NewFromItems(1, 2, 2, 3, 3, 4).
    Transform(KeepDistinct[int]())

result, err := Collect[int](ctx, flow)
if err != nil {
    log.Fatal(err)
}
fmt.Println(result) // [1, 2, 3, 4]

func KeepFirst ¶

func KeepFirst() Transform

KeepFirst creates a new Transform that takes only the first item from the stream and terminates processing. It is semantically equivalent to Keep(1).

Example:

// Get first item from stream
f := NewFromItems(1, 2, 3, 4, 5).
    Transform(KeepFirst())

result, err := Collect[int](ctx, f)
if err != nil {
    log.Fatal(err)
}
fmt.Println(result) // [1]

func KeepIf ¶

func KeepIf[T any](fn func(context.Context, T) bool) Transform

KeepIf creates a new Transform that selectively includes items from the stream based on the provided predicate function. It is an alias for Filter.

Example:

// Keep only positive numbers
flow := NewFromItems(-2, -1, 0, 1, 2).
    Transform(KeepIf(func(ctx context.Context, n int) bool {
        return n > 0
    }))

result, err := Collect[int](ctx, flow)
if err != nil {
    log.Fatal(err)
}
fmt.Println(result) // [1, 2]

func KeepLast ¶

func KeepLast() Transform

KeepLast creates a new Transform that yields only the last item from the stream. If the stream is empty, no items are yielded.

Example:

// Get the last item from a stream of numbers
flow := NewFromItems(1, 2, 3, 4, 5).
    Transform(KeepLast())

result, err := Collect[int](ctx, f)
if err != nil {
    log.Fatal(err)
}
fmt.Println(result) // [5]

func Limit ¶

func Limit(limit rate.Limit, burst int) Transform

Limit creates a new Transform that applies rate limiting to the stream using the token bucket algorithm. Items are processed according to the specified rate limit and burst size.

Example:

// Process items at a rate of 10 per second with max burst of 20
flow := NewFromItems(1, 2, 3, 4, 5).
    Transform(Limit(rate.Limit(10), 20))

result, err := Collect[int](ctx, flow)
if err != nil {
    log.Fatal(err)
}

func Map ¶

func Map[T any, U any](fn func(context.Context, T) (U, error)) Transform

Map creates a new Transform component that transforms items using the provided function. Each input item is transformed from type T to type U using function fn.

func OmitIf ¶

func OmitIf[T any](fn func(context.Context, T) bool) Transform

OmitIf creates a new Transform that selectively excludes items from the Stream based on the provided predicate function. It is the inverse of KeepIf.

Example:

// Skip negative numbers
flow := NewFromItems(-2, -1, 0, 1, 2).
    Transform(OmitIf(func(ctx context.Context, n int) bool {
        return n < 0
    }))

result, err := Collect[int](ctx, flow)
if err != nil {
    log.Fatal(err)
}
fmt.Println(result) // [0, 1, 2]

func ParallelMap ¶

func ParallelMap[T any, U any](n int, fn func(context.Context, T) (U, error)) Transform

ParallelMap creates a new Transform that transforms items in parallel using multiple workers. Each input item is transformed from type T to type U using function fn, with processing distributed across n concurrent workers.

Example:

// Process numbers in parallel
flow := NewFromItems(1, 2, 3, 4, 5).
    Transform(ParallelMap[int, int](3, func(ctx context.Context, n int) (int, error) {
        time.Sleep(time.Second) // Simulate work
        return n * 2, nil
    }))

result, err := Collect[int](ctx, flow)
if err != nil {
    log.Fatal(err)
}
fmt.Println(result) // [2, 4, 6, 8, 10] <- order is not guaranteed

func Pass ¶

func Pass() Transform

Pass creates a new Transform component that forwards items without modification. This can be useful for debugging or when you need to maintain the Flow structure without processing.

func Reduce ¶

func Reduce[T any](fn func(ctx context.Context, acc T, item T) (T, error)) Transform

Reduce creates a new Transform component that combines multiple items into one using the provided reduction function. Each input item is combined with the accumulator to produce a new accumulated value.

Example:

// Join strings with separator
flow := NewFromItems("Hello", "World", "!").
    Transform(
        Reduce(func(ctx context.Context, acc, item string) (string, error) {
            if acc == "" {
                return item, nil
            }
            return acc + " " + item, nil
        }),
    )

result, err := Collect[string](ctx, flow)
// Result: ["Hello", "Hello World", "Hello World !"]

func Skip ¶

func Skip(n int) Transform

Skip creates a new Transform that skips the first n items from the input stream and forwards all remaining items. If n is less than or equal to 0, no items are dropped.

Example:

// Skip first 2 items and process the rest
f := NewFromItems(1, 2, 3, 4, 5).
    Transform(Skip(2))

result, err := Collect[int](ctx, f)
if err != nil {
    log.Fatal(err)
}
fmt.Println(result) // [3, 4, 5]

func SlidingWindow ¶

func SlidingWindow[T any](opts ...func(*SlidingWindowOptions)) Transform

SlidingWindow creates a new Transform component that groups items using a sliding window. The window moves forward by StepSize items after each batch is emitted.

Example:

// Create windows of size 3, sliding by 1
flow := NewFromItems(1, 2, 3, 4, 5).
    Transform(SlidingWindow[int](func(opts *SlidingWindowOptions) {
        opts.WindowSize = 3
        opts.StepSize = 1
    }))

result, err := Collect[[]int](ctx, flow)
if err != nil {
    log.Fatal(err)
}
fmt.Println(result) // [[1, 2, 3], [2, 3, 4], [3, 4, 5]]