flow

README

flow

✨ xuender/flow is a streaming programming library based on iter for Go 1.23.

❗ Prerequirement

• Go 1.23+

🚀 Install

go get github.com/xuender/flow@latest

💡 Usage

Chain

package main

import (
  "fmt"

  "github.com/xuender/flow"
  "github.com/xuender/flow/seq"
)

func main() {
  start := time.Now()

  for num :=range flow.Chain(
    seq.Range(100),
    flow.Filter(func(num int) bool { return num%3 == 0 }),
    flow.Skip[int](5),
    flow.Limit[int](4),
    flow.Peek(func(num int) {
      fmt.Println("peek", num)
      time.Sleep(time.Second)
    }),
    flow.Reverse[int](),
  ) {
    fmt.Println(num)
  }

  fmt.Printf("Chain: %v\n", time.Since(start))
}

Output:

peek 15
peek 18
peek 21
peek 24
24
21
18
15
Chain: 4s

[play]

Parallel

package main

import (
  "fmt"

  "github.com/xuender/flow"
  "github.com/xuender/flow/seq"
)

func main() {
  start := time.Now()

  for num := range flow.Parallel(3,
    seq.Range(100),
    flow.Filter(func(num int) bool { return num%3 == 0 }),
    flow.Skip[int](5),
    flow.Limit[int](4),
    flow.Peek(func(num int) {
      fmt.Println("peek", num)
      time.Sleep(time.Second)
    }),
    flow.Reverse[int](),
  ) {
    fmt.Println(num)
  }

  fmt.Printf("Parallel: %v\n", time.Since(start))
}

Output:

peek 15
peek 18
peek 21
peek 24
24
21
18
15
Parallel: 2s

[play]

seq.Range

package main

import (
  "fmt"

  "github.com/xuender/flow/seq"
)

func main() {
  fmt.Println("seq.Range(3)")

  for num := range seq.Range(3) {
    fmt.Println(num)
  }

  fmt.Println("seq.Range(-2, -6)")

  for num := range seq.Range(-2, -6) {
    fmt.Println(num)
  }

  fmt.Println("seq.Range(2, 10, 3)")

  for num := range seq.Range(2, 10, 3) {
    fmt.Println(num)
  }

  fmt.Println("seq.Range(3, 7, 0)")

  for num := range seq.Range(3, 7, 0) {
    fmt.Println(num)
  }
}

Output:

seq.Range(3)
0
1
2
seq.Range(-2, -6)
-2
-3
-4
-5
seq.Range(2, 10, 3)
2
5
8
seq.Range(3, 7, 0)
3
3
3
3

[play]

Flow Functions

iter.Seq[V]	iter.Seq2[K,V]
Chain	Chain2
Parallel	Parallel2

Intermediate Functions

iter.Seq[V]	iter.Seq2[K,V]	Note
Append	Append2	Used seq.Tuple[K, V].
Distinct	Distinct2
Filter	Filter2	Parallel function.
Limit	Limit2
Merge	Merge2
Peek	Peek2	Parallel function.
Repeat	Repeat2
Reverse	Reverse2
Skip	Skip2
Sort	Sort2
SortFunc	SortFunc2
SortStableFunc	SortStableFunc2

Terminal Functions

iter.Seq[V]	iter.Seq2[K,V]
seq.AllMatch	seq.AllMatch2
seq.AnyMatch	seq.AnyMatch2
seq.Clone	seq.Clone2
seq.Chunk	seq.Chunk2
seq.Count	seq.Count2
seq.Each	seq.Each2
seq.First	seq.First2
seq.Last	seq.Last2
seq.Join
seq.Max	seq.Max2
seq.Min	seq.Min2
seq.NoneMatch	seq.NoneMatch2
seq.Reduce	seq.Reduce2
seq.Sum

Seq Functions

iter.Seq[V]	iter.Seq2[K,V]	Note
seq.Append	seq.Append2
seq.Chan	seq.Chan2	Used seq.Tuple[K, V].
seq.Collect	seq.Collect2
seq.Distinct	seq.Distinct2
seq.Emit	seq.Emit2
seq.Filter	seq.Filter2
seq.FlatMap	seq.FlatMap2
	seq.Keys
