summaryrefslogtreecommitdiff
path: root/plugins/rpc/rpc.go
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/rpc/rpc.go')
-rwxr-xr-xplugins/rpc/rpc.go68
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
}