1 files changed, 135 insertions, 0 deletions
diff --git a/examples/countlines.go b/examples/countlines.go
new file mode 100644
index 0000000..000efa5
--- /dev/null
+++ b/examples/countlines.go
@@ -0,0 +1,135 @@
+package main
+
+import (
+	"bufio"
+	"errors"
+	"flag"
+	"fmt"
+	"os"
+	"path/filepath"
+	"unicode/utf8"
+
+	"git.sr.ht/~gabrielgio/pipe"
+)
+
+var (
+	InvalidInput = errors.New("It was given a file instead of a folder.")
+	ColorGreen   = "\033[32m"
+	ColorRed     = "\033[31m"
+	ColorReset   = "\033[0m"
+)
+
+type (
+	NonTextError struct {
+		Filename string
+	}
+)
+
+func NewNonTextError(filename string) *NonTextError {
+	return &NonTextError{Filename: filename}
+}
+
+func (b *NonTextError) Error() string {
+	return fmt.Sprintf("Non text file %s", b.Filename)
+}
+
+func main() {
+	err := run()
+	if err != nil {
+		fmt.Printf("Error on main process: %s", err.Error())
+	}
+}
+
+func run() error {
+	var (
+		dir   = flag.String("dir", "", "Folder to recursively search for files. Default: \"\"")
+		quiet = flag.Bool("quiet", false, "If true only outputs the result. Default: false")
+		nproc = flag.Int("nproc", 4, "Number of goroutines used to count. Default: 4")
+	)
+
+	flag.Parse()
+
+	info, err := os.Stat(*dir)
+	if err != nil {
+		return err
+	}
+
+	if !info.IsDir() {
+		return InvalidInput
+	}
+
+	cerr := make(chan error)
+	go func(c <-chan error) {
+		for e := range c {
+			if !*quiet {
+				fmt.Printf("%sERROR%s: %s\n", ColorRed, ColorReset, e.Error())
+			}
+		}
+	}(cerr)
+
+	c := count(*dir, *nproc, cerr)
+	fmt.Printf("%sCOUNT%s: %d\n", ColorGreen, ColorReset, c)
+
+	return nil
+}
+
+func count(dir string, nproc int, cerr chan<- error) int {
+	cfiles := walkFolder(dir)
+	ccount := pipe.ProcWithError(cfiles, cerr, nproc, countLines)
+	return pipe.TailReduceWithError(ccount, cerr, sum, 0)
+}
+
+func sum(acc, lines int) (int, error) {
+	return acc + lines, nil
+}
+
+func countLines(filename string) (int, error) {
+
+	file, err := os.Open(filename)
+	if err != nil {
+		return 0, err
+	}
+
+	defer file.Close()
+
+	fileScanner := bufio.NewScanner(file)
+	fileScanner.Split(bufio.ScanLines)
+
+	var count int
+	for fileScanner.Scan() {
+		if !utf8.ValidString(string(fileScanner.Text())) {
+			return 0, NewNonTextError(filename)
+		}
+		count++
+	}
+
+	return count, nil
+}
+
+func walkFolder(folder string) <-chan string {
+	c := make(chan string)
+
+	go func(folder string, c chan string) {
+		filepath.Walk(folder, func(path string, info os.FileInfo, err error) error {
+			file, err := os.Open(path)
+			if err != nil {
+				return filepath.SkipDir
+			}
+			defer file.Close()
+
+			fileInfo, err := file.Stat()
+			if err != nil {
+				return filepath.SkipDir
+			}
+
+			if !fileInfo.IsDir() {
+				c <- path
+			}
+			return nil
+		})
+		close(c)
+
+	}(folder, c)
+
+	return c
+}