Repository: yarpc/yab
Pull request #349 (2021-09-01 12:47:37)

Yab Release v0.21.0


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0e66e8d..5a525a9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,11 @@
 Changelog
 =========
 
+# 0.21.0 (2021-09-01)
+* Fix gRPC server stream handling to be compatible with Java gRPC server.
+* Fix parsing of protobuf responses/error-details containing Any type fields, by maintaining
+  reflection server connection until end of the request.
+
 # 0.20.0 (2021-05-18)
 * Add `stream-delay-close-send` option which delays client send stream closure.
 * New: gRPC details are now printed along with the error if there are any.
diff --git a/encoding/protobuf.go b/encoding/protobuf.go
index 2ac9dc3..2dbf950 100644
--- a/encoding/protobuf.go
+++ b/encoding/protobuf.go
@@ -25,7 +25,7 @@ type protoSerializer struct {
 	serviceName string
 	methodName  string
 	method      *desc.MethodDescriptor
-	anyResolver jsonpb.AnyResolver
+	anyResolver anyResolver
 }
 
 // bytesMsg wraps a raw byte slice for serialization purposes. Especially
@@ -264,6 +264,12 @@ func (p protoStreamRequestReader) NextBody() ([]byte, error) {
 	return p.proto.encode(body)
 }
 
+// Close stops the protobuf reflection/file based descriptor provider.
+func (p protoSerializer) Close() error {
+	p.anyResolver.source.Close()
+	return nil
+}
+
 func splitMethod(fullMethod string) (svc, method string, err error) {
 	parts := strings.Split(fullMethod, "/")
 	switch len(parts) {
diff --git a/go.mod b/go.mod
index bc5b4c1..28b32ea 100644
--- a/go.mod
+++ b/go.mod
@@ -43,7 +43,7 @@ require (
 	go.uber.org/zap v1.10.0
 	golang.org/x/net v0.0.0-20190926025831-c00fd9afed17
 	golang.org/x/text v0.3.1-0.20180511172408-5c1cf69b5978 // indirect
-	google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8 // indirect
+	google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8
 	google.golang.org/grpc v1.24.0
 	gopkg.in/yaml.v2 v2.2.2
 )
diff --git a/handler.go b/handler.go
index 61cdd56..4b1739b 100644
--- a/handler.go
+++ b/handler.go
@@ -180,6 +180,10 @@ func makeServerStream(ctx context.Context, stream *yarpctransport.ClientStream,
 		return err
 	}
 
+	if err := closeSendStream(ctx, stream, 0); err != nil {
+		return err
+	}
+
 	for err == nil {
 		var resBody []byte
 		if resBody, err = receiveStreamMessage(ctx, stream); err != nil {
diff --git a/main.go b/main.go
index 58a0584..7f0e3b1 100644
--- a/main.go
+++ b/main.go
@@ -329,6 +329,9 @@ func runWithOptions(opts Options, out output, logger *zap.Logger) {
 	if err != nil {
 		out.Fatalf("Failed while parsing input: %v\n", err)
 	}
+	if serializerWithClose, ok := serializer.(io.Closer); ok {
+		defer serializerWithClose.Close()
+	}
 
 	tracer, closer := getTracer(opts, out)
 	if closer != nil {
diff --git a/request.go b/request.go
index 5ab7905..ecc7445 100644
--- a/request.go
+++ b/request.go
@@ -122,9 +122,6 @@ func NewSerializer(opts Options, resolved resolvedProtocolEncoding) (encoding.Se
 			return nil, err
 		}
 
-		// The descriptor is only used in the New function, so it's safe to defer Close.
-		defer descSource.Close()
-
 		return encoding.NewProtobuf(opts.ROpts.Procedure, descSource)
 	}
 
diff --git a/version.go b/version.go
index b2088fe..db19633 100644
--- a/version.go
+++ b/version.go
@@ -22,4 +22,4 @@ package main
 
 // versionString is the sem-ver version string for yab.
 // It will be bumped explicitly on releases.
-var versionString = "0.20.0"
+var versionString = "0.21.0"


Test patch:
diff --git a/integration_test.go b/integration_test.go
index 2ce1dba..d90d7f2 100644
--- a/integration_test.go
+++ b/integration_test.go
@@ -34,6 +34,7 @@ import (
 	"time"
 
 	"github.com/yarpc/yab/testdata/gen-go/integration"
+	testdataany "github.com/yarpc/yab/testdata/protobuf/any"
 	"github.com/yarpc/yab/testdata/protobuf/simple"
 	yintegration "github.com/yarpc/yab/testdata/yarpc/integration"
 	"github.com/yarpc/yab/testdata/yarpc/integration/fooserver"
@@ -54,9 +55,9 @@ import (
 	yhttp "go.uber.org/yarpc/transport/http"
 	ytchan "go.uber.org/yarpc/transport/tchannel"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/reflection"
 	rpb "google.golang.org/grpc/reflection/grpc_reflection_v1alpha"
-	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
@@ -383,7 +384,7 @@ func (s *simpleService) Baz(c context.Context, in *simple.Foo) (*simple.Foo, err
 
 	if in.Test == -1 {
 		st := status.New(codes.InvalidArgument, "invalid username")
-		st, err := st.WithDetails(in)
+		st, err := st.WithDetails(in, &testdataany.FooAny{Value: 123})
 		if err != nil {
 			// If this errored, it will always error
 			// here, so better panic so we can figure
@@ -1212,6 +1213,11 @@ func TestGRPCReflectionSource(t *testing.T) {
       "type.googleapis.com/Foo": {
         "test": -1
       }
+    },
+    {
+      "type.googleapis.com/FooAny": {
+        "value": 123
+      }
     }
   ]
 }