stream

package module
v0.0.0-...-7a02d84 Latest Latest
Warning

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

 Go to latest Published: Aug 14, 2023 License: MIT Imports: 4 Imported by: 0

README

go-stream

GoCI Go Report Card codecov GoDoc GitHub license GitHub issues GitHub stars

Stream Collections for Go. Inspired in Java 8 Streams.

Use channel and Go1.18+ generic support.

✨ xuender/go-stream is a Java 8 Streams style Go library based on Go 1.18+ Generics.

🚀 Install

To install the library and command line program:

go get -u github.com/xuender/go-stream@latest

💡 Usage

You can import stream using:

import "github.com/xuender/go-stream"
NewBase

New BaseStream.

package main

import (
  "fmt"

  "github.com/xuender/go-stream"
)

func main() {
  base := stream.NewBase(stream.Range2Channel(5)).
    Peek(func(num int) { fmt.Println("peek1:", num) }).
    Filter(func(num int) bool { return num > 2 }).
    Peek(func(num int) { fmt.Println("peek2:", num) })

  fmt.Println(base.Count())
}

Output:

peek1: 0
peek1: 1
peek1: 2
peek1: 3
peek1: 4
peek2: 3
peek2: 4
2

[play]

Parallel

BaseStream to ParallelStream.

package main

import (
  "fmt"
  "math/rand"
  "time"

  "github.com/xuender/go-stream"
)

func main() {
  stream.NewBase(stream.Range2Channel(1000)).
    Parallel(100).
    Filter(func(num int) bool { return num%7 == 0 }).
    ForEach(func(num int) {
      dur := time.Duration(rand.Intn(1000)) * time.Millisecond

      time.Sleep(dur)
      fmt.Printf("%d\t%s\n", num, dur)
    })
}

Output:

623     2ms
497     2ms
273     15ms
252     26ms
616     33ms
756     10ms
91      47ms
7       59ms
21      59ms
602     59ms
350     78ms
28      81ms
...

[play]

Map

Integer to string.

package main

import (
  "fmt"

  "github.com/xuender/go-stream"
)

func main() {
  base := stream.Map(
    stream.Range2Channel(100),
    func(num int) string { return fmt.Sprintf("[%d]", num) },
  ).Limit(3)

  for i := range base.C {
    fmt.Println(i)
  }
}

Output:

[0]
[1]
[2]

[play]

FlatMap
package main

import (
  "fmt"

  "github.com/xuender/go-stream"
)

func main() {
  stream.FlatMap(
    stream.Slice2Channel([]int{0, 0}, []int{1, 2}, []int{2, 4}),
    func(num int) string { return fmt.Sprintf("[%d]", num) },
  ).ForEach(func(str string) {
    fmt.Println(str)
  })
}

Output:

[0]
[0]
[1]
[2]
[2]
[4]

[play]

Sorted

OrderedStream sorted.

package main

import (
  "fmt"

  "github.com/xuender/go-stream"
)

func main() {
  stream.NewOrdered(stream.Slice2Channel(3, 2, 7, 1)).
    Sorted().
    ForEach(func(num int) {
      fmt.Println(num)
    })
}

Output:

1
2
3
7

[play]

🛩 Functions

Function Type State
AnyMatch Terminal operations, short-circuiting √
AllMatch Terminal operations, short-circuiting √
Count Terminal operations √
Filter Intermediate operations, Parallel √
FindFirst Terminal operations, short-circuiting √
ForEach Terminal operations, Parallel √
Limit Intermediate operations √
NoneMatch Terminal operations, short-circuiting √
Parallel Intermediate operations √
Peek Intermediate operations, Parallel √
Skip Intermediate operations √
Sort Intermediate operations √
Distinct Intermediate operations, Comparable √
Max Terminal operations, Ordered √
Min Terminal operations, Ordered √
Reduce Terminal operations, Ordered √
Sorted Intermediate operations, Ordered √
Sequential Intermediate operations, Parallel √
Map Function √
FlatMap Function √

📝 License

© ender, 2023~time.Now

MIT LICENSE

Documentation

Overview

Package stream implements like Java 8 Streams for Go. Use channel and Go1.18 generic support.

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

func Copy 

func Copy[T any](source *BaseStream[T]) chan T

func Count 

func Count[T any](input <-chan T) int

func Distinct 

func Distinct[C comparable](input <-chan C) chan C

Distinct returns a channel consisting of the distinct elements of input channel.

func Distribute 

func Distribute[T any](input <-chan T, output1, output2 chan<- T)

func Filter 

func Filter[T any](input <-chan T, action FilterAction[T]) chan T

