rpcserver.go 2.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586
  1. package internal
  2. import (
  3. "net"
  4. "github.com/tal-tech/go-zero/core/proc"
  5. "github.com/tal-tech/go-zero/core/stat"
  6. "github.com/tal-tech/go-zero/zrpc/internal/serverinterceptors"
  7. "google.golang.org/grpc"
  8. )
  9. type (
  10. // ServerOption defines the method to customize a rpcServerOptions.
  11. ServerOption func(options *rpcServerOptions)
  12. rpcServerOptions struct {
  13. metrics *stat.Metrics
  14. }
  15. rpcServer struct {
  16. name string
  17. *baseRpcServer
  18. }
  19. )
  20. func init() {
  21. InitLogger()
  22. }
  23. // NewRpcServer returns a Server.
  24. func NewRpcServer(address string, opts ...ServerOption) Server {
  25. var options rpcServerOptions
  26. for _, opt := range opts {
  27. opt(&options)
  28. }
  29. if options.metrics == nil {
  30. options.metrics = stat.NewMetrics(address)
  31. }
  32. return &rpcServer{
  33. baseRpcServer: newBaseRpcServer(address, options.metrics),
  34. }
  35. }
  36. func (s *rpcServer) SetName(name string) {
  37. s.name = name
  38. s.baseRpcServer.SetName(name)
  39. }
  40. func (s *rpcServer) Start(register RegisterFn) error {
  41. lis, err := net.Listen("tcp", s.address)
  42. if err != nil {
  43. return err
  44. }
  45. unaryInterceptors := []grpc.UnaryServerInterceptor{
  46. serverinterceptors.UnaryTracingInterceptor(s.name),
  47. serverinterceptors.UnaryCrashInterceptor(),
  48. serverinterceptors.UnaryStatInterceptor(s.metrics),
  49. serverinterceptors.UnaryPrometheusInterceptor(),
  50. }
  51. unaryInterceptors = append(unaryInterceptors, s.unaryInterceptors...)
  52. streamInterceptors := []grpc.StreamServerInterceptor{
  53. serverinterceptors.StreamCrashInterceptor,
  54. }
  55. streamInterceptors = append(streamInterceptors, s.streamInterceptors...)
  56. options := append(s.options, WithUnaryServerInterceptors(unaryInterceptors...),
  57. WithStreamServerInterceptors(streamInterceptors...))
  58. server := grpc.NewServer(options...)
  59. register(server)
  60. // we need to make sure all others are wrapped up
  61. // so we do graceful stop at shutdown phase instead of wrap up phase
  62. waitForCalled := proc.AddWrapUpListener(func() {
  63. server.GracefulStop()
  64. })
  65. defer waitForCalled()
  66. return server.Serve(lis)
  67. }
  68. // WithMetrics returns a func that sets metrics to a Server.
  69. func WithMetrics(metrics *stat.Metrics) ServerOption {
  70. return func(options *rpcServerOptions) {
  71. options.metrics = metrics
  72. }
  73. }