seq.Limit	seq.Limit2
seq.Map	seq.Map2
seq.Merge	seq.Merge2
seq.Peek	seq.Peek2
seq.Range	seq.Range2
seq.Reduce	seq.Reduce2
seq.Repeat	seq.Repeat2
seq.Reverse	seq.Reverse2
seq.Send	seq.Send2
seq.Skip	seq.Skip2
seq.Sorted	seq.Sorted2
seq.SortedFunc	seq.SortedFunc2	Used seq.Tuple[K, V].
seq.SortedStableFunc	seq.SortedStableFunc2	Used seq.Tuple[K, V].
seq.ToChans	seq.ToChans2	Used seq.Tuple[K, V].
	seq.Values
	seq.Tuples

👤 Contributors

📝 License

© ender, 2024~time.Now

MIT LICENSE

Documentation

Overview

Package flow implements functionality similar to Java 8 Streams.

It utilizes Go 1.23's iterator and generic type support.

Index

• func Chain[V any](input iter.Seq[V], steps ...Step[V]) iter.Seq[V]
• func Chain2[K, V any](input iter.Seq2[K, V], steps ...Step2[K, V]) iter.Seq2[K, V]
• func Parallel[V any](num int, input iter.Seq[V], steps ...Step[V]) iter.Seq[V]
• func Parallel2[K, V any](num int, input iter.Seq2[K, V], steps ...Step2[K, V]) iter.Seq2[K, V]
• type Step
  • func Append[V any](items ...V) Step[V]
  • func Distinct[V comparable]() Step[V]
  • func Filter[V any](predicate func(V) bool) Step[V]
  • func Limit[V any](limit int) Step[V]
  • func Merge[V any](seqs ...iter.Seq[V]) Step[V]
  • func Peek[V any](action func(V)) Step[V]
  • func Repeat[V any](count int) Step[V]
  • func Reverse[V any]() Step[V]
  • func Skip[V any](num int) Step[V]
  • func Sort[V cmp.Ordered]() Step[V]
  • func SortFunc[V any](cmp func(V, V) int) Step[V]
  • func SortStableFunc[V any](cmp func(V, V) int) Step[V]
• type Step2
  • func Append2[K, V any](items ...seq.Tuple[K, V]) Step2[K, V]
  • func Distinct2[K comparable, V any]() Step2[K, V]
  • func Filter2[K, V any](predicate func(K, V) bool) Step2[K, V]
  • func Limit2[K, V any](limit int) Step2[K, V]
  • func Merge2[K, V any](seqs ...iter.Seq2[K, V]) Step2[K, V]
  • func Peek2[K, V any](action func(K, V)) Step2[K, V]
  • func Repeat2[K, V any](count int) Step2[K, V]
  • func Reverse2[K, V any]() Step2[K, V]
  • func Skip2[K, V any](num int) Step2[K, V]
  • func Sort2[K cmp.Ordered, V any]() Step2[K, V]
  • func SortFunc2[K, V any](cmp func(seq.Tuple[K, V], seq.Tuple[K, V]) int) Step2[K, V]
  • func SortStableFunc2[K, V any](cmp func(seq.Tuple[K, V], seq.Tuple[K, V]) int) Step2[K, V]

Examples

• Append
• Append2
• Chain
• Chain (Map)
• Distinct
• Distinct2
• Filter
• Filter2
• Limit
• Limit2
• Merge
• Merge2
• Parallel
• Parallel2
• Peek
• Peek2
• Repeat
• Repeat2
• Reverse
• Reverse2
• Skip
• Skip2
• Sort
• Sort2
• SortFunc
• SortFunc2
• SortStableFunc
• SortStableFunc2

Functions

func Chain

func Chain[V any](input iter.Seq[V], steps ...Step[V]) iter.Seq[V]

Chain applies a series of transformation steps to the input sequence sequentially.

