Repository: kythe/kythe
Pull request #4983 (2021-07-12 22:32:57)

feat(proxy): Add support for transmitting protos in wire format
The protobuf crate for Rust currently does not support marshalling protobufs to JSON. This PR adds two new commands to the proxy: one to request a CompilationUnit in wire format, and one to send entries back to the proxy in wire format.

wcalandro:       
  

  
  
  
 The compiler throws an error if I don't have them, saying that there isn't a boolean on the left side.
  
  
  

  
  
>   
> On Jul 12, 2021 at 15:38,  <Shahms King ***@***.***)>  wrote:
>   
>   
>   
>
>
>
>   
>
> @shahms  commented on this pull request.
>
>     
>
> In  kythe/go/platform/analysis/proxy/proxy.go (https://github.com/kythe/kythe/pull/4983#discussion_r668301549):
>
>   
>   >  @@ -265,6 +325,26 @@ func decodeEntries(jsonArray json.RawMessage) ([]*spb.Entry, error) { return entries, nil } +func decodeWireEntries(jsonArray json.RawMessage) ([]*spb.Entry, error) { + var messages []string + if err := json.Unmarshal(jsonArray,  &messages); err != nil { + return nil, err + } + entries := make([]*spb.Entry, len(messages)) + for i, msg := range messages { + var e spb.Entry + b, err := base64.StdEncoding.DecodeString(msg) + if err != nil { + return nil, err + } + if err := (proto.UnmarshalOptions{}.Unmarshal(b,  &e)); err != nil {  
>   
>
> outer parens are unnecessary
>
>   
>
> —
>  You are receiving this because you authored the thread.
>  Reply to this email directly,   view it on GitHub (https://github.com/kythe/kythe/pull/4983#pullrequestreview-704598307), or   unsubscribe (https://github.com/notifications/unsubscribe-auth/APPKRXLR4CAPFR2EJXAHFB3TXNVH5ANCNFSM5AHXZT3Q).
>
>   
>   
  
  
       
creachadair: > The compiler throws an error if I don't have them, saying that there isn't a boolean on the left side.

This is a corner case in the grammar: When a struct literal appears at the end of the `if` initializer, the parentheses are necessary to disambiguate between the curly brackets ending the condition, and the brackets belonging to the initializer.

See the discussion about parsing ambiguity under https://golang.org/ref/spec#Composite_literals



Patch:
diff --git a/kythe/go/platform/analysis/proxy/BUILD b/kythe/go/platform/analysis/proxy/BUILD
index c8a940cd7..85e97e648 100644
--- a/kythe/go/platform/analysis/proxy/BUILD
+++ b/kythe/go/platform/analysis/proxy/BUILD
@@ -9,6 +9,7 @@ go_library(
         "//kythe/proto:analysis_go_proto",
         "//kythe/proto:storage_go_proto",
         "@org_golang_google_protobuf//encoding/protojson:go_default_library",
+        "@org_golang_google_protobuf//proto:go_default_library",
     ],
 )
 
diff --git a/kythe/go/platform/analysis/proxy/proxy.go b/kythe/go/platform/analysis/proxy/proxy.go
index 626b6e1bf..70e778082 100644
--- a/kythe/go/platform/analysis/proxy/proxy.go
+++ b/kythe/go/platform/analysis/proxy/proxy.go
@@ -32,10 +32,18 @@
 //   P → X: {"rsp":"ok","args":{"unit":<unit>,"rev":<revision>,"fds":<addr>}}<LF>
 //          {"rsp":"error","args":<error>}<LF>
 //
+//   X → P: {"req":"analysis_wire"}<LF>
+//   P → X: {"rsp":"ok","args":{"unit":<unit_wire>,"rev":<revision>,"fds":<addr>}}<LF>
+//          {"rsp":"error","args":<error>}<LF>
+//
 //   X → P: {"req":"output","args":[<entry>...]}<LF>
 //   P → X: {"rsp":"ok"}<LF>
 //          {"rsp":"error","args":<error>}<LF>
 //
+//   X → P: {"req":"output_wire","args":[<entry_wire>...]}<LF>
+//   P → X: {"rsp":"ok"}<LF>
+//          {"rsp":"error","args":<error>}<LF>
+//
 //   X → P: {"req":"done","args":{"ok":true,"msg":<error>}}<LF>
 //   P → X: {"rsp":"ok"}<LF>
 //          {"rsp":"error","args":<error>}<LF>
@@ -46,15 +54,17 @@
 //
 // Where:
 //
-//    <addr>     -- service address
-//    <bytes>    -- BASE-64 encoded bytes (string)
-//    <digest>   -- file content digest (string)
-//    <entry>    -- JSON encoded kythe.proto.Entry message
-//    <error>    -- error diagnostic (string)
-//    <path>     -- file path (string)
-//    <revision> -- revision marker (string)
-//    <unit>     -- JSON encoded kythe.proto.CompilationUnit message
-//    <LF>       -- line feed character (decimal code 10)
+//    <addr>       -- service address
+//    <bytes>      -- BASE-64 encoded bytes (string)
+//    <digest>     -- file content digest (string)
+//    <entry>      -- JSON encoded kythe.proto.Entry message
+//    <entry_wire> -- BASE-64 encoded kythe.proto.Entry message in wire format (string)
+//    <error>      -- error diagnostic (string)
+//    <path>       -- file path (string)
+//    <revision>   -- revision marker (string)
+//    <unit>       -- JSON encoded kythe.proto.CompilationUnit message
+//    <unit_wire>  -- BASE-64 encoded kythe.proto.CompilationUnit message in wire format (string)
+//    <LF>         -- line feed character (decimal code 10)
 //
 // When rsp="error" in a reply, the args are an error string. The ordinary flow
 // for the indexer is:
@@ -84,6 +94,7 @@ import (
 	"io"
 
 	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
 
 	apb "kythe.io/kythe/proto/analysis_go_proto"
 	spb "kythe.io/kythe/proto/storage_go_proto"
@@ -110,13 +121,20 @@ type response struct {
 	Args   interface{} `json:"args,omitempty"`
 }
 
-// A unit represents a compilation unit to be analyzed.
+// A unit represents a compilation unit (in JSON form) to be analyzed.
 type unit struct {
 	Unit            json.RawMessage `json:"unit"`
 	Revision        string          `json:"rev,omitempty"`
 	FileDataService string          `json:"fds,omitempty"`
 }
 
+// A unit represents a compilation unit (in base64-encoded wire format) to be analyzed.
+type unitWire struct {
+	Unit            []byte `json:"unit"`
+	Revision        string `json:"rev,omitempty"`
+	FileDataService string `json:"fds,omitempty"`
+}
+
 // A file represents a file request or content reply.
 type file struct {
 	Path    string `json:"path,omitempty"`
@@ -163,7 +181,7 @@ func (p *Proxy) Run(h Handler) error {
 				hasReq = true
 				u, err := protojson.MarshalOptions{UseProtoNames: true}.Marshal(req.Compilation)
 				if err != nil {
-					return fmt.Errorf("error marshalling compilation unit: %w", err)
+					return fmt.Errorf("error marshalling compilation unit as JSON: %w", err)
 				}
 				p.reply("ok", &unit{
 					Unit:            json.RawMessage(u),
@@ -172,6 +190,25 @@ func (p *Proxy) Run(h Handler) error {
 				})
 			}
 
+		case "analysis_wire":
+			// Prerequisite: There is no analysis request already in progress.
+			if hasReq {
+				p.reply("error", "an analysis is already in progress")
+			} else if req, err := h.Analysis(); err != nil {
+				p.reply("error", err.Error())
+			} else {
+				hasReq = true
+				u, err := proto.MarshalOptions{}.Marshal(req.Compilation)
+				if err != nil {
+					return fmt.Errorf("error marshalling compilation unit as wire format: %w", err)
+				}
+				p.reply("ok", &unitWire{
+					Unit:            u,
+					Revision:        req.Revision,
+					FileDataService: req.FileDataService,
+				})
+			}
+
 		case "output":
 			// Prerequisite: There is an analysis request in progress.
 			if !hasReq {
@@ -193,6 +230,27 @@ func (p *Proxy) Run(h Handler) error {
 			h.Done(err)
 			hasReq = false
 
+		case "output_wire":
+			// Prerequisite: There is an analysis request in progress.
+			if !hasReq {
+				p.reply("error", "no analysis is in progress")
+				break
+			}
+			entries, err := decodeWireEntries(req.Args)
+			if err != nil {
+				p.reply("error", "invalid entries: "+err.Error())
+			} else if err := h.Output(entries...); err != nil {
+				p.reply("error", err.Error())
+			} else {
+				p.reply("ok", nil)
+				break
+			}
+
+			// Analysis is now abandoned; report the error back to the
+			// handler and give up.
+			h.Done(err)
+			hasReq = false
+
 		case "done":
 			// Prerequisite: There is an analysis request in progress.
 			if !hasReq {
@@ -265,6 +323,22 @@ func decodeEntries(jsonArray json.RawMessage) ([]*spb.Entry, error) {
 	return entries, nil
 }
 
+func decodeWireEntries(jsonArray json.RawMessage) ([]*spb.Entry, error) {
+	var messages [][]byte
+	if err := json.Unmarshal(jsonArray, &messages); err != nil {
+		return nil, err
+	}
+	entries := make([]*spb.Entry, len(messages))
+	for i, msg := range messages {
+		var e spb.Entry
+		if err := (proto.UnmarshalOptions{}.Unmarshal(msg, &e)); err != nil {
+			return nil, err
+		}
+		entries[i] = &e
+	}
+	return entries, nil
+}
+
 func (p *Proxy) reply(status string, args interface{}) {
 	p.err = p.out.Encode(&response{
 		Status: status,


Test patch:
diff --git a/kythe/go/platform/analysis/proxy/proxy_test.go b/kythe/go/platform/analysis/proxy/proxy_test.go
index 4541e0d0c..c6a5a2a4d 100644
--- a/kythe/go/platform/analysis/proxy/proxy_test.go
+++ b/kythe/go/platform/analysis/proxy/proxy_test.go
@@ -24,6 +24,7 @@ import (
 	"testing"
 
 	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
 	"kythe.io/kythe/go/util/compare"
 
 	apb "kythe.io/kythe/proto/analysis_go_proto"
@@ -156,6 +157,7 @@ var (
 )
 
 const analysisReply = `{"rsp":"ok","args":{"fds":"Q","rev":"1","unit":{"v_name":{"signature":"test"}}}}`
+const analysisWireReply = `{"rsp":"ok","args":{"fds":"Q","rev":"1","unit":"CgYKBHRlc3Q="}}`
 
 func mustMarshal(v interface{}) string {
 	bits, err := json.Marshal(v)
@@ -182,6 +184,23 @@ func encodeEntries(es []*spb.Entry) json.RawMessage {
 	return msg
 }
 
+func encodeWireEntries(es []*spb.Entry) json.RawMessage {
+	var messages [][]byte
+	for _, e := range es {
+		rec, err := proto.MarshalOptions{}.Marshal(e)
+		if err != nil {
+			panic(err)
+		}
+		messages = append(messages, rec)
+	}
+
+	msg, err := json.Marshal(messages)
+	if err != nil {
+		panic(err)
+	}
+	return msg
+}
+
 func TestNOOP(t *testing.T) {
 	// Verify that startup and shutdown are clean.
 	if rsps, err := runProxy(handler{}); err != nil {
@@ -321,3 +340,57 @@ func TestAnalysisWorks(t *testing.T) {
 		t.Errorf("Wrong incorrect responses:\n got: %+v\nwant: %+v: %s", rsps, want, diff)
 	}
 }
+
+func TestWireAnalysisWorks(t *testing.T) {
+	// Enact a standard analyzer transaction, and verify that the responses work.
+	var doneCalled bool
+	var gotEntries []*spb.Entry
+
+	rsps, err := runProxy(handler{
+		done: func(err error) {
+			if err != nil {
+				t.Errorf("Done was called with an error: %v", err)
+			}
+			doneCalled = true
+		},
+		output: func(es ...*spb.Entry) error {
+			gotEntries = append(gotEntries, es...)
+			return nil
+		},
+		file: func(path, digest string) ([]byte, error) {
+			if path == "exists" {
+				return []byte("data"), nil
+			}
+			return nil, errors.New("notfound")
+		},
+	},
+		testreq{Type: "analysis_wire"},
+		testreq{Type: "file", Args: file{Path: "exists"}},
+		testreq{Type: "output_wire", Args: encodeWireEntries(testEntries)},
+		testreq{Type: "file", Args: file{Path: "does not exist"}},
+		testreq{Type: "done"},
+	)
+	if err != nil {
+		t.Errorf("Unexpected error from proxy: %v", err)
+	}
+
+	// Verify that the proxy followed the protocol reasonably.
+	if !doneCalled {
+		t.Error("The handler's Done method was never called")
+	}
+	if diff := compare.ProtoDiff(gotEntries, testEntries); diff != "" {
+		t.Errorf("Incorrect entries:\n got: %+v\nwant: %+v: %s", gotEntries, testEntries, diff)
+	}
+
+	// Verify that we got the expected replies back from the proxy.
+	want := []string{
+		analysisWireReply, // from Analyze Wire
+		`{"rsp":"ok","args":{"content":"ZGF0YQ==","path":"exists"}}`, // from File (1)
+		`{"rsp":"ok"}`,                      // from Output Wire
+		`{"rsp":"error","args":"notfound"}`, // from File (2)
+		`{"rsp":"ok"}`,                      // from Done
+	}
+	if diff := compare.ProtoDiff(rsps, want); diff != "" {
+		t.Errorf("Wrong incorrect responses:\n got: %+v\nwant: %+v: %s", rsps, want, diff)
+	}
+}