diff --git a/cmd/analyze.go b/cmd/analyze.go index 7e83cfe..acc4a5d 100644 --- a/cmd/analyze.go +++ b/cmd/analyze.go @@ -65,7 +65,7 @@ func analyzeFunction(c *cli.Context) error { batcher := helpers.BuildBatcherFromArguments(c) ext := helpers.BuildExtractorFromArguments(c, batcher) - helpers.RunAggregationLoop(ext, aggr, func() { + interrupt := helpers.RunAggregationLoop(ext, aggr, func() { line := writeAggrOutput(writer, aggr, extra, quantiles) writer.WriteForLine(line+1, helpers.BuildExtractorSummary(ext, aggr.ParseErrors())) writer.WriteForLine(line+2, batcher.StatusString()) @@ -73,7 +73,7 @@ func analyzeFunction(c *cli.Context) error { writer.Close() - return helpers.DetermineErrorState(batcher, ext, aggr) + return helpers.DetermineErrorState(interrupt, batcher, ext, aggr) } func analyzeCommand() *cli.Command { diff --git a/cmd/bargraph.go b/cmd/bargraph.go index e18be29..8920556 100644 --- a/cmd/bargraph.go +++ b/cmd/bargraph.go @@ -43,7 +43,7 @@ func bargraphFunction(c *cli.Context) error { ext := helpers.BuildExtractorFromArguments(c, batcher) sorter := helpers.BuildSorterOrFail(sortName) - helpers.RunAggregationLoop(ext, counter, func() { + interrupt := helpers.RunAggregationLoop(ext, counter, func() { line := 0 writer.SetKeys(counter.SubKeys()...) @@ -62,7 +62,7 @@ func bargraphFunction(c *cli.Context) error { return err } - return helpers.DetermineErrorState(batcher, ext, counter) + return helpers.DetermineErrorState(interrupt, batcher, ext, counter) } func bargraphCommand() *cli.Command { diff --git a/cmd/filter.go b/cmd/filter.go index 02ec82d..e03bc7c 100644 --- a/cmd/filter.go +++ b/cmd/filter.go @@ -3,6 +3,7 @@ package cmd import ( "bufio" "os" + "os/signal" "unicode/utf8" "github.com/zix99/rare/cmd/helpers" @@ -27,40 +28,56 @@ func filterFunction(c *cli.Context, fileGlobs ...string) error { stdout := bufio.NewWriter(os.Stdout) + exitSignal := make(chan os.Signal, 1) + signal.Notify(exitSignal, os.Interrupt) + interrupted := false + + readChan := extractor.ReadFull() + OUTER_LOOP: - for matchBatch := range extractor.ReadFull() { - for _, match := range matchBatch { - if writeLines { - color.WriteString(stdout, color.BrightGreen, match.Source) - stdout.WriteByte(' ') - color.WriteUint64(stdout, color.BrightYellow, match.LineNumber) - stdout.WriteString(": ") + for { + select { + case <-exitSignal: + interrupted = true + break OUTER_LOOP + case matchBatch, more := <-readChan: + if !more { + break OUTER_LOOP } - switch { - case customExtractor: - stdout.WriteString(match.Extracted) - case onlyText && !utf8.ValidString(match.Line): - color.WriteString(stdout, color.BrightBlue, "Binary Match") - case len(match.Indices) == 2: - // Single match, highlight entire phrase - color.WrapIndices(stdout, match.Line, match.Indices) - default: - // Multi-match groups, highlight individual groups - color.WrapIndices(stdout, match.Line, match.Indices[2:]) + for _, match := range matchBatch { + if writeLines { + color.WriteString(stdout, color.BrightGreen, match.Source) + stdout.WriteByte(' ') + color.WriteUint64(stdout, color.BrightYellow, match.LineNumber) + stdout.WriteString(": ") + } + + switch { + case customExtractor: + stdout.WriteString(match.Extracted) + case onlyText && !utf8.ValidString(match.Line): + color.WriteString(stdout, color.BrightBlue, "Binary Match") + case len(match.Indices) == 2: + // Single match, highlight entire phrase + color.WrapIndices(stdout, match.Line, match.Indices) + default: + // Multi-match groups, highlight individual groups + color.WrapIndices(stdout, match.Line, match.Indices[2:]) + } + stdout.WriteByte('\n') + + readLines++ + if numLineLimit > 0 && readLines >= numLineLimit { + break OUTER_LOOP + } } - stdout.WriteByte('\n') - readLines++ - if numLineLimit > 0 && readLines >= numLineLimit { - break OUTER_LOOP + // Flush after each batch to make file-following work as expected + if err := stdout.Flush(); err != nil { + logger.Fatal(helpers.ExitCodeOutputError, err) } } - - // Flush after each batch to make file-following work as expected - if err := stdout.Flush(); err != nil { - logger.Fatal(helpers.ExitCodeOutputError, err) - } } // Final flush @@ -80,7 +97,7 @@ OUTER_LOOP: } os.Stderr.WriteString("\n") - return helpers.DetermineErrorState(batcher, extractor, nil) + return helpers.DetermineErrorState(interrupted, batcher, extractor, nil) } func getFilterArgs(isSearch bool) []cli.Flag { diff --git a/cmd/heatmap.go b/cmd/heatmap.go index 94553ce..9b20f54 100644 --- a/cmd/heatmap.go +++ b/cmd/heatmap.go @@ -47,7 +47,7 @@ func heatmapFunction(c *cli.Context) error { writer.Scaler = helpers.BuildScalerOrFail(scalerName) writer.Formatter = helpers.BuildFormatterOrFail(formatName) - helpers.RunAggregationLoop(ext, counter, func() { + interrupt := helpers.RunAggregationLoop(ext, counter, func() { writer.WriteTable(counter, rowSorter, colSorter) writer.WriteFooter(0, helpers.BuildExtractorSummary(ext, counter.ParseErrors(), fmt.Sprintf("(R: %v; C: %v)", color.Wrapi(color.Yellow, counter.RowCount()), color.Wrapi(color.BrightBlue, counter.ColumnCount())))) @@ -60,7 +60,7 @@ func heatmapFunction(c *cli.Context) error { return err } - return helpers.DetermineErrorState(batcher, ext, counter) + return helpers.DetermineErrorState(interrupt, batcher, ext, counter) } func heatmapCommand() *cli.Command { diff --git a/cmd/helpers/exitCodes.go b/cmd/helpers/exitCodes.go index ca1dc6c..20fbcaa 100644 --- a/cmd/helpers/exitCodes.go +++ b/cmd/helpers/exitCodes.go @@ -9,6 +9,7 @@ const ( ExitCodeInvalidUsage = 2 ExitCodeReadError = 3 ExitCodeOutputError = 4 + ExitCodeSigInt = 128 + 2 // 2 is SIGINT ) type ( @@ -23,7 +24,10 @@ type ( } ) -func DetermineErrorState(b BatcherErrors, e ExtractorSummary, agg AggregationErrors) error { +func DetermineErrorState(interrupt bool, b BatcherErrors, e ExtractorSummary, agg AggregationErrors) error { + if interrupt { + return cli.Exit("", ExitCodeSigInt) + } if b.ReadErrors() > 0 { return cli.Exit("Read errors", ExitCodeReadError) } diff --git a/cmd/helpers/exitCodes_test.go b/cmd/helpers/exitCodes_test.go index 0cfdc53..fdbf168 100644 --- a/cmd/helpers/exitCodes_test.go +++ b/cmd/helpers/exitCodes_test.go @@ -24,14 +24,17 @@ func (s *mockExitState) MatchedLines() uint64 { func TestDetermineErrorState(t *testing.T) { s := mockExitState{0, 0, 1} - assert.NoError(t, DetermineErrorState(&s, &s, &s)) + assert.NoError(t, DetermineErrorState(false, &s, &s, &s)) s = mockExitState{0, 0, 0} - assert.Error(t, DetermineErrorState(&s, &s, &s)) + assert.Error(t, DetermineErrorState(false, &s, &s, &s)) s = mockExitState{0, 1, 1} - assert.Error(t, DetermineErrorState(&s, &s, &s)) + assert.Error(t, DetermineErrorState(false, &s, &s, &s)) s = mockExitState{1, 0, 1} - assert.Error(t, DetermineErrorState(&s, &s, &s)) + assert.Error(t, DetermineErrorState(false, &s, &s, &s)) + + s = mockExitState{0, 0, 1} + assert.Error(t, DetermineErrorState(true, &s, &s, &s)) } diff --git a/cmd/helpers/updatingAggregator.go b/cmd/helpers/updatingAggregator.go index b010882..42a50d7 100644 --- a/cmd/helpers/updatingAggregator.go +++ b/cmd/helpers/updatingAggregator.go @@ -18,7 +18,7 @@ import ( // writeOutput - triggered after a delay, only if there's an update // // The two functions are guaranteed to never happen at the same time -func RunAggregationLoop(ext *extractor.Extractor, aggregator aggregation.Aggregator, writeOutput func()) { +func RunAggregationLoop(ext *extractor.Extractor, aggregator aggregation.Aggregator, writeOutput func()) (interrupt bool) { logger.DeferLogs() // Updater sync variables @@ -47,6 +47,7 @@ PROCESSING_LOOP: for { select { case <-exitSignal: + interrupt = true break PROCESSING_LOOP case matchBatch, more := <-reader: if !more { @@ -62,4 +63,6 @@ PROCESSING_LOOP: outputDone <- true writeOutput() + + return } diff --git a/cmd/histo.go b/cmd/histo.go index 38e6f2e..835adbf 100644 --- a/cmd/histo.go +++ b/cmd/histo.go @@ -57,7 +57,7 @@ func histoFunction(c *cli.Context) error { fmt.Sprintf("(Groups: %s)", color.Wrapi(color.BrightBlue, counter.GroupCount()))) } - helpers.RunAggregationLoop(ext, counter, func() { + interrupt := helpers.RunAggregationLoop(ext, counter, func() { writeHistoOutput(writer, counter, topItems, sorter, atLeast) writer.WriteFooter(0, progressString()) writer.WriteFooter(1, batcher.StatusString()) @@ -81,7 +81,7 @@ func histoFunction(c *cli.Context) error { return err } - return helpers.DetermineErrorState(batcher, ext, counter) + return helpers.DetermineErrorState(interrupt, batcher, ext, counter) } // HistogramCommand Exported command diff --git a/cmd/reduce.go b/cmd/reduce.go index 4fb7a3a..6c0fc35 100644 --- a/cmd/reduce.go +++ b/cmd/reduce.go @@ -72,6 +72,8 @@ func reduceFunction(c *cli.Context) error { formatters := buildFormatterSetOrFail(aggr, formatNames...) // run the aggregation + interrupted := false + if aggr.GroupColCount() > 0 || table { // Table output table := termrenderers.NewTable(vt, colCount, rowCount) @@ -88,7 +90,7 @@ func reduceFunction(c *cli.Context) error { table.WriteRow(0, rowBuf...) } - helpers.RunAggregationLoop(extractor, aggr, func() { + interrupted = helpers.RunAggregationLoop(extractor, aggr, func() { // write data for i, group := range aggr.Groups(sorter) { rowBuf := make([]string, aggr.ColCount()) @@ -110,7 +112,7 @@ func reduceFunction(c *cli.Context) error { }) } else { // Simple output - helpers.RunAggregationLoop(extractor, aggr, func() { + interrupted = helpers.RunAggregationLoop(extractor, aggr, func() { items := aggr.Data("") colNames := aggr.DataCols() for idx, expr := range items { @@ -127,7 +129,7 @@ func reduceFunction(c *cli.Context) error { return err } - return helpers.DetermineErrorState(batcher, extractor, aggr) + return helpers.DetermineErrorState(interrupted, batcher, extractor, aggr) } func parseKeyValInitial(s, defaultInitial string) (key, initial, val string) { diff --git a/cmd/spark.go b/cmd/spark.go index b7452ca..213afdf 100644 --- a/cmd/spark.go +++ b/cmd/spark.go @@ -38,7 +38,7 @@ func sparkFunction(c *cli.Context) error { writer.Scaler = helpers.BuildScalerOrFail(scalerName) writer.Formatter = helpers.BuildFormatterOrFail(formatName) - helpers.RunAggregationLoop(ext, counter, func() { + interrupt := helpers.RunAggregationLoop(ext, counter, func() { // Trim unused data from the data store (keep memory tidy!) if !noTruncate { @@ -69,7 +69,7 @@ func sparkFunction(c *cli.Context) error { return err } - return helpers.DetermineErrorState(batcher, ext, counter) + return helpers.DetermineErrorState(interrupt, batcher, ext, counter) } func sparkCommand() *cli.Command { diff --git a/cmd/tabulate.go b/cmd/tabulate.go index 7f9e44a..62a2485 100644 --- a/cmd/tabulate.go +++ b/cmd/tabulate.go @@ -40,7 +40,7 @@ func tabulateFunction(c *cli.Context) error { writer.SetFormatter(helpers.BuildFormatterOrFail(formatExp)) } - helpers.RunAggregationLoop(ext, counter, func() { + interrupt := helpers.RunAggregationLoop(ext, counter, func() { writer.WriteTable(counter, rowSorter, colSorter) writer.WriteFooter(0, helpers.BuildExtractorSummary(ext, counter.ParseErrors(), @@ -54,7 +54,7 @@ func tabulateFunction(c *cli.Context) error { return err } - return helpers.DetermineErrorState(batcher, ext, counter) + return helpers.DetermineErrorState(interrupt, batcher, ext, counter) } func tabulateCommand() *cli.Command {