mirror of
https://github.com/johnkerl/miller.git
synced 2026-07-22 15:37:59 +00:00
82 lines
2.6 KiB
Go
82 lines
2.6 KiB
Go
package stream
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
|
|
"miller/src/cliutil"
|
|
"miller/src/input"
|
|
"miller/src/lib"
|
|
"miller/src/output"
|
|
"miller/src/transforming"
|
|
"miller/src/types"
|
|
)
|
|
|
|
// Since Go is concurrent, the context struct (AWK-like variables such as
|
|
// FILENAME, NF, NF, FNR, etc.) needs to be duplicated and passed through the
|
|
// channels along with each record.
|
|
//
|
|
// * Record-readers update FILENAME, FILENUM, NF, NR, FNR within context structs.
|
|
//
|
|
// * Record-transformers can read these from the context structs.
|
|
//
|
|
// * Record-writers don't need them (OPS et al. are already in the
|
|
// writer-options struct). However, we have chained transformers using the
|
|
// 'then' command-line syntax. This means a given transformer might be piping
|
|
// its output to a record-writer, or another transformer. So, the
|
|
// record-and-context pair goes to the record-writers even though they don't
|
|
// need the contexts.
|
|
|
|
func Stream(
|
|
options cliutil.TOptions,
|
|
recordTransformers []transforming.IRecordTransformer,
|
|
) error {
|
|
|
|
// Since Go is concurrent, the context struct needs to be duplicated and
|
|
// passed through the channels along with each record.
|
|
initialContext := types.NewContext(&options)
|
|
|
|
// Instantiate the record-reader
|
|
recordReader := input.Create(&options.ReaderOptions)
|
|
if recordReader == nil {
|
|
return errors.New("Input format not found: " + options.ReaderOptions.InputFileFormat)
|
|
}
|
|
|
|
// Instantiate the record-writer
|
|
recordWriter := output.Create(&options.WriterOptions)
|
|
if recordWriter == nil {
|
|
return errors.New("Output format not found: " + options.WriterOptions.OutputFileFormat)
|
|
}
|
|
|
|
// Set up the reader-to-transformer and transformer-to-writer channels.
|
|
inputChannel := make(chan *types.RecordAndContext, 10)
|
|
outputChannel := make(chan *types.RecordAndContext, 1)
|
|
|
|
// We're done when a fatal error is registered on input (file not found,
|
|
// etc) or when the record-writer has written all its output. We use
|
|
// channels to communicate both of these conditions.
|
|
errorChannel := make(chan error, 1)
|
|
doneChannel := make(chan bool, 1)
|
|
|
|
// Start the reader, transformer, and writer. Let them run until fatal input
|
|
// error or end-of-processing happens.
|
|
|
|
go recordReader.Read(options.FileNames, *initialContext, inputChannel, errorChannel)
|
|
go transforming.ChainTransformer(inputChannel, recordTransformers, outputChannel)
|
|
go output.ChannelWriter(outputChannel, recordWriter, doneChannel, os.Stdout)
|
|
|
|
done := false
|
|
for !done {
|
|
select {
|
|
case err := <-errorChannel:
|
|
fmt.Fprintln(os.Stderr, lib.MlrExeName(), ": ", err)
|
|
os.Exit(1)
|
|
case _ = <-doneChannel:
|
|
done = true
|
|
break
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|