interceptor can also see default call options setup by dial option; add test case

Author: jhump

call.go

@@ -27,6 +27,10 @@ import (
 //
 // All errors returned by Invoke are compatible with the status package.
 func (cc *ClientConn) Invoke(ctx context.Context, method string, args, reply interface{}, opts ...CallOption) error {
+	// allow interceptor to see all applicable call options, which means those
+	// configured as defaults from dial option as well as per-call options
+	opts = append(cc.dopts.callOptions, opts...)
+
 	if cc.dopts.unaryInt != nil {
 		return cc.dopts.unaryInt(ctx, method, args, reply, cc, invoke, opts...)
 	}

stream.go

@@ -99,6 +99,10 @@ type ClientStream interface {
 // NewStream creates a new Stream for the client side. This is typically
 // called by generated code.
 func (cc *ClientConn) NewStream(ctx context.Context, desc *StreamDesc, method string, opts ...CallOption) (ClientStream, error) {
+	// allow interceptor to see all applicable call options, which means those
+	// configured as defaults from dial option as well as per-call options
+	opts = append(cc.dopts.callOptions, opts...)
+
 	if cc.dopts.streamInt != nil {
 		return cc.dopts.streamInt(ctx, desc, cc, method, newClientStream, opts...)
 	}
@@ -137,7 +141,6 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 		}
 	}()
 
-	opts = append(cc.dopts.callOptions, opts...)
 	for _, o := range opts {
 		if err := o.before(c); err != nil {
 			return nil, toRPCErr(err)

test/end2end_test.go

@@ -627,14 +627,11 @@ func (d *nopDecompressor) Type() string {
 	return "nop"
 }
 
-func (te *test) clientConn() *grpc.ClientConn {
+func (te *test) clientConn(opts ...grpc.DialOption) *grpc.ClientConn {
 	if te.cc != nil {
 		return te.cc
 	}
-	opts := []grpc.DialOption{
-		grpc.WithDialer(te.e.dialer),
-		grpc.WithUserAgent(te.userAgent),
-	}
+	opts = append(opts, grpc.WithDialer(te.e.dialer), grpc.WithUserAgent(te.userAgent))
 
 	if te.sc != nil {
 		opts = append(opts, grpc.WithServiceConfig(te.sc))
@@ -5887,6 +5884,115 @@ func TestMethodFromServerStream(t *testing.T) {
 	}
 }
 
+func TestInterceptorCanAccessCallOptions(t *testing.T) {
+	defer leakcheck.Check(t)
+	e := tcpClearRREnv
+	te := newTest(t, e)
+	te.startServer(&testServer{security: e.security})
+	defer te.tearDown()
+
+	type observedOptions struct {
+		headers     []*metadata.MD
+		trailers    []*metadata.MD
+		peer        []*peer.Peer
+		creds       []credentials.PerRPCCredentials
+		failFast    []bool
+		maxRecvSize []int
+		maxSendSize []int
+		compressor  []string
+		subtype     []string
+		codec       []grpc.Codec
+	}
+	var observedOpts observedOptions
+	populateOpts := func(opts []grpc.CallOption) {
+		for _, o := range opts {
+			switch o := o.(type) {
+			case grpc.HeaderCallOption:
+				observedOpts.headers = append(observedOpts.headers, o.HeaderAddr)
+			case grpc.TrailerCallOption:
+				observedOpts.trailers = append(observedOpts.trailers, o.TrailerAddr)
+			case grpc.PeerCallOption:
+				observedOpts.peer = append(observedOpts.peer, o.PeerAddr)
+			case grpc.PerRPCCredsCallOption:
+				observedOpts.creds = append(observedOpts.creds, o.Creds)
+			case grpc.FailFastCallOption:
+				observedOpts.failFast = append(observedOpts.failFast, o.FailFast)
+			case grpc.MaxRecvMsgSizeCallOption:
+				observedOpts.maxRecvSize = append(observedOpts.maxRecvSize, o.MaxRecvMsgSize)
+			case grpc.MaxSendMsgSizeCallOption:
+				observedOpts.maxSendSize = append(observedOpts.maxSendSize, o.MaxSendMsgSize)
+			case grpc.CompressorCallOption:
+				observedOpts.compressor = append(observedOpts.compressor, o.CompressorType)
+			case grpc.ContentSubtypeCallOption:
+				observedOpts.subtype = append(observedOpts.subtype, o.ContentSubtype)
+			case grpc.CustomCodecCallOption:
+				observedOpts.codec = append(observedOpts.codec, o.Codec)
+			}
+		}
+	}
+
+	te.unaryClientInt = func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		populateOpts(opts)
+		return nil
+	}
+	te.streamClientInt = func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
+		populateOpts(opts)
+		return nil, nil
+	}
+
+	defaults := []grpc.CallOption{
+		grpc.FailFast(false),
+		grpc.MaxCallRecvMsgSize(1010),
+	}
+	tc := testpb.NewTestServiceClient(te.clientConn(grpc.WithDefaultCallOptions(defaults...)))
+
+	var headers metadata.MD
+	var trailers metadata.MD
+	var pr peer.Peer
+	tc.UnaryCall(context.Background(), &testpb.SimpleRequest{},
+		grpc.MaxCallRecvMsgSize(100),
+		grpc.MaxCallSendMsgSize(200),
+		grpc.PerRPCCredentials(testPerRPCCredentials{}),
+		grpc.Header(&headers),
+		grpc.Trailer(&trailers),
+		grpc.Peer(&pr))
+	expected := observedOptions{
+		failFast:    []bool{false},
+		maxRecvSize: []int{1010, 100},
+		maxSendSize: []int{200},
+		creds:       []credentials.PerRPCCredentials{testPerRPCCredentials{}},
+		headers:     []*metadata.MD{&headers},
+		trailers:    []*metadata.MD{&trailers},
+		peer:        []*peer.Peer{&pr},
+	}
+
+	if !reflect.DeepEqual(expected, observedOpts) {
+		t.Errorf("unary call did not observe expected options: expected %#v, got %#v", expected, observedOpts)
+	}
+
+	observedOpts = observedOptions{} // reset
+
+	var codec errCodec
+	tc.StreamingInputCall(context.Background(),
+		grpc.FailFast(true),
+		grpc.MaxCallSendMsgSize(2020),
+		grpc.UseCompressor("comp-type"),
+		grpc.CallContentSubtype("json"),
+		grpc.CallCustomCodec(&codec))
+	expected = observedOptions{
+		failFast:    []bool{false, true},
+		maxRecvSize: []int{1010},
+		maxSendSize: []int{2020},
+		compressor:  []string{"comp-type"},
+		subtype:     []string{"json"},
+		codec:       []grpc.Codec{&codec},
+	}
+
+	if !reflect.DeepEqual(expected, observedOpts) {
+		t.Errorf("streaming call did not observe expected options: expected %#v, got %#v", expected, observedOpts)
+	}
+}
+
 func TestCompressorRegister(t *testing.T) {
 	defer leakcheck.Check(t)
 	for _, e := range listTestEnv() {