summaryrefslogtreecommitdiff
path: root/plugins/rpc
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/rpc')
-rwxr-xr-xplugins/rpc/plugin.go (renamed from plugins/rpc/rpc.go)92
-rw-r--r--plugins/rpc/tests/.rr-rpc-disabled.yaml3
-rw-r--r--plugins/rpc/tests/.rr.yaml3
-rw-r--r--plugins/rpc/tests/plugin1.go42
-rw-r--r--plugins/rpc/tests/plugin2.go54
-rw-r--r--plugins/rpc/tests/rpc_test.go169
6 files changed, 322 insertions, 41 deletions
diff --git a/plugins/rpc/rpc.go b/plugins/rpc/plugin.go
index f299bd55..6401c0e2 100755
--- a/plugins/rpc/rpc.go
+++ b/plugins/rpc/plugin.go
@@ -1,18 +1,19 @@
package rpc
import (
+ "net"
"net/rpc"
-
- "go.uber.org/zap"
+ "sync/atomic"
"github.com/spiral/endure"
"github.com/spiral/errors"
"github.com/spiral/goridge/v2"
+ "github.com/spiral/roadrunner/v2/log"
"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.
@@ -20,21 +21,23 @@ type RPCPluggable interface {
}
// ServiceName contains default service name.
-const ServiceName = "rpc"
+const ServiceName = "RPC"
-// Service is RPC service.
-type Service struct {
+// Plugin is RPC service.
+type Plugin struct {
cfg Config
- log *zap.Logger
+ log log.Logger
rpc *rpc.Server
- services []RPCPluggable
- close chan struct{}
+ services []Pluggable
+ listener net.Listener
+ closed *uint32
}
// 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 log.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,66 +47,68 @@ 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
+ state := uint32(0)
+ s.closed = &state
+ atomic.StoreUint32(s.closed, 0)
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()
+ var err error
+ s.listener, 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))
+ s.log.Debug("Started RPC service", "address", s.cfg.Listen, "services", services)
go func() {
for {
- select {
- case <-s.close:
- // log error
- err := ln.Close()
- if err != nil {
- errCh <- errors.E(errors.Op("close RPC socket"), err)
- }
- return
- default:
- conn, err := ln.Accept()
- if err != nil {
- continue
+ conn, err := s.listener.Accept()
+ if err != nil {
+ if atomic.LoadUint32(s.closed) == 1 {
+ // just log and continue, this is not a critical issue, we just called Stop
+ s.log.Error("listener accept error, connection closed", "error", err)
+ return
}
- go s.rpc.ServeCodec(goridge.NewCodec(conn))
+ s.log.Error("listener accept error", "error", err)
+ errCh <- errors.E(errors.Op("listener accept"), errors.Serve, err)
+ return
}
+
+ go s.rpc.ServeCodec(goridge.NewCodec(conn))
}
}()
@@ -111,25 +116,30 @@ func (s *Service) Serve() chan error {
}
// Stop stops the service.
-func (s *Service) Stop() error {
- s.close <- struct{}{}
+func (s *Plugin) Stop() error {
+ // store closed state
+ atomic.StoreUint32(s.closed, 1)
+ err := s.listener.Close()
+ if err != nil {
+ return errors.E(errors.Op("stop RPC socket"), err)
+ }
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 +152,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 +161,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
diff --git a/plugins/rpc/tests/.rr-rpc-disabled.yaml b/plugins/rpc/tests/.rr-rpc-disabled.yaml
new file mode 100644
index 00000000..624fb3c5
--- /dev/null
+++ b/plugins/rpc/tests/.rr-rpc-disabled.yaml
@@ -0,0 +1,3 @@
+rpc:
+ listen: tcp://127.0.0.1:6001
+ disabled: true \ No newline at end of file
diff --git a/plugins/rpc/tests/.rr.yaml b/plugins/rpc/tests/.rr.yaml
new file mode 100644
index 00000000..76e8b440
--- /dev/null
+++ b/plugins/rpc/tests/.rr.yaml
@@ -0,0 +1,3 @@
+rpc:
+ listen: tcp://127.0.0.1:6001
+ disabled: false \ No newline at end of file
diff --git a/plugins/rpc/tests/plugin1.go b/plugins/rpc/tests/plugin1.go
new file mode 100644
index 00000000..788e6a2c
--- /dev/null
+++ b/plugins/rpc/tests/plugin1.go
@@ -0,0 +1,42 @@
+package tests
+
+import (
+ "fmt"
+
+ "github.com/spiral/roadrunner/v2/plugins/config"
+)
+
+type Plugin1 struct {
+ config config.Configurer
+}
+
+func (p1 *Plugin1) Init(cfg config.Configurer) error {
+ p1.config = cfg
+ return nil
+}
+
+func (p1 *Plugin1) Serve() chan error {
+ errCh := make(chan error, 1)
+ return errCh
+}
+
+func (p1 *Plugin1) Stop() error {
+ return nil
+}
+
+func (p1 *Plugin1) Name() string {
+ return "rpc_test.plugin1"
+}
+
+func (p1 *Plugin1) RPCService() (interface{}, error) {
+ return &PluginRpc{srv: p1}, nil
+}
+
+type PluginRpc struct {
+ srv *Plugin1
+}
+
+func (r *PluginRpc) Hello(in string, out *string) error {
+ *out = fmt.Sprintf("Hello, username: %s", in)
+ return nil
+}
diff --git a/plugins/rpc/tests/plugin2.go b/plugins/rpc/tests/plugin2.go
new file mode 100644
index 00000000..854bf097
--- /dev/null
+++ b/plugins/rpc/tests/plugin2.go
@@ -0,0 +1,54 @@
+package tests
+
+import (
+ "net"
+ "net/rpc"
+ "time"
+
+ "github.com/spiral/errors"
+ "github.com/spiral/goridge/v2"
+)
+
+// plugin2 makes a call to the plugin1 via RPC
+// this is just a simulation of external call FOR TEST
+// you don't need to do such things :)
+type Plugin2 struct {
+}
+
+func (p2 *Plugin2) Init() error {
+ return nil
+}
+
+func (p2 *Plugin2) Serve() chan error {
+ errCh := make(chan error, 1)
+
+ go func() {
+ time.Sleep(time.Second * 3)
+
+ conn, err := net.Dial("tcp", "127.0.0.1:6001")
+ if err != nil {
+ errCh <- errors.E(errors.Serve, err)
+ return
+ }
+ client := rpc.NewClientWithCodec(goridge.NewClientCodec(conn))
+ var ret string
+ err = client.Call("rpc_test.plugin1.Hello", "Valery", &ret)
+ if err != nil {
+ errCh <- err
+ return
+ }
+ if ret != "Hello, username: Valery" {
+ errCh <- errors.E("wrong response")
+ return
+ }
+ // to stop exec
+ errCh <- errors.E(errors.Disabled)
+ return
+ }()
+
+ return errCh
+}
+
+func (p2 *Plugin2) Stop() error {
+ return nil
+}
diff --git a/plugins/rpc/tests/rpc_test.go b/plugins/rpc/tests/rpc_test.go
new file mode 100644
index 00000000..88267dfb
--- /dev/null
+++ b/plugins/rpc/tests/rpc_test.go
@@ -0,0 +1,169 @@
+package tests
+
+import (
+ "os"
+ "os/signal"
+ "syscall"
+ "testing"
+ "time"
+
+ "github.com/spiral/endure"
+ "github.com/spiral/errors"
+ "github.com/spiral/roadrunner/v2/plugins/config"
+ "github.com/spiral/roadrunner/v2/plugins/logger"
+ "github.com/spiral/roadrunner/v2/plugins/rpc"
+ "github.com/stretchr/testify/assert"
+)
+
+// graph https://bit.ly/3ensdNb
+func TestRpcInit(t *testing.T) {
+ cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.DebugLevel))
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = cont.Register(&Plugin1{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = cont.Register(&Plugin2{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ v := &config.Viper{}
+ v.Path = ".rr.yaml"
+ v.Prefix = "rr"
+ err = cont.Register(v)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = cont.Register(&rpc.Plugin{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = cont.Register(&logger.ZapLogger{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = cont.Init()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ ch, err := cont.Serve()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ sig := make(chan os.Signal, 1)
+
+ signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+ tt := time.NewTimer(time.Second * 10)
+
+ for {
+ select {
+ case e := <-ch:
+ // just stop, this is ok
+ if errors.Is(errors.Disabled, e.Error) {
+ return
+ }
+ assert.Fail(t, "error", e.Error.Error())
+ err = cont.Stop()
+ if err != nil {
+ assert.FailNow(t, "error", err.Error())
+ }
+ case <-sig:
+ err = cont.Stop()
+ if err != nil {
+ assert.FailNow(t, "error", err.Error())
+ }
+ return
+ case <-tt.C:
+ // timeout
+ err = cont.Stop()
+ if err != nil {
+ assert.FailNow(t, "error", err.Error())
+ }
+ assert.Fail(t, "timeout")
+ }
+ }
+}
+
+// graph https://bit.ly/3ensdNb
+func TestRpcDisabled(t *testing.T) {
+ cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.DebugLevel))
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = cont.Register(&Plugin1{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = cont.Register(&Plugin2{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ v := &config.Viper{}
+ v.Path = ".rr-rpc-disabled.yaml"
+ v.Prefix = "rr"
+ err = cont.Register(v)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = cont.Register(&rpc.Plugin{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = cont.Register(&logger.ZapLogger{})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ err = cont.Init()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ ch, err := cont.Serve()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ sig := make(chan os.Signal, 1)
+
+ signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+ tt := time.NewTimer(time.Second * 20)
+
+ for {
+ select {
+ case e := <-ch:
+ // RPC is turned off, should be and dial error
+ if errors.Is(errors.Disabled, e.Error) {
+ assert.FailNow(t, "should not be disabled error")
+ }
+ assert.Error(t, e.Error)
+ err = cont.Stop()
+ assert.Error(t, err)
+ return
+ case <-sig:
+ err = cont.Stop()
+ if err != nil {
+ assert.FailNow(t, "error", err.Error())
+ }
+ return
+ case <-tt.C:
+ // timeout
+ return
+ }
+ }
+}
generated by cgit v1.2.3 (git 2.25.1) at 2025-01-23 13:15:23 +0000