channels

README

Channels

Simplifies working with channels in Go by abstracting Pipelines, filtering, map-reduce, etc. patterns for Channels.

Please check out the package documentation for more information.

Documentation

Index

• func BuildMapFromEntries[K comparable, V any](entries []maps.Entry[K, V]) map[K]V
• func CollectAsMap[I any, OK comparable, OV any](input <-chan I, fn MapBuilderFunc[I, OK, OV]) map[OK]OV
• func CollectAsSlice[T any](input <-chan T) []T
• func Filter[T any](input <-chan T, fn FilterFunc[T]) <-chan T
• func FromMap[K comparable, V any](input map[K]V) <-chan maps.Entry[K, V]
• func FromSlice[T any](input []T) <-chan T
• func Map[I, O any](input <-chan I, fn MapFunc[I, O]) chan O
• func Reduce[I, O any](input <-chan I, fn ReduceFunc[I, O]) <-chan O
• type FilterFunc
• type MapBuilderFunc
• type MapFunc
• type Pipeline
  • func NewPipeline[I, O any](input <-chan I, fn PipelineCreationFunc[I, O]) *Pipeline[I, O]
  • func (p Pipeline[I, O]) CollectAsSlice() []O
• type PipelineCreationFunc
• type ReduceFunc

Examples

• CollectAsMap
• CollectAsSlice
• Filter
• FromMap
• FromSlice
• Map
• Pipeline.CollectAsSlice
• Reduce

Functions

func BuildMapFromEntries

func BuildMapFromEntries[K comparable, V any](entries []maps.Entry[K, V]) map[K]V

BuildMapFromEntries takes a slice of maps.Entry and returns a map built from those entries.

func CollectAsMap

func CollectAsMap[I any, OK comparable, OV any](input <-chan I, fn MapBuilderFunc[I, OK, OV]) map[OK]OV

CollectAsMap reads all elements from the input channel and returns them as a map. This function will block until the input channel is closed.

Example

package main

import (
	"fmt"
	"github.com/pickeringtech/go-collections/channels"
	"github.com/pickeringtech/go-collections/maps"
)

func main() {
	input := channels.FromSlice([]string{"hello", "generous", "and", "glorious", "world"})
	output := channels.CollectAsMap(input, func(element string) maps.Entry[string, int] {
		return maps.Entry[string, int]{
			Key:   element,
			Value: len(element),
		}
	})
	fmt.Printf("result: %v", output)
}

Output:
result: map[and:3 generous:8 glorious:8 hello:5 world:5]

func CollectAsSlice

func CollectAsSlice[T any](input <-chan T) []T

CollectAsSlice reads all elements from the input channel and returns them as a slice. This function will block until the input channel is closed.

Example

package main

import (
	"fmt"
	"github.com/pickeringtech/go-collections/channels"
)

func main() {
	input := channels.FromSlice([]int{1, 2, 3, 4, 5})
	mapOut := channels.Map(input, func(element int) int {
		return element * 2
	})
	output := channels.CollectAsSlice(mapOut)
	fmt.Printf("result: %v", output)
}

Output:
result: [2 4 6 8 10]

func Filter

func Filter[T any](input <-chan T, fn FilterFunc[T]) <-chan T

Filter reads all elements from the input channel and writes them to the output channel if the given FilterFunc returns true for that element. This function will block until the input channel is closed.

Example

package main

import (
	"fmt"
	"github.com/pickeringtech/go-collections/channels"
)

func main() {
	input := channels.FromSlice([]string{"hello", "everyone", "world", "goodness", "gracious"})
	output := channels.Filter(input, func(element string) bool {
		return len(element) > 5
	})

	// Capture results in a slice.
	results := channels.CollectAsSlice(output)

	// Print results.
	fmt.Printf("Results: %v", results)
}

Output:
Results: [everyone goodness gracious]

func FromMap

func FromMap[K comparable, V any](input map[K]V) <-chan maps.Entry[K, V]

FromMap converts a map into a channel, writing the entries to the channel one-by-one. The channel will be closed after all entries have been read.

Example

package main

import (
	"fmt"
	"github.com/pickeringtech/go-collections/channels"
	"github.com/pickeringtech/go-collections/maps"
	"github.com/pickeringtech/go-collections/slices"
)

func main() {
	input := map[int]string{
		1:  "one",
		5:  "five",
		2:  "two",
		-1: "negative one",
	}
	output := channels.FromMap(input)

	// Capture results in a slice.
	results := channels.CollectAsSlice(output)

	// Sort results.
	results = slices.SortByOrderedField(results, slices.AscendingSortFunc[int], func(element maps.Entry[int, string]) int {
		return element.Key
	})

	// Print results.
	fmt.Printf("results: %v", results)
}