This function takes an input sequence `input` and a variadic list of `steps`. Each step is applied sequentially to the input sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Max(flow.Chain(
		seq.Range(100),
		flow.Limit[int](3),
		flow.Filter(func(num int) bool { return num%2 == 0 }),
	)))

}

Output:

2 true

Example (Map)

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	items := flow.Chain(
		seq.Range(100),
		flow.Limit[int](3),
		flow.Filter(func(num int) bool { return num%2 == 0 }),
	)

	fmt.Println(
		seq.Join(flow.Chain(
			seq.Map(items, func(num int) string { return fmt.Sprintf("num:%d", num) }),
			flow.Limit[string](2),
		), ","),
	)

}

Output:

num:0,num:2

func Chain2

func Chain2[K, V any](input iter.Seq2[K, V], steps ...Step2[K, V]) iter.Seq2[K, V]

Chain2 applies multiple Step2 functions sequentially to the input sequence.

It returns the final sequence after applying all steps.

func Parallel

func Parallel[V any](num int, input iter.Seq[V], steps ...Step[V]) iter.Seq[V]

Parallel processes the input sequence using multiple workers and applies a series of steps.

This function takes an input sequence `input` and a variadic list of `steps`. Each step is applied sequentially to the input sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	for num := range flow.Parallel(2,
		seq.Range(100),
		flow.Limit[int](8),
		flow.Filter(func(num int) bool { return num%3 == 0 }),
		flow.Sort[int](),
	) {
		if num > 3 {
			break
		}

		fmt.Println(num)
	}

}

Output:

0
3

func Parallel2

func Parallel2[K, V any](num int, input iter.Seq2[K, V], steps ...Step2[K, V]) iter.Seq2[K, V]

Parallel2 processes the input sequence in parallel using multiple workers.

It returns a new sequence with the processed results.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	for key, val := range flow.Parallel2(2,
		seq.Range2(100),
		flow.Limit2[int, int](8),
		flow.Filter2(func(key, _ int) bool { return key%3 == 0 }),
		flow.Sort2[int, int](),
	) {
		if key > 3 {
			break
		}

		fmt.Println(key, val)
	}

}

Output:

0 0
3 3

Types

type Step

type Step[V any] struct {
	Next     func(iter.Seq[V]) iter.Seq[V]
	Parallel bool
}

Step represents a processing step in an iterator.

It includes a Next function to perform the next operation on a sequence, and a Parallel boolean indicating whether it can be executed in parallel.

func Append

func Append[V any](items ...V) Step[V]

Append adds multiple elements to the end of a sequence.

It takes a variable number of elements and appends them to the input sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	for num := range flow.Chain(
		seq.Range(2),
		flow.Append(9),
	) {
		fmt.Println(num)
	}

}

Output:

0
1
9

func Distinct

func Distinct[V comparable]() Step[V]

Distinct returns a transformation step that filters out duplicate elements from a sequence.

This function returns a `Step` that can be used to create a new sequence containing only the distinct elements from the input sequence.

Example

package main

import (
	"fmt"
	"slices"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Sum(flow.Chain(
		slices.Values([]int{1, 2, 2, 3, 3}),
		flow.Distinct[int](),
	)))

}

Output:

6

func Filter

func Filter[V any](predicate func(V) bool) Step[V]

Filter returns a transformation step that filters elements based on a predicate.

This function returns a `Step` that can be used to create a new sequence containing only the elements that satisfy the given `predicate`.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Sum(flow.Chain(
		seq.Range(10),
		flow.Filter(func(i int) bool {
			return i%3 == 0
		}),
	)))

}

Output:

18

func Limit

func Limit[V any](limit int) Step[V]

Limit returns a transformation step that limits the number of elements in a sequence.

This function returns a `Step` that creates a new sequence containing at most `limit` elements from the input sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Sum(flow.Chain(
		seq.Range(10),
		flow.Limit[int](4),
	)))

}

Output:

6

func Merge

func Merge[V any](seqs ...iter.Seq[V]) Step[V]

Merge combines multiple sequences into a single sequence.

