diff options
Diffstat (limited to 'plugins/rpc')
-rwxr-xr-x | plugins/rpc/rpc.go | 68 |
1 files changed, 37 insertions, 31 deletions
diff --git a/plugins/rpc/rpc.go b/plugins/rpc/rpc.go index 0f6c9753..d272fc72 100755 --- a/plugins/rpc/rpc.go +++ b/plugins/rpc/rpc.go @@ -3,6 +3,8 @@ package rpc import ( "net/rpc" + "go.uber.org/zap" + "github.com/spiral/endure" "github.com/spiral/endure/errors" "github.com/spiral/goridge/v2" @@ -20,65 +22,70 @@ type RPCPluggable interface { // ServiceName contains default service name. const ServiceName = "rpc" -type services struct { - service interface{} - name string -} - // Service is RPC service. type Service struct { + cfg Config + log *zap.Logger rpc *rpc.Server - services []services - config Config + services []RPCPluggable close chan struct{} } // Init rpc service. Must return true if service is enabled. -func (s *Service) Init(cfg config.Provider) error { +func (s *Service) Init(cfg config.Provider, log *zap.Logger) error { if !cfg.Has(ServiceName) { return errors.E(errors.Disabled) } - err := cfg.UnmarshalKey(ServiceName, &s.config) + err := cfg.UnmarshalKey(ServiceName, &s.cfg) if err != nil { return err } - s.config.InitDefaults() + s.cfg.InitDefaults() - if s.config.Disabled { + if s.cfg.Disabled { return errors.E(errors.Disabled) } - return s.config.Valid() -} + s.log = log -// Name contains service name. -func (s *Service) Name() string { - return ServiceName + return s.cfg.Valid() } // Serve serves the service. func (s *Service) Serve() chan error { - s.close = make(chan struct{}, 1) errCh := make(chan error, 1) + s.close = make(chan struct{}, 1) s.rpc = rpc.NewServer() + names := make([]string, 0, len(s.services)) + // Attach all services for i := 0; i < len(s.services); i++ { - err := s.Register(s.services[i].name, s.services[i].service) + svc, err := s.services[i].RPCService() + if err != nil { + errCh <- errors.E(errors.Op("register service"), err) + return errCh + } + + err = s.Register(s.services[i].Name(), svc) if err != nil { errCh <- errors.E(errors.Op("register service"), err) return errCh } + + names = append(names, s.services[i].Name()) } - ln, err := s.config.Listener() + ln, err := s.cfg.Listener() if err != nil { errCh <- err return errCh } + s.log.Debug("Started RPC service", zap.String("address", s.cfg.Listen), zap.Any("services", names)) + go func() { for { select { @@ -109,22 +116,21 @@ func (s *Service) Stop() error { return nil } +// Name contains service name. +func (s *Service) Name() string { + return ServiceName +} + +// Depends declares services to collect for RPC. func (s *Service) Depends() []interface{} { return []interface{}{ - s.RegisterService, + s.RegisterPlugin, } } -func (s *Service) RegisterService(p RPCPluggable) error { - service, err := p.RPCService() - if err != nil { - return err - } - - s.services = append(s.services, services{ - service: service, - name: p.Name(), - }) +// RegisterPlugin registers RPC service plugin. +func (s *Service) RegisterPlugin(p RPCPluggable) error { + s.services = append(s.services, p) return nil } @@ -146,7 +152,7 @@ func (s *Service) Register(name string, svc interface{}) error { // Client creates new RPC client. func (s *Service) Client() (*rpc.Client, error) { - conn, err := s.config.Dialer() + conn, err := s.cfg.Dialer() if err != nil { return nil, err } |