summaryrefslogtreecommitdiff
path: root/plugins/rpc/rpc.go
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/rpc/rpc.go')
-rwxr-xr-xplugins/rpc/rpc.go46
1 files changed, 24 insertions, 22 deletions
diff --git a/plugins/rpc/rpc.go b/plugins/rpc/rpc.go
index f299bd55..cda42aa0 100755
--- a/plugins/rpc/rpc.go
+++ b/plugins/rpc/rpc.go
@@ -11,8 +11,8 @@ import (
"github.com/spiral/roadrunner/v2/plugins/config"
)
-// RPCPluggable declares the ability to create set of public RPC methods.
-type RPCPluggable interface {
+// Pluggable declares the ability to create set of public RPC methods.
+type Pluggable interface {
endure.Named
// Provides RPC methods for the given service.
@@ -22,19 +22,20 @@ type RPCPluggable interface {
// ServiceName contains default service name.
const ServiceName = "rpc"
-// Service is RPC service.
-type Service struct {
+// Plugin is RPC service.
+type Plugin struct {
cfg Config
log *zap.Logger
rpc *rpc.Server
- services []RPCPluggable
+ services []Pluggable
close chan struct{}
}
// Init rpc service. Must return true if service is enabled.
-func (s *Service) Init(cfg config.Provider, log *zap.Logger) error {
+func (s *Plugin) Init(cfg config.Configurer, log *zap.Logger) error {
+ const op = errors.Op("RPC Init")
if !cfg.Has(ServiceName) {
- return errors.E(errors.Disabled)
+ return errors.E(op, errors.Disabled)
}
err := cfg.UnmarshalKey(ServiceName, &s.cfg)
@@ -44,38 +45,39 @@ func (s *Service) Init(cfg config.Provider, log *zap.Logger) error {
s.cfg.InitDefaults()
if s.cfg.Disabled {
- return errors.E(errors.Disabled)
+ return errors.E(op, errors.Disabled)
}
s.log = log
+ s.close = make(chan struct{}, 1)
return s.cfg.Valid()
}
// Serve serves the service.
-func (s *Service) Serve() chan error {
+func (s *Plugin) Serve() chan error {
+ const op = errors.Op("register service")
errCh := make(chan error, 1)
- s.close = make(chan struct{}, 1)
s.rpc = rpc.NewServer()
- names := make([]string, 0, len(s.services))
+ services := make([]string, 0, len(s.services))
// Attach all services
for i := 0; i < len(s.services); i++ {
svc, err := s.services[i].RPCService()
if err != nil {
- errCh <- errors.E(errors.Op("register service"), err)
+ errCh <- errors.E(op, err)
return errCh
}
err = s.Register(s.services[i].Name(), svc)
if err != nil {
- errCh <- errors.E(errors.Op("register service"), err)
+ errCh <- errors.E(op, err)
return errCh
}
- names = append(names, s.services[i].Name())
+ services = append(services, s.services[i].Name())
}
ln, err := s.cfg.Listener()
@@ -84,14 +86,14 @@ func (s *Service) Serve() chan error {
return errCh
}
- s.log.Debug("Started RPC service", zap.String("address", s.cfg.Listen), zap.Any("services", names))
+ s.log.Debug("Started RPC service", zap.String("address", s.cfg.Listen), zap.Any("services", services))
go func() {
for {
select {
case <-s.close:
// log error
- err := ln.Close()
+ err = ln.Close()
if err != nil {
errCh <- errors.E(errors.Op("close RPC socket"), err)
}
@@ -111,25 +113,25 @@ func (s *Service) Serve() chan error {
}
// Stop stops the service.
-func (s *Service) Stop() error {
+func (s *Plugin) Stop() error {
s.close <- struct{}{}
return nil
}
// Name contains service name.
-func (s *Service) Name() string {
+func (s *Plugin) Name() string {
return ServiceName
}
// Depends declares services to collect for RPC.
-func (s *Service) Depends() []interface{} {
+func (s *Plugin) Collects() []interface{} {
return []interface{}{
s.RegisterPlugin,
}
}
// RegisterPlugin registers RPC service plugin.
-func (s *Service) RegisterPlugin(p RPCPluggable) error {
+func (s *Plugin) RegisterPlugin(p Pluggable) error {
s.services = append(s.services, p)
return nil
}
@@ -142,7 +144,7 @@ func (s *Service) RegisterPlugin(p RPCPluggable) error {
// - one return value, of type error
// It returns an error if the receiver is not an exported type or has
// no suitable methods. It also logs the error using package log.
-func (s *Service) Register(name string, svc interface{}) error {
+func (s *Plugin) Register(name string, svc interface{}) error {
if s.rpc == nil {
return errors.E("RPC service is not configured")
}
@@ -151,7 +153,7 @@ func (s *Service) Register(name string, svc interface{}) error {
}
// Client creates new RPC client.
-func (s *Service) Client() (*rpc.Client, error) {
+func (s *Plugin) Client() (*rpc.Client, error) {
conn, err := s.cfg.Dialer()
if err != nil {
return nil, err