It merges the input sequence with additional sequences.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Count(flow.Chain(
		seq.Range(3),
		flow.Merge(seq.Range(4), seq.Range(2)),
	)))

	fmt.Println(seq.Count(flow.Chain(
		seq.Range(3),
		flow.Merge[int](),
	)))

}

Output:

9
3

func Peek

func Peek[V any](action func(V)) Step[V]

Peek returns a transformation step that applies an action to each element in the sequence.

This function returns a `Step` that calls the `action` function on each element of the input sequence without modifying the sequence itself.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	seq.Emit(flow.Chain(
		seq.Range(3),
		flow.Peek(func(num int) {
			fmt.Println(num)
		}),
	))

}

Output:

0
1
2

func Repeat

func Repeat[V any](count int) Step[V]

Repeat creates a new step that repeats the input sequence a specified number of times.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Count(flow.Chain(
		seq.Range(3),
		flow.Repeat[int](4),
	)))

}

Output:

12

func Reverse

func Reverse[V any]() Step[V]

Reverse returns a transformation step that reverses the order of elements in a sequence.

It returns a `Step` that creates a new sequence with the elements in reverse order.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.First(flow.Chain(
		seq.Range(100),
		flow.Reverse[int](),
	)))

}

Output:

99 true

func Skip

func Skip[V any](num int) Step[V]

Skip returns a transformation step that skips the first `num` elements in a sequence.

It returns a `Step` that creates a new sequence by skipping the first `num` elements of the input sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.First(flow.Chain(
		seq.Range(100),
		flow.Skip[int](99),
	)))

}

Output:

99 true

func Sort

func Sort[V cmp.Ordered]() Step[V]

Sort returns a transformation step that sorts the elements in a sequence.

It returns a `Step` function that sorts the sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.First(flow.Chain(
		seq.Range(100),
		flow.Reverse[int](),
		flow.Sort[int](),
	)))

}

Output:

0 true

func SortFunc

func SortFunc[V any](cmp func(V, V) int) Step[V]

SortFunc returns a transformation step that sorts the elements in a sequence using a custom comparison function.

It returns a `Step` that creates a new sequence with the elements sorted according to the `cmp` function.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.First(flow.Chain(
		seq.Range(100),
		flow.SortFunc(func(num1, num2 int) int {
			return num2 - num1
		}),
	)))

}

Output:

99 true

func SortStableFunc

func SortStableFunc[V any](cmp func(V, V) int) Step[V]

SortStableFunc sorts the elements in the sequence using a custom comparison function with stable sorting.

It returns a `Step` function that sorts the sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.First(flow.Chain(
		seq.Range(100),
		flow.SortStableFunc(func(num1, num2 int) int {
			return num2 - num1
		}),
	)))

}

Output:

99 true

type Step2

type Step2[K, V any] struct {
	Next     func(iter.Seq2[K, V]) iter.Seq2[K, V]
	Parallel bool
}

Step2 represents a processing step in an iterator for sequences with two types K and V.

It includes a Next function to perform the next operation on a sequence, and a Parallel boolean indicating whether it can be executed in parallel.

func Append2

func Append2[K, V any](items ...seq.Tuple[K, V]) Step2[K, V]

Append2 appends additional (key, value) tuples to the input sequence.

It returns a new sequence with the appended tuples.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	for key, val := range flow.Chain2(
		seq.Range2(2),
		flow.Append2(seq.T(7, 8)),
	) {
		fmt.Println(key, val)
	}

}

Output:

0 0
1 1
7 8

func Distinct2

func Distinct2[K comparable, V any]() Step2[K, V]

Distinct2 removes duplicate (key, value) pairs from the sequence.

It returns a `Step2` function that filters duplicates.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Count2(flow.Chain2(
		seq.Range2(3),
		flow.Append2(seq.T(2, 7), seq.T(8, 4)),
		flow.Distinct2[int, int](),
	)))

}

Output:

4

func Filter2

func Filter2[K, V any](predicate func(K, V) bool) Step2[K, V]

Filter2 filters (key, value) pairs based on a predicate function.