Filter returns a channel consisting of the elements of input channel that match the given action.

Example 
chi := stream.Filter(
	stream.Range2Channel(10),
	func(num int) bool { return num > 5 },
)

for i := range chi {
	fmt.Println(i)
}

Output:

6
7
8
9

func FilterParallel 

func FilterParallel[T any](input <-chan T, size int, action FilterAction[T]) chan T

FilterParallel returns a channel consisting of the elements of input channel that match the given action, parallel.

func Limit 

func Limit[T any](input <-chan T, maxSize int) chan T

Limit returns a channel consisting of the elements of input channel, truncated to be no longer than maxSize in length.

func Peek 

func Peek[T any](input <-chan T, action Action[T]) chan T

Peek returns a channel consisting of the elements of input channel, additionally performing the provided action on each element as elements are consumed from the resulting stream.

func PeekParallel 

func PeekParallel[T any](input <-chan T, size int, action Action[T]) chan T

func Range2Channel 

func Range2Channel(length int) chan int
Example 
for i := range stream.Range2Channel(3) {
	fmt.Println(i)
}

for i := range stream.Range2Channel(-3) {
	fmt.Println(i)
}

Output:

0
1
2
0
-1
-2

func RangeFrom2Channel 

func RangeFrom2Channel(start, length int) chan int
Example 
for i := range stream.RangeFrom2Channel(3, 2) {
	fmt.Println(i)
}

for i := range stream.RangeFrom2Channel(8, -3) {
	fmt.Println(i)
}

Output:

3
4
8
7
6

func RangeWithSteps2Channel 

func RangeWithSteps2Channel(start, end, step int) chan int
Example 
for i := range stream.RangeWithSteps2Channel(0, 4, 3) {
	fmt.Println(i)
}

for i := range stream.RangeWithSteps2Channel(9, 5, -3) {
	fmt.Println(i)
}

Output:

0
3
9
6

func Reduce 

func Reduce[O constraints.Ordered](input <-chan O) chan O

func Skip 

func Skip[T any](input <-chan T, size int) chan T

func Slice2Channel 

func Slice2Channel[T any](elems ...T) chan T
Example 
for i := range stream.Slice2Channel(1, 2, 3) {
	fmt.Println(i)
}

Output:

1
2
3

func Sort 

func Sort[T any](input <-chan T, action LessAction[T]) chan T

func Sorted 

func Sorted[O constraints.Ordered](input <-chan O) chan O

Types

type Action 

type Action[T any] func(T)

type BaseStream 

type BaseStream[T any] struct {
	C chan T
}

func FlatMap 

func FlatMap[I, O any](input chan []I, action MapAction[I, O]) *BaseStream[O]
Example 
stream.FlatMap(
	stream.Slice2Channel([]int{0, 0}, []int{1, 2}, []int{2, 4}),
	func(num int) string { return fmt.Sprintf("[%d]", num) },
).ForEach(func(str string) {
	fmt.Println(str)
})

Output:

[0]
[0]
[1]
[2]
[2]
[4]

func Map 

func Map[I, O any](input chan I, action MapAction[I, O]) *BaseStream[O]
Example 
base1 := stream.NewBase(stream.Range2Channel(10)).
	Filter(func(num int) bool { return num > 5 })
base2 := stream.Map(
	base1.C,
	func(num int) string { return fmt.Sprintf("[%d]", num) },
)

for num := range base2.C {
	fmt.Println(num)
}

Output:

[6]
[7]
[8]
[9]

func NewBase 

func NewBase[T any](input chan T) *BaseStream[T]
Example 
base := stream.NewBase(stream.Range2Channel(10)).
	Filter(func(num int) bool { return num > 5 })

fmt.Println(base.Count())

Output:

4

func (*BaseStream[T]) AllMatch 

func (p *BaseStream[T]) AllMatch(action FilterAction[T]) bool
Example 
base1 := stream.NewBase(stream.Range2Channel(10))
base2 := stream.NewBase(stream.Copy(base1))

fmt.Println(base1.AllMatch(func(num int) bool { return num > 8 }))
fmt.Println(base2.AllMatch(func(num int) bool { return num >= 0 }))

Output:

false
true

func (*BaseStream[T]) AnyMatch 

func (p *BaseStream[T]) AnyMatch(action FilterAction[T]) bool
Example 
base1 := stream.NewBase(stream.Range2Channel(10))
base2 := stream.NewBase(stream.Copy(base1))

fmt.Println(base1.AnyMatch(func(num int) bool { return num > 100 }))
fmt.Println(base2.AnyMatch(func(num int) bool { return num > 1 }))

Output:

false
true

