<commit_before>/*
Package pipe provides filters that can be chained together in a manner
similar to Unix pipelines.

Each filter is a function that takes as input a sequence of
strings (read from a channel) and produces as output a sequence of
strings (written to a channel).

Filters can be chained together (e.g., via the Run function), the
output of one filter is fed as input to the next filter.  The empty
input is passed to the first filter. The following sequence will
print two lines to standard output:

	err := pipe.Run(
		pipe.Echo("hello", "world"),
		pipe.Reverse(),
		pipe.WriteLines(os.Stdout),
	)

An application can implement its own filters easily. For example,
repeat(n) returns a filter that repeats every input n times.

	func repeat(n int) Filter {
		return func(arg pipe.Arg) error {
			for s := range arg.In {
				for i := 0; i < n; i++ {
					arg.Out <- s
				}
			}
			return nil
		}
	}

	pipe.Run(
		pipe.Echo("hello"),
		repeat(10),
	)

Note that repeat is not a Filter since it needs to accept the
parameter n. Instead, it returns a Filter.  This convention is
followed throughout this library: all filtering functionality is
provided by functions that return a Filter.

*/
package pipe

import (
	"fmt"
	"sync"
)

// filterErrors records errors accumulated during the execution of a filter.
type filterErrors struct {
	mu     sync.Mutex
	errors []error
}

func (e *filterErrors) getError() error {
	e.mu.Lock()
	defer e.mu.Unlock()
	switch len(e.errors) {
	case 0:
		return nil
	case 1:
		return e.errors[0]
	default:
		return fmt.Errorf("pipe.Filter errors: %s", e.errors)
	}
}

// Arg contains the data passed to a Filter. Arg.In is a channel that
// produces the input to the filter, and Arg.Out is a channel that
// receives the output from the filter.
type Arg struct {
	In  <-chan string
	Out chan<- string
}

// Filter is the type of a function that reads a sequence of strings
// from a channel and produces a sequence on another channel. A Filter
// returns nil on success, an error otherwise.
type Filter func(Arg) error

// Sequence returns a filter that is the concatenation of all filter arguments.
// The output of a filter is fed as input to the next filter.
func Sequence(filters ...Filter) Filter {
	return func(arg Arg) error {
		e := &filterErrors{}
		in := arg.In
		for _, f := range filters {
			c := make(chan string, 10000)
			go runFilter(f, Arg{in, c}, e)
			in = c
		}
		for s := range in {
			arg.Out <- s
		}
		return e.getError()
	}
}

// Run executes the sequence of filters and discards all output.
// It returns either nil, an error if any filter reported an error.
func Run(filters ...Filter) error {
	return ForEach(Sequence(filters...), func(s string) {})
}

// ForEach calls fn(s) for every item s in the output of filter and
// returns either nil, or any error reported by the execution of the filter.
func ForEach(filter Filter, fn func(s string)) error {
	in := make(chan string, 0)
	close(in)
	out := make(chan string, 10000)
	e := &filterErrors{}
	go runFilter(filter, Arg{in, out}, e)
	for s := range out {
		fn(s)
	}
	return e.getError()
}

// Output returns a slice that contains all items that are
// the output of filters.
func Output(filters ...Filter) ([]string, error) {
	var result []string
	err := ForEach(Sequence(filters...), func(s string) {
		result = append(result, s)
	})
	if err != nil {
		result = nil // Discard results on error
	}
	return result, err
}

func discard(in <-chan string) {
	for _ = range in {
	}
}

func runFilter(f Filter, arg Arg, e *filterErrors) {
	err := f(arg)
	close(arg.Out)
	discard(arg.In)
	if err != nil {
		e.mu.Lock()
		e.errors = append(e.errors, err)<commit_msg>removed single-use function<commit_after>/*
Package pipe provides filters that can be chained together in a manner
similar to Unix pipelines.

Each filter is a function that takes as input a sequence of
strings (read from a channel) and produces as output a sequence of
strings (written to a channel).

Filters can be chained together (e.g., via the Run function), the
output of one filter is fed as input to the next filter.  The empty
input is passed to the first filter. The following sequence will
print two lines to standard output:

	err := pipe.Run(
		pipe.Echo("hello", "world"),
		pipe.Reverse(),
		pipe.WriteLines(os.Stdout),
	)

An application can implement its own filters easily. For example,
repeat(n) returns a filter that repeats every input n times.

	func repeat(n int) Filter {
		return func(arg pipe.Arg) error {
			for s := range arg.In {
				for i := 0; i < n; i++ {
					arg.Out <- s
				}
			}
			return nil
		}
	}

	pipe.Run(
		pipe.Echo("hello"),
		repeat(10),
	)

Note that repeat is not a Filter since it needs to accept the
parameter n. Instead, it returns a Filter.  This convention is
followed throughout this library: all filtering functionality is
provided by functions that return a Filter.

*/
package pipe

import (
	"fmt"
	"sync"
)

// filterErrors records errors accumulated during the execution of a filter.
type filterErrors struct {
	mu     sync.Mutex
	errors []error
}

func (e *filterErrors) getError() error {
	e.mu.Lock()
	defer e.mu.Unlock()
	switch len(e.errors) {
	case 0:
		return nil
	case 1:
		return e.errors[0]
	default:
		return fmt.Errorf("pipe.Filter errors: %s", e.errors)
	}
}

// Arg contains the data passed to a Filter. Arg.In is a channel that
// produces the input to the filter, and Arg.Out is a channel that
// receives the output from the filter.
type Arg struct {
	In  <-chan string
	Out chan<- string
}

// Filter is the type of a function that reads a sequence of strings
// from a channel and produces a sequence on another channel. A Filter
// returns nil on success, an error otherwise.
type Filter func(Arg) error

// Sequence returns a filter that is the concatenation of all filter arguments.
// The output of a filter is fed as input to the next filter.
func Sequence(filters ...Filter) Filter {
	return func(arg Arg) error {
		e := &filterErrors{}
		in := arg.In
		for _, f := range filters {
			c := make(chan string, 10000)
			go runFilter(f, Arg{in, c}, e)
			in = c
		}
		for s := range in {
			arg.Out <- s
		}
		return e.getError()
	}
}

// Run executes the sequence of filters and discards all output.
// It returns either nil, an error if any filter reported an error.
func Run(filters ...Filter) error {
	return ForEach(Sequence(filters...), func(s string) {})
}

// ForEach calls fn(s) for every item s in the output of filter and
// returns either nil, or any error reported by the execution of the filter.
func ForEach(filter Filter, fn func(s string)) error {
	in := make(chan string, 0)
	close(in)
	out := make(chan string, 10000)
	e := &filterErrors{}
	go runFilter(filter, Arg{in, out}, e)
	for s := range out {
		fn(s)
	}
	return e.getError()
}

// Output returns a slice that contains all items that are
// the output of filters.
func Output(filters ...Filter) ([]string, error) {
	var result []string
	err := ForEach(Sequence(filters...), func(s string) {
		result = append(result, s)
	})
	if err != nil {
		result = nil // Discard results on error
	}
	return result, err
}

func runFilter(f Filter, arg Arg, e *filterErrors) {
	err := f(arg)
	close(arg.Out)
	for _ = range arg.In { // Discard all unhandled input
	}
	if err != nil {
		e.mu.Lock()
		e.errors = append(e.errors, err)
{'fraction_non_alphanumeric': 0.10285326086956521, 'fraction_numerical': 0.004619565217391305, 'mean_word_length': 3.3071971913399647, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '687704', 'n_tokens_mistral': 2423, 'n_tokens_neox': 2264, 'n_words': 1022}