inprocgrpc should support dynamic messages (#32)

go.mod

@@ -2,7 +2,7 @@ module github.com/fullstorydev/grpchan
 
 require (
 	github.com/golang/protobuf v1.1.0
-	github.com/jhump/protoreflect v1.0.0
+	github.com/jhump/protoreflect v1.2.0
 	golang.org/x/net v0.0.0-20180530234432-1e491301e022
 	google.golang.org/genproto v0.0.0-20170818100345-ee236bd376b0
 	google.golang.org/grpc v1.11.0

go.sum

@@ -1,7 +1,7 @@
 github.com/golang/protobuf v1.1.0 h1:0iH4Ffd/meGoXqF2lSAhZHt8X+cPgkfn/cb6Cce5Vpc=
 github.com/golang/protobuf v1.1.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-github.com/jhump/protoreflect v1.0.0 h1:l94KtQ6gRI3ouKVcXNdofCQJWoHATzcI6tDizOgUaf0=
-github.com/jhump/protoreflect v1.0.0/go.mod h1:kG/zRVeS2M91gYaCvvUbPkMjjtFQS4qqjcPFzFkh2zE=
+github.com/jhump/protoreflect v1.2.0 h1:W4HDXsqxgmE9OUmEXWaU2ZoUap/A29J5hW+zRXWbdC8=
+github.com/jhump/protoreflect v1.2.0/go.mod h1:kG/zRVeS2M91gYaCvvUbPkMjjtFQS4qqjcPFzFkh2zE=
 golang.org/x/net v0.0.0-20180530234432-1e491301e022 h1:MVYFTUmVD3/+ERcvRRI+P/C2+WOUimXh+Pd8LVsklZ4=
 golang.org/x/net v0.0.0-20180530234432-1e491301e022/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=

inprocgrpc/in_process_test.go

@@ -1,12 +1,20 @@
 package inprocgrpc_test
 
 import (
+	"bytes"
+	"reflect"
 	"runtime"
 	"testing"
 	"time"
 
 	"github.com/fullstorydev/grpchan/grpchantesting"
 	"github.com/fullstorydev/grpchan/inprocgrpc"
+	"github.com/jhump/protoreflect/desc"
+	"github.com/jhump/protoreflect/dynamic"
+	"github.com/jhump/protoreflect/dynamic/grpcdynamic"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
 )
 
 func TestInProcessChannel(t *testing.T) {
@@ -32,3 +40,89 @@ func TestInProcessChannel(t *testing.T) {
 	}
 	t.Errorf("%d goroutines leaked", after-before)
 }
+
+func TestUseDynamicMessage(t *testing.T) {
+	// This uses dynamic messages for request and response and
+	// ensures the in-process channel works correctly that way.
+
+	svr := &grpchantesting.TestServer{}
+
+	var cc inprocgrpc.Channel
+	grpchantesting.RegisterHandlerTestService(&cc, svr)
+	stub := grpcdynamic.NewStub(&cc)
+
+	fd, err := desc.LoadFileDescriptor("test.proto")
+	if err != nil {
+		t.Fatalf("failed to load descriptor for test.proto: %v", err)
+	}
+	md := fd.FindMessage("grpchantesting.Message")
+	if md == nil {
+		t.Fatalf("could not find descriptor for grpchantesting.Message")
+	}
+	sd := fd.FindService("grpchantesting.TestService")
+	if sd == nil {
+		t.Fatalf("could not find descriptor for grpchantesting.TestService")
+	}
+	mtd := sd.FindMethodByName("Unary")
+	if mtd == nil {
+		t.Fatalf("could not find descriptor for grpchantesting.TestService/Unary")
+	}
+
+	testPayload := []byte{100, 90, 80, 70, 60, 50, 40, 30, 20, 10, 0}
+	testOutgoingMd := map[string]string{
+		"foo": "bar",
+	}
+	testMdHeaders := map[string]string{
+		"foo1": "bar2",
+	}
+	testMdTrailers := map[string]string{
+		"foo3": "bar4",
+	}
+
+	ctx := metadata.NewOutgoingContext(context.Background(), metadata.New(testOutgoingMd))
+	req := dynamic.NewMessage(md)
+	req.SetFieldByName("payload", testPayload)
+	req.SetFieldByName("headers", testMdHeaders)
+	req.SetFieldByName("trailers", testMdTrailers)
+
+	var hdr, tlr metadata.MD
+	rsp, err := stub.InvokeRpc(ctx, mtd, req, grpc.Header(&hdr), grpc.Trailer(&tlr))
+	if err != nil {
+		t.Fatalf("RPC failed: %v", err)
+	}
+	msg := rsp.(*dynamic.Message)
+
+	payload := msg.GetFieldByName("payload")
+	if !bytes.Equal(testPayload, payload.([]byte)) {
+		t.Fatalf("wrong payload returned: expecting %v; got %v", testPayload, payload)
+	}
+	reqHeaders := map[string]string{}
+	for k, v := range msg.GetFieldByName("headers").(map[interface{}]interface{}) {
+		reqHeaders[k.(string)] = v.(string)
+	}
+	if !reflect.DeepEqual(testOutgoingMd, reqHeaders) {
+		t.Fatalf("wrong request headers echoed back: expecting %v; got %v", testOutgoingMd, reqHeaders)
+	}
+
+	actualHdrs := map[string]string{}
+	for k, v := range hdr {
+		if len(v) > 1 {
+			t.Fatalf("too many values for response header %q", k)
+		}
+		actualHdrs[k] = v[0]
+	}
+	if !reflect.DeepEqual(testMdHeaders, actualHdrs) {
+		t.Fatalf("wrong response headers echoed back: expecting %v; got %v", testMdHeaders, actualHdrs)
+	}
+
+	actualTlrs := map[string]string{}
+	for k, v := range tlr {
+		if len(v) > 1 {
+			t.Fatalf("too many values for response trailer %q", k)
+		}
+		actualTlrs[k] = v[0]
+	}
+	if !reflect.DeepEqual(testMdTrailers, actualTlrs) {
+		t.Fatalf("wrong response trailers echoed back: expecting %v; got %v", testMdTrailers, actualTlrs)
+	}
+}

internal/misc.go

@@ -5,6 +5,7 @@ import (
 	"reflect"
 
 	"github.com/golang/protobuf/proto"
+	"github.com/jhump/protoreflect/dynamic"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -19,19 +20,15 @@ func CopyMessage(out, in interface{}) error {
 	if !ok {
 		return fmt.Errorf("value to copy is not a proto.Message: %T; use a custom cloner", in)
 	}
-	if reflect.TypeOf(in) != reflect.TypeOf(out) {
-		return fmt.Errorf("incompatible types: %v != %v", reflect.TypeOf(in), reflect.TypeOf(out))
-	}
-	rvOut := reflect.ValueOf(out)
-	if rvOut.Kind() == reflect.Ptr && rvOut.IsNil() {
-		return fmt.Errorf("copy destination cannot be nil")
+	pmOut, ok := out.(proto.Message)
+	if !ok {
+		return fmt.Errorf("destination for copy is not a proto.Message: %T; use a custom cloner", in)
 	}
 
-	// this does a proper deep copy
-	pmOut := out.(proto.Message)
 	pmOut.Reset()
-	proto.Merge(pmOut, pmIn)
-	return nil
+	// This will check that types are compatible and return an error if not.
+	// Unlike proto.Merge, this allows one or the other to be a dynamic message.
+	return dynamic.TryMerge(pmOut, pmIn)
 }
 
 // CloneMessage returns a copy of the given value.