From d4aec5c2aa2b671fdd29c766392283b89deb6b67 Mon Sep 17 00:00:00 2001 From: John Kerl Date: Tue, 25 Aug 2020 21:44:09 -0400 Subject: [PATCH] streamer iterate --- go/mlr.go | 62 +++----------------------------- go/src/input/channel_reader.go | 40 +++++++++++++++++++++ go/src/mapping/channel_mapper.go | 19 ++++++++++ go/src/mapping/mapper_foo.go | 14 ++++++++ go/src/output/channel_writer.go | 24 +++++++++++++ go/src/stream/argf.go | 23 ++++++++++++ go/src/stream/stream.go | 60 ++++++++++++++----------------- 7 files changed, 151 insertions(+), 91 deletions(-) create mode 100644 go/src/input/channel_reader.go create mode 100644 go/src/mapping/channel_mapper.go create mode 100644 go/src/mapping/mapper_foo.go create mode 100644 go/src/output/channel_writer.go create mode 100644 go/src/stream/argf.go diff --git a/go/mlr.go b/go/mlr.go index b1ce0daad..53dbeafc5 100644 --- a/go/mlr.go +++ b/go/mlr.go @@ -2,17 +2,13 @@ package main import ( // System: - "bufio" "flag" "fmt" - "io" "log" "os" "runtime/pprof" - "strings" // Miller: - //"containers" - "input" + "stream" ) // ---------------------------------------------------------------- @@ -49,58 +45,8 @@ func main() { defer pprof.StopCPUProfile() } - if len(args) == 0 { - err := stream("-") - if err != nil { - log.Fatal(err) - } - } else { - for _, arg := range args { - err := stream(arg) - if err != nil { - log.Fatal(err) - } - } + err := stream.Stream(args) + if err != nil { + log.Fatal(err) } } - -// ---------------------------------------------------------------- -func stream(sourceName string) error { - inputStream := os.Stdin - if sourceName != "-" { - var err error - if inputStream, err = os.Open(sourceName); err != nil { - return err - } - } - - reader := bufio.NewReader(inputStream) - - eof := false - - for !eof { - line, err := reader.ReadString('\n') - if err == io.EOF { - err = nil - eof = true - } else if err != nil { - return err - } else { - if false { - fmt.Print(line) - } else { - // This is how to do a chomp: - line = strings.TrimRight(line, "\n") - - // xxx temp - ifs := "," - ips := "=" - lrec := input.LrecFromDKVPLine(&line, &ifs, &ips) - - lrec.Print(os.Stdout) - } - } - } - - return nil -} diff --git a/go/src/input/channel_reader.go b/go/src/input/channel_reader.go new file mode 100644 index 000000000..1b5057ca6 --- /dev/null +++ b/go/src/input/channel_reader.go @@ -0,0 +1,40 @@ +package input + +import ( + // System: + "bufio" + "io" + "strings" + // Miller: + "containers" +) + +func ChannelReader( + reader *bufio.Reader, + inrecs chan<- *containers.Lrec, + echan chan error, +) { + eof := false + + for !eof { + line, err := reader.ReadString('\n') + if err == io.EOF { + err = nil + eof = true + } else if err != nil { + echan <- err + } else { + + // This is how to do a chomp: + line = strings.TrimRight(line, "\n") + + // xxx temp + ifs := "," + ips := "=" + lrec := LrecFromDKVPLine(&line, &ifs, &ips) + inrecs <- lrec + } + } + + inrecs <- nil // signals end of input record stream +} diff --git a/go/src/mapping/channel_mapper.go b/go/src/mapping/channel_mapper.go new file mode 100644 index 000000000..f4f9263e4 --- /dev/null +++ b/go/src/mapping/channel_mapper.go @@ -0,0 +1,19 @@ +package mapping + +import ( + "containers" +) + +func ChannelMapper( + inrecs <-chan *containers.Lrec, + outrecs chan<- *containers.Lrec, +) { + for { + lrec := <-inrecs + if lrec == nil { + outrecs <- nil + break + } + MapperFoo(lrec, outrecs) + } +} diff --git a/go/src/mapping/mapper_foo.go b/go/src/mapping/mapper_foo.go new file mode 100644 index 000000000..76f6d8ecf --- /dev/null +++ b/go/src/mapping/mapper_foo.go @@ -0,0 +1,14 @@ +package mapping + +import ( + "containers" +) + +func MapperFoo(lrec *containers.Lrec, dest chan<- *containers.Lrec) { + k := "foo" + v := "bar" + // To-do: put-by-value variant + lrec.Put(&k, &v) + dest <- lrec + dest <- lrec +} diff --git a/go/src/output/channel_writer.go b/go/src/output/channel_writer.go new file mode 100644 index 000000000..08f42ae4a --- /dev/null +++ b/go/src/output/channel_writer.go @@ -0,0 +1,24 @@ +package output + +import ( + // System: + "os" + // Miller: + "containers" +) + +func ChannelWriter( + ostream *os.File, + outrecs <-chan *containers.Lrec, + done chan<- bool, +) { + for { + lrec := <-outrecs + if lrec == nil { + done <- true + break + } else { + lrec.Print(os.Stdout) + } + } +} diff --git a/go/src/stream/argf.go b/go/src/stream/argf.go new file mode 100644 index 000000000..7903e7173 --- /dev/null +++ b/go/src/stream/argf.go @@ -0,0 +1,23 @@ +package stream + +import ( + "io" + "os" +) + +func Argf(filenames []string) (io.Reader, error) { + if len(filenames) == 0 { + return os.Stdin, nil + } else { + readers := make([]io.Reader, len(filenames)) + for i, filename := range filenames { + handle, err := os.Open(filename) + if err == nil { + readers[i] = handle + } else { + return nil, err + } + } + return io.MultiReader(readers...), nil + } +} diff --git a/go/src/stream/stream.go b/go/src/stream/stream.go index 95565e560..6bd705cec 100644 --- a/go/src/stream/stream.go +++ b/go/src/stream/stream.go @@ -3,48 +3,42 @@ package stream import ( // System: "bufio" - "fmt" - "io" + "log" "os" - "strings" // Miller: + "containers" "input" + "mapping" + "output" ) -func Stream(sourceName string) error { - inputStream := os.Stdin - if sourceName != "-" { - var err error - if inputStream, err = os.Open(sourceName); err != nil { - return err - } +// ---------------------------------------------------------------- +func Stream(filenames []string) error { + istream, err := Argf(filenames) + if err != nil { + return err + os.Exit(1) } + reader := bufio.NewReader(istream) - reader := bufio.NewReader(inputStream) + inrecs := make(chan *containers.Lrec, 10) + echan := make(chan error, 1) + outrecs := make(chan *containers.Lrec, 1) + donechan := make(chan bool, 1) - eof := false + go input.ChannelReader(reader, inrecs, echan) + // TODO: needs mappers ... + go mapping.ChannelMapper(inrecs, outrecs) + go output.ChannelWriter(os.Stdout, outrecs, donechan) - for !eof { - line, err := reader.ReadString('\n') - if err == io.EOF { - err = nil - eof = true - } else if err != nil { - return err - } else { - if false { - fmt.Print(line) - } else { - // This is how to do a chomp: - line = strings.TrimRight(line, "\n") - - // xxx temp - ifs := "," - ips := "=" - lrec := input.LrecFromDKVPLine(&line, &ifs, &ips) - - lrec.Print(os.Stdout) - } + done := false + for !done { + select { + case err := <-echan: + log.Fatal(err) + case _ = <-donechan: + done = true + break } }