func (*BaseStream[T]) Count 

func (p *BaseStream[T]) Count() int

func (*BaseStream[T]) Filter 

func (p *BaseStream[T]) Filter(action FilterAction[T]) *BaseStream[T]

Filter returns a stream consisting of the elements of this stream that match the given action.

Example 
stream.NewBase(stream.Range2Channel(10)).
	Filter(func(num int) bool { return num > 5 }).
	ForEach(func(num int) {
		fmt.Println(num)
	})

Output:

6
7
8
9

func (*BaseStream[T]) FindFirst 

func (p *BaseStream[T]) FindFirst() T
Example 
base := stream.NewBase(stream.Range2Channel(10)).
	Filter(func(num int) bool { return num > 5 })

fmt.Println(base.FindFirst())

Output:

6

func (*BaseStream[T]) ForEach 

func (p *BaseStream[T]) ForEach(action Action[T])
Example 
stream.NewBase(stream.Range2Channel(10)).
	ForEach(func(num int) { fmt.Println(num) })

Output:

0
1
2
3
4
5
6
7
8
9

func (*BaseStream[T]) Limit 

func (p *BaseStream[T]) Limit(maxSize int) *BaseStream[T]

Limit returns a stream consisting of the elements of this stream, truncated to be no longer than maxSize in length.

Example 
stream.NewBase(stream.Range2Channel(10)).
	Limit(3).
	ForEach(func(num int) {
		fmt.Println(num)
	})

Output:

0
1
2

func (*BaseStream[T]) NoneMatch 

func (p *BaseStream[T]) NoneMatch(action FilterAction[T]) bool
Example 
base1 := stream.NewBase(stream.Range2Channel(10))
base2 := stream.NewBase(stream.Copy(base1))

fmt.Println(base1.NoneMatch(func(num int) bool { return num > 100 }))
fmt.Println(base2.NoneMatch(func(num int) bool { return num > 1 }))

Output:

true
false

func (*BaseStream[T]) Parallel 

func (p *BaseStream[T]) Parallel(size int) *ParallelStream[T]

func (*BaseStream[T]) Peek 

func (p *BaseStream[T]) Peek(action Action[T]) *BaseStream[T]

Peek returns a stream consisting of the elements of this stream, additionally performing the provided action on each element as elements are consumed from the resulting stream.

Example 
count1 := 0
count2 := 0
base := stream.NewBase(stream.Range2Channel(3)).
	Peek(func(num int) { count1++ }).
	Filter(func(num int) bool { return num%2 == 0 }).
	Peek(func(num int) { count2++ })

fmt.Println(base.Count())
fmt.Println(count1)
fmt.Println(count2)

Output:

2
3
2

func (*BaseStream[T]) Skip 

func (p *BaseStream[T]) Skip(size int) *BaseStream[T]
Example 
stream.NewBase(stream.Range2Channel(6)).
	Skip(4).
	ForEach(func(num int) {
		fmt.Println(num)
	})

Output:

4
5

func (*BaseStream[T]) Sort 

func (p *BaseStream[T]) Sort(action LessAction[T]) *BaseStream[T]
Example 
stream.NewBase(stream.Slice2Channel(3, 2, 7, 1)).
	Sort(func(num1, num2 int) bool { return num2 < num1 }).
	ForEach(func(num int) {
		fmt.Println(num)
	})

Output:

7
3
2
1

type ComparableStream 

type ComparableStream[C comparable] struct {
	BaseStream[C]
}

func FlatMapComparable 

func FlatMapComparable[I any, O comparable](input chan []I, action MapAction[I, O]) *ComparableStream[O]
Example 
stream.FlatMapOrdered(
	stream.Slice2Channel([]int{1, 2}, []int{2, 4}, []int{3, 6}),
	func(num int) string { return fmt.Sprintf("[%d]", num) },
).
	Sorted().
	ForEach(func(str string) {
		fmt.Println(str)
	})

Output:

[1]
[2]
[2]
[3]
[4]
[6]

func MapComparable 

func MapComparable[I any, O comparable](input chan I, action MapAction[I, O]) *ComparableStream[O]
Example 
com := stream.MapComparable(
	stream.Slice2Channel(1, 1, 1, 2, 3, 3, 4),
	func(num int) string { return fmt.Sprintf("[%d]", num) },
).Distinct()

for i := range com.C {
	fmt.Println(i)
}

Output:

[1]
[2]
[3]
[4]

func NewComparable 

func NewComparable[C comparable](input chan C) *ComparableStream[C]

func (*ComparableStream[C]) Distinct 

func (p *ComparableStream[C]) Distinct() *ComparableStream[C]

