diff --git a/CHANGELOG.md b/CHANGELOG.md index fb1228f..c09ecf4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,10 @@ below says plainly whether an upgrade can break a caller. call holding a `*digrpc.Call`, `digrpc.Module` provides it, and `Interceptor.Options()` gives the server options. It is versioned on its own, as `digrpc/vX.Y.Z`. +- `digrpc.Register[H](srv, desc)`, in `digrpc/v0.2.0`, serves a generated + service with an `H` resolved from each call's scope: the implementation is + `Scoped` when it takes the call, is built after the server's interceptors + have run, and a constructor's status error fails the call with that status. ## [0.17.1] - 2026-09-19 diff --git a/CLAUDE.md b/CLAUDE.md index f3133a9..1487cff 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -566,6 +566,16 @@ reverse: caught by the fuzzer in 0.06s, missed by 400 seeded sequences). health service over `bufconn`, so nothing is generated from protobuf. The coverage badge does not include it. `examples/` reaches it through a second `replace`, and `examples/grpc` is the lifecycle example the README embeds. + `Register[H]` copies the generated `ServiceDesc` with its handlers replaced: + a unary wrapper hands the generated handler a fake interceptor, so the + generated code still decodes and builds the info, then runs the server's + real chain with a handler that resolves `H` and calls the method through + `reflect`; a stream wrapper resolves `H` and passes it to the generated + handler unchanged, since stream interceptors run outside it. `H` is + therefore built after every interceptor, the info's `Server` is nil, and a + constructor's status is taken from inside the build error with + `errors.AsType` so the client never sees the registration site. That rests + on `grpc.MethodHandler` being the public contract it is documented as. - **`dislog/` is the slog bridge for `Observe`** and imports nothing but `log/slog` and the library. `dislog.New` returns the `func(di.Event)` that `Observe` takes, not a `slog.Handler`. Failed steps log at Error with the site diff --git a/README.md b/README.md index 188940a..846f624 100644 --- a/README.md +++ b/README.md @@ -455,8 +455,11 @@ srv := &http.Server{Handler: dihttp.NewMiddleware(app)(mux)} For gRPC, [`digrpc`](digrpc) is the same adapter as a separate module, so the library itself does not depend on grpc: `go get github.com/floatdrop/di/digrpc`. Its `Interceptor` opens a scope per call holding a `*digrpc.Call`, the method -and the incoming context; `Module` provides it; a service implementation -reaches the scope with `di.FromContext`. `GracefulStop` is the drain hook. +and the incoming context, and `Module` provides it. `digrpc.Register[*Users](srv, &pb.Users_ServiceDesc)` +serves a generated service with an implementation resolved from that scope, +so the implementation is a service like a handler type under `dihttp.Handle`: +`Scoped()` when it takes the call, built after the interceptors have run, and +a constructor's status error is the call's. `GracefulStop` is the drain hook.
examples/grpc/main.go, a gRPC server with a service built per call from the call's metadata @@ -466,8 +469,9 @@ reaches the scope with `di.FromContext`. `GracefulStop` is the drain hook. // Graceful shutdown of a gRPC server, with a service built per call. // // The digrpc interceptor opens a scope for every call and registers the -// *digrpc.Call in it, so a Caller declared Scoped is built per call from the -// call's metadata. Run starts the scope, waits for SIGINT/SIGTERM or a +// *digrpc.Call in it, and digrpc.Register serves the health service with an +// implementation resolved from that scope, so a Scoped Health is built per +// call with a Caller read from the call's metadata. Run starts the scope, waits for SIGINT/SIGTERM or a // Shutdown call, then stops everything in reverse order with a bounded // context. The server's OnDrain calls GracefulStop, which stops accepting // calls and waits for in-flight ones. Draining runs before anything is torn @@ -501,28 +505,27 @@ func NewCaller(c *digrpc.Call) *Caller { return &Caller{Name: cmp.Or(strings.Join(md.Get("x-caller"), ","), "anonymous")} } -// Health is a service implementation. grpc registers one value for the whole -// server, so it is a singleton that reaches per-call services through the -// scope on its context. +// Health is the service implementation, built per call because it takes the +// Caller. Methods it does not define fall through to the embedded type. type Health struct { grpc_health_v1.UnimplementedHealthServer - db *DB + db *DB + caller *Caller } -func NewHealth(db *DB) *Health { return &Health{db: db} } +func NewHealth(db *DB, c *Caller) *Health { return &Health{db: db, caller: c} } -func (h *Health) Check(ctx context.Context, _ *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) { - scope, _ := di.FromContext(ctx) - caller := scope.Get[*Caller]() +func (h *Health) Check(context.Context, *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) { time.Sleep(2 * time.Second) // simulate slow work that must not be cut short - log.Println("checked by", caller.Name, "against", h.db.dsn) + log.Println("checked by", h.caller.Name, "against", h.db.dsn) return &grpc_health_v1.HealthCheckResponse{Status: grpc_health_v1.HealthCheckResponse_SERVING}, nil } -// NewServer is a plain constructor: the interceptor arrives as a dependency. -func NewServer(ic digrpc.Interceptor, h *Health) *grpc.Server { +// NewServer is a plain constructor: the interceptor arrives as a dependency, +// and Register resolves *Health from each call's scope. +func NewServer(ic digrpc.Interceptor) *grpc.Server { srv := grpc.NewServer(ic.Options()...) - grpc_health_v1.RegisterHealthServer(srv, h) + digrpc.Register[*Health](srv, &grpc_health_v1.Health_ServiceDesc) return srv } @@ -533,7 +536,7 @@ func main() { app.Wire[*DB](func() *DB { return &DB{dsn: "postgres://localhost/app"} }). OnStop(func(ctx context.Context, db *DB) error { log.Println("db closed"); return nil }) app.Wire[*Caller](NewCaller).Scoped() - app.Wire[*Health](NewHealth) + app.Wire[*Health](NewHealth).Scoped() app.Wire[*grpc.Server](NewServer). Eager(). diff --git a/digrpc/digrpc.go b/digrpc/digrpc.go index 7535bb8..847de02 100644 --- a/digrpc/digrpc.go +++ b/digrpc/digrpc.go @@ -2,7 +2,9 @@ // // An [Interceptor] gives every call its own child scope holding a [*Call], // so services that depend on the call are declared once in the application -// scope as Scoped and built per call. A handler reaches the scope through +// scope as Scoped and built per call. [Register] serves a generated service +// with an implementation resolved from that scope, so the implementation is +// such a service too; one registered the plain way reaches the scope through // [di.FromContext] on the context it is given. [Module] registers the // interceptor as a service; [New] makes one directly. // diff --git a/digrpc/example_test.go b/digrpc/example_test.go index 47664a7..6c2df6f 100644 --- a/digrpc/example_test.go +++ b/digrpc/example_test.go @@ -24,9 +24,9 @@ func NewCaller(c *digrpc.Call) *Caller { return &Caller{Name: strings.Join(md.Get("x-caller"), ",")} } -// Health is a service implementation: one value for the whole server, as -// grpc registers it, reaching per-call services through the scope on the -// context. +// Health is a service implementation registered the plain way: one value +// for the whole server, reaching per-call services through the scope on the +// context. See ExampleRegister for the implementation built per call. type Health struct { grpc_health_v1.UnimplementedHealthServer } @@ -75,3 +75,51 @@ func ExampleModule() { // check from ada // SERVING } + +// Greeter is the service implementation built per call: it takes the Caller, +// so it is Scoped, and Register resolves it from each call's scope. +type Greeter struct { + grpc_health_v1.UnimplementedHealthServer + caller *Caller +} + +func NewGreeter(c *Caller) *Greeter { return &Greeter{caller: c} } + +func (g *Greeter) Check(context.Context, *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) { + fmt.Println("check from", g.caller.Name) + return &grpc_health_v1.HealthCheckResponse{Status: grpc_health_v1.HealthCheckResponse_SERVING}, nil +} + +func ExampleRegister() { + app := di.New() + app.Use(digrpc.Module) + app.Wire[*Caller](NewCaller).Scoped() + app.Wire[*Greeter](NewGreeter).Scoped() + app.Wire[*grpc.Server](func(ic digrpc.Interceptor) *grpc.Server { + srv := grpc.NewServer(ic.Options()...) + digrpc.Register[*Greeter](srv, &grpc_health_v1.Health_ServiceDesc) + return srv + }) + fmt.Println(app.Validate(di.Provided[*digrpc.Call]()).Err()) + + lis := bufconn.Listen(1 << 20) + srv := app.Get[*grpc.Server]() + go func() { _ = srv.Serve(lis) }() + defer srv.Stop() + + conn, err := grpc.NewClient("passthrough:///bufnet", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }), + grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + panic(err) + } + defer func() { _ = conn.Close() }() + + ctx := metadata.AppendToOutgoingContext(context.Background(), "x-caller", "ada") + res, err := grpc_health_v1.NewHealthClient(conn).Check(ctx, &grpc_health_v1.HealthCheckRequest{}) + fmt.Println(res.GetStatus(), err) + // Output: + // + // check from ada + // SERVING +} diff --git a/digrpc/register.go b/digrpc/register.go new file mode 100644 index 0000000..52f5274 --- /dev/null +++ b/digrpc/register.go @@ -0,0 +1,120 @@ +package digrpc + +import ( + "context" + "errors" + "fmt" + "reflect" + + "github.com/floatdrop/di" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// Register serves the service desc describes with an H resolved from each +// call's scope, so the implementation is a service like any other: declared +// Scoped when it needs the call, and once for the application when it does +// not. The desc is the generated one, and H must implement its interface: +// +// digrpc.Register[*Users](srv, &pb.Users_ServiceDesc) +// +// H may also be the service interface itself, resolved from whatever serves +// it. H is resolved after the server's interceptors have run, so what they +// attach to the context is there for its constructors. A constructor that +// returns a status error fails the call with that status; any other failure, +// or a nil H, is codes.Internal, with a failed build reported to the scope's +// observers. Interceptors see a nil Server in the call's info, since the +// implementation does not exist yet when they run. Register panics when H +// does not implement the service. +func Register[H any](srv grpc.ServiceRegistrar, desc *grpc.ServiceDesc) { + t := reflect.TypeFor[H]() + iface := reflect.TypeOf(desc.HandlerType).Elem() + if !t.Implements(iface) { + panic(fmt.Sprintf("digrpc: %s does not implement %s", t, iface)) + } + wrapped := *desc + wrapped.Methods = make([]grpc.MethodDesc, len(desc.Methods)) + for i, m := range desc.Methods { + wrapped.Methods[i] = grpc.MethodDesc{MethodName: m.MethodName, Handler: unary[H](m.Handler, m.MethodName)} + } + wrapped.Streams = make([]grpc.StreamDesc, len(desc.Streams)) + for i, s := range desc.Streams { + s.Handler = streaming[H](s.Handler) + wrapped.Streams[i] = s + } + srv.RegisterService(&wrapped, placeholder[H]()) +} + +// placeholder is a value of H for RegisterService's implements check, or nil +// for an interface H, which RegisterService accepts. It is never called: +// every handler ignores the value it is given. +func placeholder[H any]() any { + if t := reflect.TypeFor[H](); t.Kind() == reflect.Pointer { + return reflect.New(t.Elem()).Interface() + } + var zero H + return zero +} + +// unary lets the generated handler decode the request and build the call's +// info, then runs the server's interceptor chain with a handler that resolves +// H and calls the method. The method is looked up on the value, so H may be +// an interface. The generated handler always calls the interceptor it is +// given, so the value it is given as the server is never used. +func unary[H any](orig grpc.MethodHandler, name string) grpc.MethodHandler { + return func(_ any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) { + return orig(nil, ctx, dec, func(ctx context.Context, req any, info *grpc.UnaryServerInfo, _ grpc.UnaryHandler) (any, error) { + handler := func(ctx context.Context, req any) (any, error) { + impl, err := resolve[H](ctx) + if err != nil { + return nil, err + } + out := reflect.ValueOf(impl).MethodByName(name).Call([]reflect.Value{reflect.ValueOf(ctx), reflect.ValueOf(req)}) + err, _ = out[1].Interface().(error) + return out[0].Interface(), err + } + if interceptor == nil { + return handler(ctx, req) + } + return interceptor(ctx, req, info, handler) + }) + } +} + +// streaming resolves H once the stream interceptors have run and hands it +// to the generated handler as the implementation. +func streaming[H any](orig grpc.StreamHandler) grpc.StreamHandler { + return func(_ any, ss grpc.ServerStream) error { + impl, err := resolve[H](ss.Context()) + if err != nil { + return err + } + return orig(impl, ss) + } +} + +// resolve is H from the call's scope, as a status error when it fails. +func resolve[H any](ctx context.Context) (H, error) { + var zero H + s, ok := di.FromContext(ctx) + if !ok { + return zero, status.Error(codes.Internal, "digrpc: no call scope on the context; is the Interceptor on this server?") + } + impl, err := s.Resolve[H]() + if err == nil { + if any(impl) == nil { + return zero, status.Error(codes.Internal, "digrpc: the service resolved to nil") + } + return impl, nil + } + // A constructor's own status, not the build error wrapping it, is what + // the client should see. + if se, ok := errors.AsType[interface { + error + GRPCStatus() *status.Status + }](err); ok { + return zero, se.GRPCStatus().Err() + } + return zero, status.Error(codes.Internal, "digrpc: the service could not be built") +} diff --git a/digrpc/register_test.go b/digrpc/register_test.go new file mode 100644 index 0000000..b1c3934 --- /dev/null +++ b/digrpc/register_test.go @@ -0,0 +1,237 @@ +package digrpc_test + +import ( + "context" + "errors" + "net" + "strings" + "testing" + + "github.com/floatdrop/di" + "github.com/floatdrop/di/digrpc" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" +) + +type caller struct{ name string } + +// newCaller is where a call is validated: its status error is the call's. +func newCaller(c *digrpc.Call) (*caller, error) { + md, _ := metadata.FromIncomingContext(c.Context) + name := strings.Join(md.Get("x-caller"), ",") + switch name { + case "": + return nil, status.Error(codes.Unauthenticated, "no x-caller") + case "broken": + return nil, errors.New("something the client should not read") + } + return &caller{name}, nil +} + +// health is the whole service implementation, built per call. +type health struct { + grpc_health_v1.UnimplementedHealthServer + caller *caller + log *[]string +} + +func (h *health) Check(context.Context, *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) { + *h.log = append(*h.log, "check by "+h.caller.name) + return &grpc_health_v1.HealthCheckResponse{Status: grpc_health_v1.HealthCheckResponse_SERVING}, nil +} + +func (h *health) Watch(_ *grpc_health_v1.HealthCheckRequest, ss grpc.ServerStreamingServer[grpc_health_v1.HealthCheckResponse]) error { + *h.log = append(*h.log, "watch by "+h.caller.name) + return ss.Send(&grpc_health_v1.HealthCheckResponse{Status: grpc_health_v1.HealthCheckResponse_NOT_SERVING}) +} + +// serve registers health through Register on a server with the interceptor +// and one unary interceptor of the test's own, and returns a client. +func serve(t *testing.T, log *[]string, opts ...grpc.ServerOption) grpc_health_v1.HealthClient { + t.Helper() + app := di.Test(t) + app.Use(digrpc.Module) + app.Wire[*caller](newCaller).Scoped() + app.Wire[*health](func(c *caller) *health { + *log = append(*log, "build") + return &health{caller: c, log: log} + }).Scoped() + app.Wire[*grpc.Server](func(ic digrpc.Interceptor) *grpc.Server { + srv := grpc.NewServer(append(ic.Options(), opts...)...) + digrpc.Register[*health](srv, &grpc_health_v1.Health_ServiceDesc) + return srv + }) + if err := app.Validate(di.Provided[*digrpc.Call]()).Err(); err != nil { + t.Fatal(err) + } + lis := bufconn.Listen(1 << 20) + srv := app.Get[*grpc.Server]() + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + conn, err := grpc.NewClient("passthrough:///bufnet", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }), + grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + return grpc_health_v1.NewHealthClient(conn) +} + +func as(t *testing.T, name string) context.Context { + t.Helper() + return metadata.AppendToOutgoingContext(t.Context(), "x-caller", name) +} + +func TestRegisterBuildsTheServicePerCall(t *testing.T) { + var log []string + client := serve(t, &log) + res, err := client.Check(as(t, "ada"), &grpc_health_v1.HealthCheckRequest{}) + if err != nil || res.GetStatus() != grpc_health_v1.HealthCheckResponse_SERVING { + t.Fatalf("unary: %v, %v", res, err) + } + w, err := client.Watch(as(t, "bob"), &grpc_health_v1.HealthCheckRequest{}) + if err != nil { + t.Fatal(err) + } + if msg, err := w.Recv(); err != nil || msg.GetStatus() != grpc_health_v1.HealthCheckResponse_NOT_SERVING { + t.Fatalf("stream: %v, %v", msg, err) + } + want := []string{"build", "check by ada", "build", "watch by bob"} + if strings.Join(log, ",") != strings.Join(want, ",") { + t.Errorf("log %q, want %q", log, want) + } +} + +func TestRegisterResolvesAfterTheInterceptors(t *testing.T) { + var log []string + seen := func(ctx context.Context, req any, info *grpc.UnaryServerInfo, h grpc.UnaryHandler) (any, error) { + log = append(log, "interceptor "+info.FullMethod) + if info.Server != nil { + t.Error("the info names a server before the service is built") + } + return h(ctx, req) + } + client := serve(t, &log, grpc.ChainUnaryInterceptor(seen)) + if _, err := client.Check(as(t, "ada"), &grpc_health_v1.HealthCheckRequest{}); err != nil { + t.Fatal(err) + } + want := []string{"interceptor /grpc.health.v1.Health/Check", "build", "check by ada"} + if strings.Join(log, ",") != strings.Join(want, ",") { + t.Errorf("log %q, want %q", log, want) + } +} + +func TestRegisterFailsTheCallWithTheConstructorsStatus(t *testing.T) { + var log []string + client := serve(t, &log) + _, err := client.Check(t.Context(), &grpc_health_v1.HealthCheckRequest{}) + if st := status.Convert(err); st.Code() != codes.Unauthenticated || st.Message() != "no x-caller" { + t.Errorf("got %v, want Unauthenticated with the constructor's message", err) + } + _, err = client.Check(as(t, "broken"), &grpc_health_v1.HealthCheckRequest{}) + if st := status.Convert(err); st.Code() != codes.Internal || strings.Contains(st.Message(), "should not read") { + t.Errorf("got %v, want Internal without the constructor's text", err) + } + if len(log) != 0 { + t.Errorf("a rejected call built the service: %q", log) + } +} + +func TestRegisterLeavesUnimplementedMethodsAlone(t *testing.T) { + var log []string + client := serve(t, &log) + _, err := client.List(as(t, "ada"), &grpc_health_v1.HealthListRequest{}) + if status.Code(err) != codes.Unimplemented { + t.Errorf("List: %v, want Unimplemented from the embedded type", err) + } +} + +func TestRegisterWithoutTheInterceptorIsInternal(t *testing.T) { + srv := grpc.NewServer() + digrpc.Register[*health](srv, &grpc_health_v1.Health_ServiceDesc) + lis := bufconn.Listen(1 << 20) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + conn, err := grpc.NewClient("passthrough:///bufnet", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }), + grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + _, err = grpc_health_v1.NewHealthClient(conn).Check(t.Context(), &grpc_health_v1.HealthCheckRequest{}) + if st := status.Convert(err); st.Code() != codes.Internal || !strings.Contains(st.Message(), "Interceptor") { + t.Errorf("got %v, want Internal naming the Interceptor", err) + } +} + +type notAHealth struct{} + +func TestRegisterRejectsATypeThatDoesNotImplement(t *testing.T) { + defer func() { + msg, _ := recover().(string) + if !strings.Contains(msg, "notAHealth does not implement") { + t.Errorf("panic %q, want one naming the type", msg) + } + }() + digrpc.Register[notAHealth](grpc.NewServer(), &grpc_health_v1.Health_ServiceDesc) +} + +func TestRegisterAcceptsTheServiceInterface(t *testing.T) { + var log []string + app := di.Test(t) + app.Use(digrpc.Module) + app.Wire[*caller](newCaller).Scoped() + app.Wire[grpc_health_v1.HealthServer](func(c *caller) grpc_health_v1.HealthServer { + if c.name == "nobody" { + return nil + } + log = append(log, "build") + return &health{caller: c, log: &log} + }).Scoped() + srv := grpc.NewServer(digrpc.New(app).Options()...) + digrpc.Register[grpc_health_v1.HealthServer](srv, &grpc_health_v1.Health_ServiceDesc) + lis := bufconn.Listen(1 << 20) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + conn, err := grpc.NewClient("passthrough:///bufnet", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }), + grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + client := grpc_health_v1.NewHealthClient(conn) + + if _, err := client.Check(as(t, "ada"), &grpc_health_v1.HealthCheckRequest{}); err != nil { + t.Fatal("unary:", err) + } + w, err := client.Watch(as(t, "bob"), &grpc_health_v1.HealthCheckRequest{}) + if err != nil { + t.Fatal(err) + } + if _, err := w.Recv(); err != nil { + t.Fatal("stream:", err) + } + want := []string{"build", "check by ada", "build", "watch by bob"} + if strings.Join(log, ",") != strings.Join(want, ",") { + t.Errorf("log %q, want %q", log, want) + } + _, err = client.Check(as(t, "nobody"), &grpc_health_v1.HealthCheckRequest{}) + if st := status.Convert(err); st.Code() != codes.Internal || !strings.Contains(st.Message(), "nil") { + t.Errorf("nil unary: %v, want Internal naming nil", err) + } + w, err = client.Watch(as(t, "nobody"), &grpc_health_v1.HealthCheckRequest{}) + if err != nil { + t.Fatal(err) + } + if _, err := w.Recv(); status.Code(err) != codes.Internal { + t.Errorf("nil stream: %v, want Internal", err) + } +} diff --git a/examples/grpc/main.go b/examples/grpc/main.go index 2c723e7..c79e99d 100644 --- a/examples/grpc/main.go +++ b/examples/grpc/main.go @@ -1,8 +1,9 @@ // Graceful shutdown of a gRPC server, with a service built per call. // // The digrpc interceptor opens a scope for every call and registers the -// *digrpc.Call in it, so a Caller declared Scoped is built per call from the -// call's metadata. Run starts the scope, waits for SIGINT/SIGTERM or a +// *digrpc.Call in it, and digrpc.Register serves the health service with an +// implementation resolved from that scope, so a Scoped Health is built per +// call with a Caller read from the call's metadata. Run starts the scope, waits for SIGINT/SIGTERM or a // Shutdown call, then stops everything in reverse order with a bounded // context. The server's OnDrain calls GracefulStop, which stops accepting // calls and waits for in-flight ones. Draining runs before anything is torn @@ -36,28 +37,27 @@ func NewCaller(c *digrpc.Call) *Caller { return &Caller{Name: cmp.Or(strings.Join(md.Get("x-caller"), ","), "anonymous")} } -// Health is a service implementation. grpc registers one value for the whole -// server, so it is a singleton that reaches per-call services through the -// scope on its context. +// Health is the service implementation, built per call because it takes the +// Caller. Methods it does not define fall through to the embedded type. type Health struct { grpc_health_v1.UnimplementedHealthServer - db *DB + db *DB + caller *Caller } -func NewHealth(db *DB) *Health { return &Health{db: db} } +func NewHealth(db *DB, c *Caller) *Health { return &Health{db: db, caller: c} } -func (h *Health) Check(ctx context.Context, _ *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) { - scope, _ := di.FromContext(ctx) - caller := scope.Get[*Caller]() +func (h *Health) Check(context.Context, *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) { time.Sleep(2 * time.Second) // simulate slow work that must not be cut short - log.Println("checked by", caller.Name, "against", h.db.dsn) + log.Println("checked by", h.caller.Name, "against", h.db.dsn) return &grpc_health_v1.HealthCheckResponse{Status: grpc_health_v1.HealthCheckResponse_SERVING}, nil } -// NewServer is a plain constructor: the interceptor arrives as a dependency. -func NewServer(ic digrpc.Interceptor, h *Health) *grpc.Server { +// NewServer is a plain constructor: the interceptor arrives as a dependency, +// and Register resolves *Health from each call's scope. +func NewServer(ic digrpc.Interceptor) *grpc.Server { srv := grpc.NewServer(ic.Options()...) - grpc_health_v1.RegisterHealthServer(srv, h) + digrpc.Register[*Health](srv, &grpc_health_v1.Health_ServiceDesc) return srv } @@ -68,7 +68,7 @@ func main() { app.Wire[*DB](func() *DB { return &DB{dsn: "postgres://localhost/app"} }). OnStop(func(ctx context.Context, db *DB) error { log.Println("db closed"); return nil }) app.Wire[*Caller](NewCaller).Scoped() - app.Wire[*Health](NewHealth) + app.Wire[*Health](NewHealth).Scoped() app.Wire[*grpc.Server](NewServer). Eager().