gostreams

Overview ¶

Package gostreams provides a set of operations on streams of elements. Streams form a pipeline of operations that elements are passing through.

Streams consist of an initial ProducerFunc, which can produce elements from slices, channels, or any arbitrary source.

Intermediate ProducerFuncs may apply mapping, filtering, and sorting operations to the elements. Some of these operations can work on elements concurrently to increase throughput.

Finally, ConsumerFuncs consume the elements, and perform a final operation on them, such as collecting them into slices or maps, grouping/partitioning them, checking for matching elements, or simply iterating over them.

Stream operations receive a context.CancelCauseFunc. Calling the cancel function cancels the entire stream, short-circuiting the processing of elements. Depending on the intermediate operations and the final consumer, the result of the consumer may be undefined. Producer implementations must be prepared to handle cancellation at any time by checking the provided context.Context.

Streams are always lazy, meaning that producers will produce a new element only after a downstream producer or consumer has consumed the previous element.

// construct a producer from a slice
ints := Produce([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10})

// filter for even elements
// since we only need the elements themselves, we can use FuncPredicate
ints = Filter(ints, FuncPredicate(func(elem int) bool {
	return elem%2 == 0
}))

// map elements by doubling them
// since we only need the elements themselves, we can use FuncMapper
ints = Map(ints, FuncMapper(func(elem int) int {
	return elem * 2
}))

// map elements by converting them to strings
intStrs := Map(ints, FuncMapper(strconv.Itoa))

// perform a reduction to collect the strings into a slice
strs, _ := ReduceSlice(context.Background(), intStrs)

fmt.Printf("%+v\n", strs)

Output:

[4 8 12 16 20]