streaming: support JSON flag (#492)

This adds support for `-json` to streaming search. 

Events are split into individual results and written as JSON to stdout.
Each line contains 1 result with possibly many matches.

Author: stefanhengl

cmd/src/search.go

@@ -55,7 +55,7 @@ Other tips:
 		explainJSONFlag = flagSet.Bool("explain-json", false, "Explain the JSON output schema and exit.")
 		apiFlags        = api.NewFlags(flagSet)
 		lessFlag        = flagSet.Bool("less", true, "Pipe output to 'less -R' (only if stdout is terminal, and not json flag).")
-		streamFlag      = flagSet.Bool("stream", false, "Consume results as stream. Streaming search only supports a subset of flags and parameters: trace, insecure-skip-verify, display.")
+		streamFlag      = flagSet.Bool("stream", false, "Consume results as stream. Streaming search only supports a subset of flags and parameters: trace, insecure-skip-verify, display, json.")
 		display         = flagSet.Int("display", -1, "Limit the number of results that are displayed. Only supported together with stream flag. Statistics continue to report all results.")
 	)
 
@@ -68,6 +68,7 @@ Other tips:
 			opts := streaming.Opts{
 				Display: *display,
 				Trace:   apiFlags.Trace(),
+				Json:    *jsonFlag,
 			}
 			client := cfg.apiClient(apiFlags, flagSet.Output())
 			return streamSearch(flagSet.Arg(0), opts, client, os.Stdout)

cmd/src/search_stream.go

@@ -2,13 +2,15 @@ package main
 
 import (
 	"bytes"
+	"encoding/json"
 	"fmt"
 	"io"
 	"os"
 	"os/exec"
 	"regexp"
 	"strconv"
 	"strings"
+	"text/template"
 
 	"github.com/sourcegraph/src-cli/internal/api"
 	"github.com/sourcegraph/src-cli/internal/streaming"
@@ -21,20 +23,77 @@ func init() {
 }
 
 func streamSearch(query string, opts streaming.Opts, client api.Client, w io.Writer) error {
-	t, err := parseTemplate(streamingTemplate)
-	if err != nil {
-		panic(err)
+	var d streaming.Decoder
+	if opts.Json {
+		d = jsonDecoder(w)
+	} else {
+		t, err := parseTemplate(streamingTemplate)
+		if err != nil {
+			return err
+		}
+		d = textDecoder(query, t, w)
+	}
+	return streaming.Search(query, opts, client, d)
+}
+
+// jsonDecoder streams results as JSON to w.
+func jsonDecoder(w io.Writer) streaming.Decoder {
+	// write json.Marshals data and writes it as one line to w plus a newline.
+	write := func(data interface{}) error {
+		b, err := json.Marshal(data)
+		if err != nil {
+			return err
+		}
+		_, err = w.Write(b)
+		if err != nil {
+			return err
+		}
+		_, err = w.Write([]byte("\n"))
+		if err != nil {
+			return err
+		}
+		return nil
 	}
-	logError := func(msg string) {
-		_, _ = fmt.Fprintf(os.Stderr, msg)
+
+	return streaming.Decoder{
+		OnProgress: func(progress *streaming.Progress) {
+			if !progress.Done {
+				return
+			}
+			err := write(progress)
+			if err != nil {
+				logError(err.Error())
+			}
+		},
+		OnMatches: func(matches []streaming.EventMatch) {
+			for _, match := range matches {
+				err := write(match)
+				if err != nil {
+					logError(err.Error())
+				}
+			}
+		},
+		OnAlert: func(alert *streaming.EventAlert) {
+			err := write(alert)
+			if err != nil {
+				logError(err.Error())
+			}
+		},
+		OnError: func(eventError *streaming.EventError) {
+			// Errors are just written to stderr.
+			logError(eventError.Message)
+		},
 	}
-	decoder := streaming.Decoder{
+}
+
+func textDecoder(query string, t *template.Template, w io.Writer) streaming.Decoder {
+	return streaming.Decoder{
 		OnProgress: func(progress *streaming.Progress) {
 			// We only show the final progress.
 			if !progress.Done {
 				return
 			}
-			err = t.ExecuteTemplate(w, "progress", progress)
+			err := t.ExecuteTemplate(w, "progress", progress)
 			if err != nil {
 				logError(fmt.Sprintf("error when executing template: %s\n", err))
 			}
@@ -52,7 +111,7 @@ func streamSearch(query string, opts streaming.Opts, client api.Client, w io.Wri
 				})
 			}
 
-			err = t.ExecuteTemplate(w, "alert", searchResultsAlert{
+			err := t.ExecuteTemplate(w, "alert", searchResultsAlert{
 				Title:           alert.Title,
 				Description:     alert.Description,
 				ProposedQueries: proposedQueries,
@@ -66,7 +125,7 @@ func streamSearch(query string, opts streaming.Opts, client api.Client, w io.Wri
 			for _, match := range matches {
 				switch match := match.(type) {
 				case *streaming.EventFileMatch:
-					err = t.ExecuteTemplate(w, "file", struct {
+					err := t.ExecuteTemplate(w, "file", struct {
 						Query string
 						*streaming.EventFileMatch
 					}{
@@ -79,7 +138,7 @@ func streamSearch(query string, opts streaming.Opts, client api.Client, w io.Wri
 						return
 					}
 				case *streaming.EventRepoMatch:
-					err = t.ExecuteTemplate(w, "repo", struct {
+					err := t.ExecuteTemplate(w, "repo", struct {
 						SourcegraphEndpoint string
 						*streaming.EventRepoMatch
 					}{
@@ -91,7 +150,7 @@ func streamSearch(query string, opts streaming.Opts, client api.Client, w io.Wri
 						return
 					}
 				case *streaming.EventCommitMatch:
-					err = t.ExecuteTemplate(w, "commit", struct {
+					err := t.ExecuteTemplate(w, "commit", struct {
 						SourcegraphEndpoint string
 						*streaming.EventCommitMatch
 					}{
@@ -103,7 +162,7 @@ func streamSearch(query string, opts streaming.Opts, client api.Client, w io.Wri
 						return
 					}
 				case *streaming.EventSymbolMatch:
-					err = t.ExecuteTemplate(w, "symbol", struct {
+					err := t.ExecuteTemplate(w, "symbol", struct {
 						SourcegraphEndpoint string
 						*streaming.EventSymbolMatch
 					}{
@@ -119,8 +178,6 @@ func streamSearch(query string, opts streaming.Opts, client api.Client, w io.Wri
 			}
 		},
 	}
-
-	return streaming.Search(query, opts, client, decoder)
 }
 
 const streamingTemplate = `
@@ -362,3 +419,7 @@ func streamConvertMatchToHighlights(m streaming.EventLineMatch, isPreview bool)
 	}
 	return highlights
 }
+
+func logError(msg string) {
+	_, _ = fmt.Fprintf(os.Stderr, msg)
+}