Distinct returns a stream consisting of the distinct elements of this stream.

Example 
stream.NewComparable(stream.Slice2Channel(1, 1, 2, 3, 3, 4)).
	Distinct().
	ForEach(func(num int) {
		fmt.Println(num)
	})

Output:

1
2
3
4

type FilterAction 

type FilterAction[T any] func(T) bool

type LessAction 

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

type MapAction 

type MapAction[I, O any] func(I) O

type OrderedStream 

type OrderedStream[O constraints.Ordered] struct {
	ComparableStream[O]
}

func FlatMapOrdered 

func FlatMapOrdered[I any, O constraints.Ordered](input chan []I, action MapAction[I, O]) *OrderedStream[O]
Example 
stream.FlatMapComparable(stream.Slice2Channel([]int{1, 2}, []int{1, 2}, []int{2, 4},
	[]int{3, 6}, []int{3, 6}, []int{4, 8}),
	func(num int) string { return fmt.Sprintf("[%d]", num) },
).
	Distinct().
	ForEach(func(str string) {
		fmt.Println(str)
	})

Output:

[1]
[2]
[4]
[3]
[6]
[8]

func MapOrdered 

func MapOrdered[I any, O constraints.Ordered](input chan I, action MapAction[I, O]) *OrderedStream[O]
Example 
ordered := stream.MapOrdered(
	stream.Range2Channel(10),
	func(num int) string { return fmt.Sprintf("[%d]", num) },
)

fmt.Println(ordered.Max())

Output:

[9]

func NewOrdered 

func NewOrdered[O constraints.Ordered](input chan O) *OrderedStream[O]

func (*OrderedStream[O]) Max 

func (p *OrderedStream[O]) Max() O

func (*OrderedStream[O]) Min 

func (p *OrderedStream[O]) Min() O
Example 
ordered := stream.MapOrdered(stream.Range2Channel(10), func(num int) string { return fmt.Sprintf("[%d]", num) })

fmt.Println(ordered.Min())

Output:

[0]

func (*OrderedStream[O]) Reduce 

func (p *OrderedStream[O]) Reduce() *OrderedStream[O]
Example 
stream.NewOrdered(stream.Slice2Channel(3, 2, 7, 1)).
	Reduce().
	ForEach(func(num int) {
		fmt.Println(num)
	})

Output:

7
3
2
1

func (*OrderedStream[O]) Sorted 

func (p *OrderedStream[O]) Sorted() *OrderedStream[O]
Example 
stream.NewOrdered(stream.Slice2Channel(3, 2, 7, 1)).
	Sorted().ForEach(func(num int) {
	fmt.Println(num)
})

Output:

1
2
3
7

type ParallelStream 

type ParallelStream[T any] struct {
	BaseStream[T]
	Size int
}

func NewParallel 

func NewParallel[T any](input chan T, size int) *ParallelStream[T]

func (*ParallelStream[T]) Filter 

func (p *ParallelStream[T]) Filter(action FilterAction[T]) *ParallelStream[T]

Filter returns a stream consisting of the elements of this stream that match the given action, parallel.

Example 
stream.NewBase(stream.Range2Channel(10)).
	Parallel(3).
	Filter(func(num int) bool { return num > 5 }).
	ForEach(func(_ int) {
		fmt.Println("x")
	})

Output:

x
x
x
x

func (*ParallelStream[T]) ForEach 

func (p *ParallelStream[T]) ForEach(action Action[T])
Example 
stream.NewParallel(stream.Range2Channel(3), 3).
	ForEach(func(num int) {
		time.Sleep(time.Duration((3-num)*100) * time.Millisecond)
		fmt.Println(num)
	})

Output:

2
1
0

func (*ParallelStream[T]) Peek 

func (p *ParallelStream[T]) Peek(action Action[T]) *ParallelStream[T]
Example 
count := stream.NewParallel(stream.Range2Channel(3), 3).
	Peek(func(num int) {
		time.Sleep(time.Duration((3-num)*100) * time.Millisecond)
		fmt.Println(num)
	}).Count()

fmt.Println(count)

Output:

2
1
0
3

func (*ParallelStream[T]) Sequential 

func (p *ParallelStream[T]) Sequential() *BaseStream[T]
Example 
strea := stream.NewParallel(stream.Range2Channel(10), 10).
	Sequential().
	Filter(func(num int) bool { return num > 5 })

fmt.Println(strea.FindFirst())

Output:

6

Source Files

View all Source files

Directories

Path Synopsis
_examples
base command
count command
filter command
flat_map command
map command
match command
parallel command
peek command
sorted command

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.