I'm building a filtering pipeline with channels by function composition instead of 120+LOC all in a single method since I might reuse some part of this pipeline later.
I'm not been able to make it work as intended. I suspect that the funcion readValuesFromFile
is exiting before the scanner.Scan()
puts a value in the inputStream
channel (ie
that method's main goroutine is exiting before (1) goroutine).
If I replace the scanner.Scan()
with just putting some random string in the channel
the whole pipeline works as expected.
Is this the problem or I'm missing something?
How can this be fixed in an elegant way?
Thanks!
func readValuesFromFile(filename string) <-chan string {
file, err := os.Open(filename)
if err != nil {
log.Fatal(err)
}
defer file.Close()
inputStream := make(chan string)
go func() { //(1)
count := 0
scanner := bufio.NewScanner(file)
for scanner.Scan() { // (2)
inputStream <- strings.TrimSpace(scanner.Text())
count = count + 1
}
close(inputStream)
}()
return inputStream
}
func validateValues(inputStream <-chan string) <-chan string {
//read from the input stream + validate&filter + creating and putting values in an output stream
}
func writeResults(validStream <-chan string) {
//read from the validated stream and write data to file
}
func main() {
valueStream := readValuesFromFile("myfile.txt")
validatedStream := validateValues(valueStream)
writeResults(validatedStream)
}