flowkit

README

FlowKit

A lightweight generic workflow and pipeline toolkit for Go.

Features

• Generic pipelines
• Producer / Transformer / Consumer model
• Stream processing
• Functional operators
• Iterator support
• Zero external dependencies
• Built with Go generics

Installation

go get github.com/bhaskar253/flowkit

Quick Example

package main

import (
    "fmt"

    "github.com/bhaskar253/flowkit"
)

func main() {
    count := 1
    pipeline := flowkit.New(

        func() (int, bool) {
            if count > 5 {
                return 0, false
            }
            value := count
            count++
            return value, true
        },

        func(value int) int {
            return value * value
        },

        func(value int) {
            fmt.Println(value)
        },
    )
    pipeline.Run()
}

Output:

1
4
9
16
25

Development

Format:

go fmt ./...

Test:

go test ./...

Validate:

go vet ./...

License

Apache License 2.0

Documentation

Overview

Package flowkit provides generic pipeline primitives for composing data processing workflows.

FlowKit supports producers, consumers, iterators, streams, and functional operators.

Index

• func Chunk[T any](values []T, size int) [][]T
• func Collect[T any](iterator Iterator[T]) []T
• func Compose[T any](functions ...func(T) T) func(T) T
• func Filter[T any](values []T, predicate func(T) bool) []T
• func Fold[T any, R any](values []T, initial R, fn func(R, T) R) R
• func Map[T any, R any](values []T, fn func(T) R) []R
• func Pipe[T any](value T, functions ...func(T) T) T
• func Reduce[T any, R any](values []T, initial R, fn func(R, T) R) R
• func Tap[T any](value T, fn func(T)) T
• type Consumer
• type Iterator
  • func NewIterator[T any](producer Producer[T]) Iterator[T]
  • func (i Iterator[T]) ForEach(consumer Consumer[T])
  • func (i Iterator[T]) Next() (T, bool)
• type Pipeline
  • func New[T any, R any](producer Producer[T], transformer Transformer[T, R], consumer Consumer[R]) *Pipeline[T, R]
  • func (p *Pipeline[T, R]) Run()
• type Producer
• type Stream
  • func From[T any](values []T) Stream[T]
  • func (s Stream[T]) Filter(fn func(T) bool) Stream[T]
  • func (s Stream[T]) Map(fn func(T) T) Stream[T]
  • func (s Stream[T]) Reduce(initial T, fn func(T, T) T) T
  • func (s Stream[T]) Values() []T
• type Transformer

Functions

func Chunk

func Chunk[T any](
	values []T,
	size int,
) [][]T

Chunk splits a slice into smaller groups.

func Collect

func Collect[T any](
	iterator Iterator[T],
) []T

Collect consumes an iterator into a slice.

func Compose

func Compose[T any](
	functions ...func(T) T,
) func(T) T

Compose creates a transformation chain.

Functions execute from left to right.

func Filter

func Filter[T any](
	values []T,
	predicate func(T) bool,
) []T

Filter keeps only values matching the predicate.

func Fold

func Fold[T any, R any](
	values []T,
	initial R,
	fn func(R, T) R,
) R

Fold is an alias for Reduce.

func Map

func Map[T any, R any](
	values []T,
	fn func(T) R,
) []R

Map transforms each value into another type.

func Pipe

func Pipe[T any](
	value T,
	functions ...func(T) T,
) T

Pipe applies a value through multiple functions.

func Reduce

func Reduce[T any, R any](
	values []T,
	initial R,
	fn func(R, T) R,
) R

Reduce combines all values into a single result.

func Tap

func Tap[T any](
	value T,
	fn func(T),
) T

Tap executes a side effect and returns the original value.

Types

type Consumer

type Consumer[T any] func(T)

Consumer receives processed values.

type Iterator

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

Iterator represents a lazy data source.

func NewIterator

func NewIterator[T any](
	producer Producer[T],
) Iterator[T]

NewIterator creates a new iterator.

func (Iterator[T]) ForEach

func (i Iterator[T]) ForEach(
	consumer Consumer[T],
)

ForEach consumes all iterator values.

func (Iterator[T]) Next

func (i Iterator[T]) Next() (T, bool)

Next returns the next value.

type Pipeline

type Pipeline[T any, R any] struct {
	// contains filtered or unexported fields
}

Pipeline represents:

Producer -> Transformer -> Consumer

It reads values from a producer, transforms them, and sends them to a consumer.

func New

func New[T any, R any](
	producer Producer[T],
	transformer Transformer[T, R],
	consumer Consumer[R],
) *Pipeline[T, R]

New creates a new pipeline.

func (*Pipeline[T, R]) Run

func (p *Pipeline[T, R]) Run()

Run executes the pipeline.

Execution stops when the producer returns false.

type Producer

type Producer[T any] func() (T, bool)

Producer generates values. The bool indicates whether another value exists.

type Stream

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

Stream provides chainable collection operations.

func From

func From[T any](
	values []T,
) Stream[T]

From creates a stream from a slice.

func (Stream[T]) Filter

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

Filter removes values that do not match.

func (Stream[T]) Map

func (s Stream[T]) Map(
	fn func(T) T,
) Stream[T]

Map transforms stream values.

func (Stream[T]) Reduce

func (s Stream[T]) Reduce(
	initial T,
	fn func(T, T) T,
) T

Reduce aggregates stream values.

func (Stream[T]) Values

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

Values returns stream contents.

type Transformer

type Transformer[T any, R any] func(T) R

Transformer converts one value type into another.