Output:
results: [{-1 negative one} {1 one} {2 two} {5 five}]

func FromSlice

func FromSlice[T any](input []T) <-chan T

FromSlice converts a slice into a channel, writing them to the channel one-by-one. The channel will be closed after all elements have been read.

Example

package main

import (
	"fmt"
	"github.com/pickeringtech/go-collections/channels"
)

func main() {
	input := []int{1, 2, 5, 4, 3}
	output := channels.FromSlice(input)

	// Capture results in a slice.
	results := channels.CollectAsSlice(output)

	// Print results.
	fmt.Printf("Results: %v", results)
}

Output:
Results: [1 2 5 4 3]

func Map

func Map[I, O any](input <-chan I, fn MapFunc[I, O]) chan O

Map takes an input channel, transforms each of its entries using the MapFunc until the input channel is closed. The results are output to an output channel returned from this function.

Example

package main

import (
	"fmt"
	"github.com/pickeringtech/go-collections/channels"
)

func main() {
	input := channels.FromSlice([]string{"one", "two", "three", "four", "five"})
	output := channels.Map(input, func(s string) int {
		return len(s)
	})

	// Capture results in a slice.
	results := channels.CollectAsSlice(output)

	// Print results.
	fmt.Printf("Results: %v", results)
}

Output:
Results: [3 3 5 4 4]

func Reduce

func Reduce[I, O any](input <-chan I, fn ReduceFunc[I, O]) <-chan O

Reduce reads all elements from the input channel and reduces them to a single value using the given ReduceFunc.

Example

package main

import (
	"fmt"
	"github.com/pickeringtech/go-collections/channels"
)

func main() {
	input := channels.FromSlice([]int{1, 2, 3, 4, 5})

	// Creates a new pipeline which totals the input channel.
	pipeline := channels.NewPipeline[int, int](input, func(input <-chan int) <-chan int {
		return channels.Reduce(input, func(accumulator int, element int) int {
			return accumulator + element
		})
	})

	// Capture results in a slice.
	results := pipeline.CollectAsSlice()

	fmt.Printf("Results: %v", results)
}

Output:
Results: [15]

Types

type FilterFunc

type FilterFunc[T any] func(element T) bool

FilterFunc is a function which takes an input element and returns true if the element should be included in the output channel, or false if it should be excluded.

type MapBuilderFunc

type MapBuilderFunc[I any, OK comparable, OV any] func(input I) maps.Entry[OK, OV]

MapBuilderFunc is a function which takes an input element and returns a maps.Entry, which is used to build a map.

type MapFunc

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

type Pipeline

type Pipeline[I, O any] struct {
	// contains filtered or unexported fields
}

Pipeline represents a channel backed pipeline, with a given start and end channel. This type is useful for ensuring that a given pipeline starts and ends with a given type, but the operations which occur in the middle of the pipeline (i.e. how an input is converted into the required output) are not specified.

func NewPipeline

func NewPipeline[I, O any](input <-chan I, fn PipelineCreationFunc[I, O]) *Pipeline[I, O]

NewPipeline creates a new Pipeline, with the given input channel and PipelineCreationFunc. The PipelineCreationFunc is used to create the end channel of the pipeline.

func (Pipeline[I, O]) CollectAsSlice

func (p Pipeline[I, O]) CollectAsSlice() []O

CollectAsSlice collects all elements from the end channel of the pipeline into a slice, which is returned. This function will block until the end channel is closed.

Example

package main

import (
	"fmt"
	"github.com/pickeringtech/go-collections/channels"
	"strconv"
)

func main() {
	input := channels.FromSlice([]int{1, 2, 5, 4, 3})

	// Creates a new pipeline which totals and then stringifies the input channel.
	pipeline := channels.NewPipeline[int, string](input, func(input <-chan int) <-chan string {
		reducer := channels.Reduce(input, func(accumulator int, element int) int {
			return accumulator + element
		})

		stringifier := channels.Map[int, string](reducer, func(element int) string {
			return strconv.Itoa(element)
		})

		return stringifier
	})

	// Capture results in a slice.
	results := pipeline.CollectAsSlice()

	fmt.Printf("Results: %v", results)
}

Output:
Results: [15]

type PipelineCreationFunc

type PipelineCreationFunc[I, O any] func(input <-chan I) <-chan O

PipelineCreationFunc is a function which takes a channel of the input type and returns a channel of the output type.

type ReduceFunc

type ReduceFunc[I, O any] func(accumulator O, element I) O

ReduceFunc is a function which takes an accumulator and an input element, and returns the new accumulator value.