5 files changed, 218 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
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..feeae4af
--- /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 <- 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..85a11eec
--- /dev/null
+++ b/plugins/rpc/tests/rpc_test.go
@@ -0,0 +1,95 @@
+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), endure.Visualize(endure.StdOut, ""))
+	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")
+		}
+	}
+}