Update exit signals to respect interrupts

cmd/analyze.go

@@ -65,15 +65,15 @@ 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())
 	})
 
 	writer.Close()
 
-	return helpers.DetermineErrorState(batcher, ext, aggr)
+	return helpers.DetermineErrorState(interrupt, batcher, ext, aggr)
 }
 
 func analyzeCommand() *cli.Command {

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 {

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 {

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 {

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)
 	}

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))
 }

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
 }

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

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) {

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 {

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 {