It returns a new sequence with filtered pairs.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Count2(flow.Chain2(
		seq.Range2(10),
		flow.Filter2(func(key, _ int) bool {
			return key%3 == 0
		}),
	)))

}

Output:

4

func Limit2

func Limit2[K, V any](limit int) Step2[K, V]

Limit2 limits the number of (key, value) pairs in the sequence.

It returns a new sequence with a maximum number of pairs.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Count2(flow.Chain2(
		seq.Range2(10),
		flow.Limit2[int, int](4),
	)))

}

Output:

4

func Merge2

func Merge2[K, V any](seqs ...iter.Seq2[K, V]) Step2[K, V]

Merge2 merges multiple sequences of (key, value) pairs.

It returns a new sequence containing all pairs.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Count2(flow.Chain2(
		seq.Range2(3),
		flow.Merge2(seq.Range2(4), seq.Range2(2)),
	)))

	fmt.Println(seq.Count2(flow.Chain2(
		seq.Range2(3),
		flow.Merge2[int, int](),
	)))

}

Output:

9
3

func Peek2

func Peek2[K, V any](action func(K, V)) Step2[K, V]

Peek2 applies an action to each (key, value) pair in the sequence.

It returns a new sequence with the same pairs.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	seq.Emit2(flow.Chain2(
		seq.Range2(3),
		flow.Peek2(func(key, val int) {
			fmt.Println(key, val)
		}),
	))

}

Output:

0 0
1 1
2 2

func Repeat2

func Repeat2[K, V any](count int) Step2[K, V]

Repeat2 creates a new step that repeats each element of the input sequence a specified number of times.

Tt returns a `Step2` that repeats each element of the input sequence `count` times.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.Count2(flow.Chain2(
		seq.Range2(3),
		flow.Repeat2[int, int](4),
	)))

}

Output:

12

func Reverse2

func Reverse2[K, V any]() Step2[K, V]

Reverse2 reverses the order of (key, value) pairs in the sequence.

It returns a `Step2` function that reverses the sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.First2(flow.Chain2(
		seq.Range2(100),
		flow.Reverse2[int, int](),
	)))

}

Output:

99 99 true

func Skip2

func Skip2[K, V any](num int) Step2[K, V]

Skip2 skips the first 'num' (key, value) pairs in the sequence.

It returns a `Skip2` with the remaining pairs.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.First2(flow.Chain2(
		seq.Range2(100),
		flow.Skip2[int, int](99),
	)))

}

Output:

99 99 true

func Sort2

func Sort2[K cmp.Ordered, V any]() Step2[K, V]

Sort2 sorts the (key, value) pairs in the sequence.

It returns a `Step2` function that sorts the sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.First2(flow.Chain2(
		seq.Range2(100),
		flow.Reverse2[int, int](),
		flow.Sort2[int, int](),
	)))

}

Output:

0 0 true

func SortFunc2

func SortFunc2[K, V any](cmp func(seq.Tuple[K, V], seq.Tuple[K, V]) int) Step2[K, V]

SortFunc2 sorts (key, value) pairs using a custom comparison function.

It returns a `Step2` function that sorts the sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.First2(flow.Chain2(
		seq.Range2(100),
		flow.SortFunc2(func(num1, num2 seq.Tuple[int, int]) int {
			return num2.K - num1.K
		}),
	)))

}

Output:

99 99 true

func SortStableFunc2

func SortStableFunc2[K, V any](cmp func(seq.Tuple[K, V], seq.Tuple[K, V]) int) Step2[K, V]

SortStableFunc2 sorts (key, value) pairs using a custom comparison function with stable sorting.

It returns a `Step2` function that sorts the sequence.

Example

package main

import (
	"fmt"

	"github.com/xuender/flow"
	"github.com/xuender/flow/seq"
)

func main() {
	fmt.Println(seq.First2(flow.Chain2(
		seq.Range2(100),
		flow.SortStableFunc2(func(num1, num2 seq.Tuple[int, int]) int {
			return num2.K - num1.K
		}),
	)))

}

Output:

99 99 true