diff --git a/README.md b/README.md index 6b37a60..6d18b71 100644 --- a/README.md +++ b/README.md @@ -34,7 +34,7 @@ Interceptor configuration functions \(AddUnaryServerInterceptor, SetFilterFunc, - [func DefaultStreamInterceptors\(\) \[\]grpc.StreamServerInterceptor](<#DefaultStreamInterceptors>) - [func DoHTTPtoGRPC\(ctx context.Context, svr interface\{\}, handler func\(ctx context.Context, req interface\{\}\) \(interface\{\}, error\), in interface\{\}\) \(interface\{\}, error\)](<#DoHTTPtoGRPC>) - [func FilterMethodsFunc\(ctx context.Context, fullMethodName string\) bool](<#FilterMethodsFunc>) -- [func GRPCClientInterceptor\(options ...interface\{\}\) grpc.UnaryClientInterceptor](<#GRPCClientInterceptor>) +- [func GRPCClientInterceptor\(\_ ...interface\{\}\) grpc.UnaryClientInterceptor](<#GRPCClientInterceptor>) - [func HystrixClientInterceptor\(defaultOpts ...grpc.CallOption\) grpc.UnaryClientInterceptor](<#HystrixClientInterceptor>) - [func NRHttpTracer\(pattern string, h http.HandlerFunc\) \(string, http.HandlerFunc\)](<#NRHttpTracer>) - [func NewRelicClientInterceptor\(\) grpc.UnaryClientInterceptor](<#NewRelicClientInterceptor>) @@ -47,6 +47,8 @@ Interceptor configuration functions \(AddUnaryServerInterceptor, SetFilterFunc, - [func ServerErrorStreamInterceptor\(\) grpc.StreamServerInterceptor](<#ServerErrorStreamInterceptor>) - [func SetClientMetricsOptions\(opts ...grpcprom.ClientMetricsOption\)](<#SetClientMetricsOptions>) - [func SetFilterFunc\(ctx context.Context, ff FilterFunc\)](<#SetFilterFunc>) +- [func SetFilterMethods\(ctx context.Context, methods \[\]string\)](<#SetFilterMethods>) +- [func SetResponseTimeLogErrorOnly\(errorOnly bool\)](<#SetResponseTimeLogErrorOnly>) - [func SetResponseTimeLogLevel\(ctx context.Context, level loggers.Level\)](<#SetResponseTimeLogLevel>) - [func SetServerMetricsOptions\(opts ...grpcprom.ServerMetricsOption\)](<#SetServerMetricsOptions>) - [func TraceIdInterceptor\(\) grpc.UnaryServerInterceptor](<#TraceIdInterceptor>) @@ -69,13 +71,16 @@ const SupportPackageIsVersion1 = true ```go var ( - //FilterMethods is the list of methods that are filtered by default + // Deprecated: FilterMethods is the list of methods that are filtered by default. + // Use SetFilterMethods instead. Only some direct mutations (replacing the slice + // or changing the first element) are detected by internal change detection; + // other in-place changes may not invalidate caches correctly. FilterMethods = []string{"healthcheck", "readycheck", "serverreflectioninfo"} ) ``` -## func [AddStreamClientInterceptor]() +## func [AddStreamClientInterceptor]() ```go func AddStreamClientInterceptor(ctx context.Context, i ...grpc.StreamClientInterceptor) @@ -84,7 +89,7 @@ func AddStreamClientInterceptor(ctx context.Context, i ...grpc.StreamClientInter AddStreamClientInterceptor adds a client stream interceptor to default client stream interceptors. Must be called during initialization, before any RPCs are made. Not safe for concurrent use. -## func [AddStreamServerInterceptor]() +## func [AddStreamServerInterceptor]() ```go func AddStreamServerInterceptor(ctx context.Context, i ...grpc.StreamServerInterceptor) @@ -93,7 +98,7 @@ func AddStreamServerInterceptor(ctx context.Context, i ...grpc.StreamServerInter AddStreamServerInterceptor adds a server interceptor to default server interceptors. Must be called during initialization, before the server starts. Not safe for concurrent use. -## func [AddUnaryClientInterceptor]() +## func [AddUnaryClientInterceptor]() ```go func AddUnaryClientInterceptor(ctx context.Context, i ...grpc.UnaryClientInterceptor) @@ -102,7 +107,7 @@ func AddUnaryClientInterceptor(ctx context.Context, i ...grpc.UnaryClientInterce AddUnaryClientInterceptor adds a client interceptor to default client interceptors. Must be called during initialization, before any RPCs are made. Not safe for concurrent use. -## func [AddUnaryServerInterceptor]() +## func [AddUnaryServerInterceptor]() ```go func AddUnaryServerInterceptor(ctx context.Context, i ...grpc.UnaryServerInterceptor) @@ -111,7 +116,7 @@ func AddUnaryServerInterceptor(ctx context.Context, i ...grpc.UnaryServerInterce AddUnaryServerInterceptor adds a server interceptor to default server interceptors. Must be called during initialization, before the server starts. Not safe for concurrent use. -## func [DebugLoggingInterceptor]() +## func [DebugLoggingInterceptor]() ```go func DebugLoggingInterceptor() grpc.UnaryServerInterceptor @@ -120,7 +125,7 @@ func DebugLoggingInterceptor() grpc.UnaryServerInterceptor DebugLoggingInterceptor is the interceptor that logs all request/response from a handler -## func [DefaultClientInterceptor]() +## func [DefaultClientInterceptor]() ```go func DefaultClientInterceptor(defaultOpts ...interface{}) grpc.UnaryClientInterceptor @@ -129,7 +134,7 @@ func DefaultClientInterceptor(defaultOpts ...interface{}) grpc.UnaryClientInterc DefaultClientInterceptor are the set of default interceptors that should be applied to all client calls -## func [DefaultClientInterceptors]() +## func [DefaultClientInterceptors]() ```go func DefaultClientInterceptors(defaultOpts ...interface{}) []grpc.UnaryClientInterceptor @@ -138,7 +143,7 @@ func DefaultClientInterceptors(defaultOpts ...interface{}) []grpc.UnaryClientInt DefaultClientInterceptors are the set of default interceptors that should be applied to all client calls -## func [DefaultClientStreamInterceptor]() +## func [DefaultClientStreamInterceptor]() ```go func DefaultClientStreamInterceptor(defaultOpts ...interface{}) grpc.StreamClientInterceptor @@ -147,7 +152,7 @@ func DefaultClientStreamInterceptor(defaultOpts ...interface{}) grpc.StreamClien DefaultClientStreamInterceptor are the set of default interceptors that should be applied to all stream client calls -## func [DefaultClientStreamInterceptors]() +## func [DefaultClientStreamInterceptors]() ```go func DefaultClientStreamInterceptors(defaultOpts ...interface{}) []grpc.StreamClientInterceptor @@ -156,7 +161,7 @@ func DefaultClientStreamInterceptors(defaultOpts ...interface{}) []grpc.StreamCl DefaultClientStreamInterceptors are the set of default interceptors that should be applied to all stream client calls -## func [DefaultInterceptors]() +## func [DefaultInterceptors]() ```go func DefaultInterceptors() []grpc.UnaryServerInterceptor @@ -165,7 +170,7 @@ func DefaultInterceptors() []grpc.UnaryServerInterceptor DefaultInterceptors are the set of default interceptors that are applied to all coldbrew methods -## func [DefaultStreamInterceptors]() +## func [DefaultStreamInterceptors]() ```go func DefaultStreamInterceptors() []grpc.StreamServerInterceptor @@ -174,16 +179,33 @@ func DefaultStreamInterceptors() []grpc.StreamServerInterceptor DefaultStreamInterceptors are the set of default interceptors that should be applied to all coldbrew streams -## func [DoHTTPtoGRPC]() +## func [DoHTTPtoGRPC]() ```go func DoHTTPtoGRPC(ctx context.Context, svr interface{}, handler func(ctx context.Context, req interface{}) (interface{}, error), in interface{}) (interface{}, error) ``` +DoHTTPtoGRPC allows calling the interceptors when you use the Register\HandlerServer in grpc\-gateway. This enables in\-process HTTP\-to\-gRPC calls with the full interceptor chain \(logging, tracing, metrics, panic recovery\) without a network hop — the fastest option for gateway performance. The interceptor chain is cached on first invocation. All interceptor configuration \(AddUnaryServerInterceptor, SetFilterFunc, etc.\) must be finalized before the first call. See example below for reference. +``` +func (s *svc) Echo(ctx context.Context, req *proto.EchoRequest) (*proto.EchoResponse, error) { + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return s.echo(ctx, req.(*proto.EchoRequest)) + } + r, err := DoHTTPtoGRPC(ctx, s, handler, req) + if err != nil { + return nil, err + } + return r.(*proto.EchoResponse), nil +} + +func (s *svc) echo(ctx context.Context, req *proto.EchoRequest) (*proto.EchoResponse, error) { + .... implementation .... +} +``` -## func [FilterMethodsFunc]() +## func [FilterMethodsFunc]() ```go func FilterMethodsFunc(ctx context.Context, fullMethodName string) bool @@ -192,16 +214,16 @@ func FilterMethodsFunc(ctx context.Context, fullMethodName string) bool FilterMethodsFunc is the default implementation of Filter function -## func [GRPCClientInterceptor]() +## func [GRPCClientInterceptor]() ```go -func GRPCClientInterceptor(options ...interface{}) grpc.UnaryClientInterceptor +func GRPCClientInterceptor(_ ...interface{}) grpc.UnaryClientInterceptor ``` Deprecated: GRPCClientInterceptor is no longer needed. gRPC tracing is now handled by otelgrpc.NewClientHandler stats handler configured at the client level. This function is retained for backwards compatibility but returns a no\-op interceptor. -## func [HystrixClientInterceptor]() +## func [HystrixClientInterceptor]() ```go func HystrixClientInterceptor(defaultOpts ...grpc.CallOption) grpc.UnaryClientInterceptor @@ -214,7 +236,7 @@ Note: This interceptor wraps github.com/afex/hystrix\-go which has been unmainta The interceptor applies provided default and per\-call client options to configure Hystrix behavior \(for example the command name, disabled flag, excluded errors, and excluded gRPC status codes\). If Hystrix is disabled via options, the RPC is invoked directly. If the underlying RPC returns an error that matches any configured excluded error or whose gRPC status code matches any configured excluded code, Hystrix fallback is skipped and the RPC error is returned. Panics raised during the RPC invocation are captured and reported to the notifier before being converted into an error. If the RPC itself returns an error, that error is returned; otherwise any error produced by Hystrix is returned. -## func [NRHttpTracer]() +## func [NRHttpTracer]() ```go func NRHttpTracer(pattern string, h http.HandlerFunc) (string, http.HandlerFunc) @@ -223,25 +245,25 @@ func NRHttpTracer(pattern string, h http.HandlerFunc) (string, http.HandlerFunc) NRHttpTracer adds newrelic tracing to this http function -## func [NewRelicClientInterceptor]() +## func [NewRelicClientInterceptor]() ```go func NewRelicClientInterceptor() grpc.UnaryClientInterceptor ``` -NewRelicClientInterceptor intercepts all client actions and reports them to newrelic +NewRelicClientInterceptor intercepts all client actions and reports them to newrelic. When NewRelic app is nil \(no license key configured\), returns a pass\-through interceptor to avoid overhead. -## func [NewRelicInterceptor]() +## func [NewRelicInterceptor]() ```go func NewRelicInterceptor() grpc.UnaryServerInterceptor ``` -NewRelicInterceptor intercepts all server actions and reports them to newrelic +NewRelicInterceptor intercepts all server actions and reports them to newrelic. When NewRelic app is nil \(no license key configured\), returns a pass\-through interceptor to avoid overhead. -## func [OptionsInterceptor]() +## func [OptionsInterceptor]() ```go func OptionsInterceptor() grpc.UnaryServerInterceptor @@ -250,7 +272,7 @@ func OptionsInterceptor() grpc.UnaryServerInterceptor -## func [PanicRecoveryInterceptor]() +## func [PanicRecoveryInterceptor]() ```go func PanicRecoveryInterceptor() grpc.UnaryServerInterceptor @@ -259,7 +281,7 @@ func PanicRecoveryInterceptor() grpc.UnaryServerInterceptor -## func [ResponseTimeLoggingInterceptor]() +## func [ResponseTimeLoggingInterceptor]() ```go func ResponseTimeLoggingInterceptor(ff FilterFunc) grpc.UnaryServerInterceptor @@ -268,7 +290,7 @@ func ResponseTimeLoggingInterceptor(ff FilterFunc) grpc.UnaryServerInterceptor ResponseTimeLoggingInterceptor logs response time for each request on server -## func [ResponseTimeLoggingStreamInterceptor]() +## func [ResponseTimeLoggingStreamInterceptor]() ```go func ResponseTimeLoggingStreamInterceptor() grpc.StreamServerInterceptor @@ -277,7 +299,7 @@ func ResponseTimeLoggingStreamInterceptor() grpc.StreamServerInterceptor ResponseTimeLoggingStreamInterceptor logs response time for stream RPCs. -## func [ServerErrorInterceptor]() +## func [ServerErrorInterceptor]() ```go func ServerErrorInterceptor() grpc.UnaryServerInterceptor @@ -286,7 +308,7 @@ func ServerErrorInterceptor() grpc.UnaryServerInterceptor ServerErrorInterceptor intercepts all server actions and reports them to error notifier -## func [ServerErrorStreamInterceptor]() +## func [ServerErrorStreamInterceptor]() ```go func ServerErrorStreamInterceptor() grpc.StreamServerInterceptor @@ -295,7 +317,7 @@ func ServerErrorStreamInterceptor() grpc.StreamServerInterceptor ServerErrorStreamInterceptor intercepts server errors for stream RPCs and reports them to the error notifier. -## func [SetClientMetricsOptions]() +## func [SetClientMetricsOptions]() ```go func SetClientMetricsOptions(opts ...grpcprom.ClientMetricsOption) @@ -304,7 +326,7 @@ func SetClientMetricsOptions(opts ...grpcprom.ClientMetricsOption) SetClientMetricsOptions appends gRPC client metrics options. Must be called during initialization, before the server starts. Not safe for concurrent use. -## func [SetFilterFunc]() +## func [SetFilterFunc]() ```go func SetFilterFunc(ctx context.Context, ff FilterFunc) @@ -312,8 +334,26 @@ func SetFilterFunc(ctx context.Context, ff FilterFunc) SetFilterFunc sets the default filter function to be used by interceptors. Must be called during initialization, before the server starts. Not safe for concurrent use. + +## func [SetFilterMethods]() + +```go +func SetFilterMethods(ctx context.Context, methods []string) +``` + +SetFilterMethods sets the list of method substrings to exclude from tracing/logging. It rebuilds the internal cache. Must be called during initialization, before the server starts. Not safe for concurrent use. + + +## func [SetResponseTimeLogErrorOnly]() + +```go +func SetResponseTimeLogErrorOnly(errorOnly bool) +``` + +SetResponseTimeLogErrorOnly when set to true, only logs response time when the request returns an error. Successful requests are not logged. Must be called during initialization, before the server starts. Not safe for concurrent use. + -## func [SetResponseTimeLogLevel]() +## func [SetResponseTimeLogLevel]() ```go func SetResponseTimeLogLevel(ctx context.Context, level loggers.Level) @@ -322,7 +362,7 @@ func SetResponseTimeLogLevel(ctx context.Context, level loggers.Level) SetResponseTimeLogLevel sets the log level for response time logging. Default is InfoLevel. Must be called during initialization, before the server starts. Not safe for concurrent use. -## func [SetServerMetricsOptions]() +## func [SetServerMetricsOptions]() ```go func SetServerMetricsOptions(opts ...grpcprom.ServerMetricsOption) @@ -331,7 +371,7 @@ func SetServerMetricsOptions(opts ...grpcprom.ServerMetricsOption) SetServerMetricsOptions appends gRPC server metrics options \(histogram, labels, namespace, etc.\). Must be called during initialization, before the server starts. Not safe for concurrent use. -## func [TraceIdInterceptor]() +## func [TraceIdInterceptor]() ```go func TraceIdInterceptor() grpc.UnaryServerInterceptor @@ -340,7 +380,7 @@ func TraceIdInterceptor() grpc.UnaryServerInterceptor TraceIdInterceptor allows injecting trace id from request objects -## func [UseColdBrewClientInterceptors]() +## func [UseColdBrewClientInterceptors]() ```go func UseColdBrewClientInterceptors(ctx context.Context, flag bool) @@ -349,7 +389,7 @@ func UseColdBrewClientInterceptors(ctx context.Context, flag bool) UseColdBrewClientInterceptors allows enabling/disabling coldbrew client interceptors. When set to false, the coldbrew client interceptors will not be used. Must be called during initialization, before any RPCs are made. Not safe for concurrent use. -## func [UseColdBrewServerInterceptors]() +## func [UseColdBrewServerInterceptors]() ```go func UseColdBrewServerInterceptors(ctx context.Context, flag bool) @@ -358,7 +398,7 @@ func UseColdBrewServerInterceptors(ctx context.Context, flag bool) UseColdBrewServerInterceptors allows enabling/disabling coldbrew server interceptors. When set to false, the coldbrew server interceptors will not be used. Must be called during initialization, before the server starts. Not safe for concurrent use. -## type [FilterFunc]() +## type [FilterFunc]() If it returns false, the given request will not be traced. diff --git a/interceptors.go b/interceptors.go index 183e521..484c23e 100644 --- a/interceptors.go +++ b/interceptors.go @@ -24,11 +24,11 @@ import ( "github.com/go-coldbrew/options" nrutil "github.com/go-coldbrew/tracing/newrelic" grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus" - "github.com/prometheus/client_golang/prometheus" grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/retry" "github.com/grpc-ecosystem/grpc-gateway/v2/runtime" "github.com/newrelic/go-agent/v3/integrations/nrgrpc" newrelic "github.com/newrelic/go-agent/v3/newrelic" + "github.com/prometheus/client_golang/prometheus" "google.golang.org/grpc" "google.golang.org/grpc/status" ) @@ -47,14 +47,14 @@ var ( // Use SetFilterMethods instead. Only some direct mutations (replacing the slice // or changing the first element) are detected by internal change detection; // other in-place changes may not invalidate caches correctly. - FilterMethods = []string{"healthcheck", "readycheck", "serverreflectioninfo"} - defaultFilterFunc = FilterMethodsFunc - unaryServerInterceptors = []grpc.UnaryServerInterceptor{} - streamServerInterceptors = []grpc.StreamServerInterceptor{} - useCBServerInterceptors = true - unaryClientInterceptors = []grpc.UnaryClientInterceptor{} - streamClientInterceptors = []grpc.StreamClientInterceptor{} - useCBClientInterceptors = true + FilterMethods = []string{"healthcheck", "readycheck", "serverreflectioninfo"} + defaultFilterFunc = FilterMethodsFunc + unaryServerInterceptors = []grpc.UnaryServerInterceptor{} + streamServerInterceptors = []grpc.StreamServerInterceptor{} + useCBServerInterceptors = true + unaryClientInterceptors = []grpc.UnaryClientInterceptor{} + streamClientInterceptors = []grpc.StreamClientInterceptor{} + useCBClientInterceptors = true responseTimeLogLevel loggers.Level = loggers.InfoLevel responseTimeLogErrorOnly bool srvMetricsOpts []grpcprom.ServerMetricsOption @@ -307,18 +307,32 @@ func chainStreamClient(interceptors []grpc.StreamClientInterceptor) grpc.StreamC } } +var ( + httpToGRPCOnce sync.Once + httpToGRPCInterceptor grpc.UnaryServerInterceptor +) + +func getHTTPtoGRPCInterceptor() grpc.UnaryServerInterceptor { + httpToGRPCOnce.Do(func() { + httpToGRPCInterceptor = chainUnaryServer(DefaultInterceptors()) + }) + return httpToGRPCInterceptor +} + // DoHTTPtoGRPC allows calling the interceptors when you use the RegisterHandlerServer in grpc-gateway. +// This enables in-process HTTP-to-gRPC calls with the full interceptor chain (logging, tracing, metrics, +// panic recovery) without a network hop — the fastest option for gateway performance. // The interceptor chain is cached on first invocation. All interceptor configuration // (AddUnaryServerInterceptor, SetFilterFunc, etc.) must be finalized before the first call. -// See example below for reference +// See example below for reference. // // func (s *svc) Echo(ctx context.Context, req *proto.EchoRequest) (*proto.EchoResponse, error) { // handler := func(ctx context.Context, req interface{}) (interface{}, error) { // return s.echo(ctx, req.(*proto.EchoRequest)) // } -// r, e := doHTTPtoGRPC(ctx, s, handler, req) -// if e != nil { -// return nil, e.(error) +// r, err := DoHTTPtoGRPC(ctx, s, handler, req) +// if err != nil { +// return nil, err // } // return r.(*proto.EchoResponse), nil // } @@ -326,18 +340,6 @@ func chainStreamClient(interceptors []grpc.StreamClientInterceptor) grpc.StreamC // func (s *svc) echo(ctx context.Context, req *proto.EchoRequest) (*proto.EchoResponse, error) { // .... implementation .... // } -var ( - httpToGRPCOnce sync.Once - httpToGRPCInterceptor grpc.UnaryServerInterceptor -) - -func getHTTPtoGRPCInterceptor() grpc.UnaryServerInterceptor { - httpToGRPCOnce.Do(func() { - httpToGRPCInterceptor = chainUnaryServer(DefaultInterceptors()) - }) - return httpToGRPCInterceptor -} - func DoHTTPtoGRPC(ctx context.Context, svr interface{}, handler func(ctx context.Context, req interface{}) (interface{}, error), in interface{}) (interface{}, error) { method, ok := runtime.RPCMethod(ctx) if ok { @@ -473,7 +475,7 @@ func ResponseTimeLoggingInterceptor(ff FilterFunc) grpc.UnaryServerInterceptor { func OptionsInterceptor() grpc.UnaryServerInterceptor { return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { ctx = options.AddToOptions(ctx, "", "") - //loggers.AddToLogContext(ctx, "transport", "gRPC") + // loggers.AddToLogContext(ctx, "transport", "gRPC") return handler(ctx, req) } } @@ -664,7 +666,6 @@ func ServerErrorStreamInterceptor() grpc.StreamServerInterceptor { }) } return err - } } @@ -703,4 +704,4 @@ func TraceIdInterceptor() grpc.UnaryServerInterceptor { } return handler(ctx, req) } -} \ No newline at end of file +} diff --git a/interceptors_test.go b/interceptors_test.go index 2782915..52a47af 100644 --- a/interceptors_test.go +++ b/interceptors_test.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "sync" + "sync/atomic" "testing" "github.com/go-coldbrew/log/loggers" @@ -41,6 +42,8 @@ func resetGlobals() { useCBClientInterceptors = true responseTimeLogErrorOnly = false responseTimeLogLevel = loggers.InfoLevel + httpToGRPCOnce = sync.Once{} + httpToGRPCInterceptor = nil } func TestFilterMethodsFunc(t *testing.T) { @@ -814,3 +817,221 @@ func TestResponseTimeLogErrorOnly_LogsErrors(t *testing.T) { t.Fatalf("expected handler error, got %v", err) } } + +func TestDoHTTPtoGRPC_HandlerError(t *testing.T) { + defer resetGlobals() + UseColdBrewServerInterceptors(context.Background(), false) + + testErr := errors.New("handler failed") + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return nil, testErr + } + + // Without RPCMethod — error should propagate directly. + resp, err := DoHTTPtoGRPC(context.Background(), nil, handler, "input") + if err != testErr { + t.Fatalf("expected testErr, got %v", err) + } + if resp != nil { + t.Fatalf("expected nil resp, got %v", resp) + } + + // With RPCMethod — error should propagate through interceptor chain. + req, _ := http.NewRequest("GET", "http://localhost/test", nil) + mux := runtime.NewServeMux() + ctxWithRPC, err := runtime.AnnotateIncomingContext(context.Background(), mux, req, "/test.Service/Echo") + if err != nil { + t.Fatalf("AnnotateIncomingContext: %v", err) + } + resp, err = DoHTTPtoGRPC(ctxWithRPC, nil, handler, "input") + if err != testErr { + t.Fatalf("expected testErr through chain, got %v", err) + } + if resp != nil { + t.Fatalf("expected nil resp through chain, got %v", resp) + } +} + +func TestDoHTTPtoGRPC_MethodPassedToInfo(t *testing.T) { + defer resetGlobals() + UseColdBrewServerInterceptors(context.Background(), false) + + var capturedMethod string + AddUnaryServerInterceptor(context.Background(), func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { + capturedMethod = info.FullMethod + return handler(ctx, req) + }) + + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return "ok", nil + } + + req, _ := http.NewRequest("GET", "http://localhost/test", nil) + mux := runtime.NewServeMux() + ctxWithRPC, err := runtime.AnnotateIncomingContext(context.Background(), mux, req, "/test.Service/Echo") + if err != nil { + t.Fatalf("AnnotateIncomingContext: %v", err) + } + + _, err = DoHTTPtoGRPC(ctxWithRPC, nil, handler, "input") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if capturedMethod != "/test.Service/Echo" { + t.Errorf("expected FullMethod '/test.Service/Echo', got %q", capturedMethod) + } +} + +func TestDoHTTPtoGRPC_InputPassedThrough(t *testing.T) { + defer resetGlobals() + UseColdBrewServerInterceptors(context.Background(), false) + + var capturedReq interface{} + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + capturedReq = req + return "ok", nil + } + + // Without RPCMethod — input goes directly to handler. + if _, err := DoHTTPtoGRPC(context.Background(), nil, handler, "direct-input"); err != nil { + t.Fatalf("DoHTTPtoGRPC without RPCMethod: %v", err) + } + if capturedReq != "direct-input" { + t.Errorf("expected 'direct-input', got %v", capturedReq) + } + + // With RPCMethod — input goes through interceptor chain. + capturedReq = nil + req, _ := http.NewRequest("GET", "http://localhost/test", nil) + mux := runtime.NewServeMux() + ctxWithRPC, err := runtime.AnnotateIncomingContext(context.Background(), mux, req, "/test.Service/Echo") + if err != nil { + t.Fatalf("AnnotateIncomingContext: %v", err) + } + _, err = DoHTTPtoGRPC(ctxWithRPC, nil, handler, "chain-input") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if capturedReq != "chain-input" { + t.Errorf("expected 'chain-input', got %v", capturedReq) + } +} + +func TestDoHTTPtoGRPC_ServerPassedToInfo(t *testing.T) { + defer resetGlobals() + UseColdBrewServerInterceptors(context.Background(), false) + + type fakeServer struct{ Name string } + svr := &fakeServer{Name: "test-server"} + + var capturedServer interface{} + AddUnaryServerInterceptor(context.Background(), func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { + capturedServer = info.Server + return handler(ctx, req) + }) + + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return "ok", nil + } + + req, _ := http.NewRequest("GET", "http://localhost/test", nil) + mux := runtime.NewServeMux() + ctxWithRPC, err := runtime.AnnotateIncomingContext(context.Background(), mux, req, "/test.Service/Echo") + if err != nil { + t.Fatalf("AnnotateIncomingContext: %v", err) + } + + _, err = DoHTTPtoGRPC(ctxWithRPC, svr, handler, "input") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if capturedServer != svr { + t.Errorf("expected server %v, got %v", svr, capturedServer) + } +} + +func TestDoHTTPtoGRPC_Concurrent(t *testing.T) { + defer resetGlobals() + UseColdBrewServerInterceptors(context.Background(), false) + + var callCount int64 + AddUnaryServerInterceptor(context.Background(), func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { + atomic.AddInt64(&callCount, 1) + return handler(ctx, req) + }) + + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return req, nil + } + + req, _ := http.NewRequest("GET", "http://localhost/test", nil) + mux := runtime.NewServeMux() + ctxWithRPC, err := runtime.AnnotateIncomingContext(context.Background(), mux, req, "/test.Service/Echo") + if err != nil { + t.Fatalf("AnnotateIncomingContext: %v", err) + } + + const goroutines = 50 + var wg sync.WaitGroup + for i := 0; i < goroutines; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + resp, err := DoHTTPtoGRPC(ctxWithRPC, nil, handler, fmt.Sprintf("req-%d", n)) + if err != nil { + t.Errorf("goroutine %d: unexpected error: %v", n, err) + } + expected := fmt.Sprintf("req-%d", n) + if resp != expected { + t.Errorf("goroutine %d: expected %q, got %v", n, expected, resp) + } + }(i) + } + wg.Wait() + + if got := atomic.LoadInt64(&callCount); got != goroutines { + t.Errorf("expected %d interceptor calls, got %d", goroutines, got) + } +} + +func TestDoHTTPtoGRPC_InterceptorCaching(t *testing.T) { + defer resetGlobals() + UseColdBrewServerInterceptors(context.Background(), false) + + AddUnaryServerInterceptor(context.Background(), func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { + return handler(ctx, req) + }) + + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return "ok", nil + } + + req, _ := http.NewRequest("GET", "http://localhost/test", nil) + mux := runtime.NewServeMux() + ctxWithRPC, err := runtime.AnnotateIncomingContext(context.Background(), mux, req, "/test.Service/Echo") + if err != nil { + t.Fatalf("AnnotateIncomingContext: %v", err) + } + + // Call twice — interceptor should be built only once. + if _, err := DoHTTPtoGRPC(ctxWithRPC, nil, handler, "first"); err != nil { + t.Fatalf("first DoHTTPtoGRPC: %v", err) + } + if _, err := DoHTTPtoGRPC(ctxWithRPC, nil, handler, "second"); err != nil { + t.Fatalf("second DoHTTPtoGRPC: %v", err) + } + + // Adding a new interceptor after first call should NOT affect the cached chain. + interceptor2Called := false + AddUnaryServerInterceptor(context.Background(), func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { + interceptor2Called = true + return handler(ctx, req) + }) + + if _, err := DoHTTPtoGRPC(ctxWithRPC, nil, handler, "third"); err != nil { + t.Fatalf("third DoHTTPtoGRPC: %v", err) + } + if interceptor2Called { + t.Error("interceptor added after first DoHTTPtoGRPC call should not be in the cached chain") + } +}