Update protoc plugin,

Update serverOptions,
Update codec.

Signed-off-by: Valery Piashchynski <[email protected]>

14 files changed, 206 insertions, 14 deletions

diff --git a/go.mod b/go.mod
index 97e2bdc6..b7141c4a 100644
--- a/go.mod
+++ b/go.mod
@@ -88,6 +88,7 @@ require (
 	go.uber.org/atomic v1.9.0 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/tools v0.1.5 // indirect
+	google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c // indirect
 	gopkg.in/ini.v1 v1.63.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
diff --git a/go.sum b/go.sum
index 6096347d..7dec39f0 100644
--- a/go.sum
+++ b/go.sum
@@ -769,6 +769,7 @@ google.golang.org/genproto v0.0.0-20210303154014-9728d6b83eeb/go.mod h1:FWY/as6D
 google.golang.org/genproto v0.0.0-20210310155132-4ce2db91004e/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210319143718-93e7006c17a6/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210402141018-6c239bbf2bb1/go.mod h1:9lPAdzaEmUacj36I+k7YKbEc5CXzPIeORRgDAUOu28A=
+google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c h1:wtujag7C+4D6KMoulW9YauvK2lgdvCMS260jsqqBXr0=
 google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
diff --git a/plugins/grpc/codec.go b/plugins/grpc/codec.go
index dd299601..aeb373b9 100644
--- a/plugins/grpc/codec.go
+++ b/plugins/grpc/codec.go
@@ -4,6 +4,7 @@ import "google.golang.org/grpc/encoding"
 
 type rawMessage []byte
 
+const cName string = "proto"
 const rm string = "rawMessage"
 
 func (r rawMessage) Reset()       {}
@@ -31,6 +32,10 @@ func (c *codec) Unmarshal(data []byte, v interface{}) error {
 	return c.base.Unmarshal(data, v)
 }
 
+func (c *codec) Name() string {
+	return cName
+}
+
 // String return codec name.
 func (c *codec) String() string {
 	return "raw:" + c.base.Name()
diff --git a/plugins/grpc/config.go b/plugins/grpc/config.go
index 9a9f8c2c..53c50e22 100644
--- a/plugins/grpc/config.go
+++ b/plugins/grpc/config.go
@@ -32,3 +32,11 @@ type TLS struct {
 func (c *Config) InitDefaults() {
 
 }
+
+func (c *Config) EnableTLS() bool {
+	if c.TLS != nil {
+		return (c.TLS.RootCA != "" && c.TLS.Key != "" && c.TLS.Cert != "") || (c.TLS.Key != "" && c.TLS.Cert != "")
+	}
+
+	return false
+}
diff --git a/plugins/grpc/plugin.go b/plugins/grpc/plugin.go
index 4871a8a7..b424c5d0 100644
--- a/plugins/grpc/plugin.go
+++ b/plugins/grpc/plugin.go
@@ -4,6 +4,8 @@ import (
 	"github.com/spiral/errors"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/logger"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/encoding"
 )
 
 const (
@@ -11,6 +13,9 @@ const (
 )
 
 type Plugin struct {
+	config *Config
+	opts   []grpc.ServerOption
+
 	cfg config.Configurer
 	log logger.Logger
 }
@@ -18,6 +23,9 @@ type Plugin struct {
 func (p *Plugin) Init(cfg config.Configurer, log logger.Logger) error {
 	const op = errors.Op("grpc_plugin_init")
 
+	// register the codec
+	encoding.RegisterCodec(&codec{})
+
 	return nil
 }
 
diff --git a/plugins/grpc/protoc-gen-php-grpc/main.go b/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/main.go
index c9c4a573..0894a7a8 100644
--- a/plugins/grpc/protoc-gen-php-grpc/main.go
+++ b/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/main.go
@@ -27,7 +27,7 @@ import (
 	"io/ioutil"
 	"os"
 
-	"github.com/spiral/roadrunner/v2/plugins/grpc/protoc-gen-php-grpc/php"
+	"github.com/spiral/roadrunner/v2/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/php"
 	"google.golang.org/protobuf/proto"
 	plugin "google.golang.org/protobuf/types/pluginpb"
 )
diff --git a/plugins/grpc/protoc-gen-php-grpc/php/generate.go b/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/php/generate.go
index 03c48ac8..03c48ac8 100644
--- a/plugins/grpc/protoc-gen-php-grpc/php/generate.go
+++ b/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/php/generate.go
diff --git a/plugins/grpc/protoc-gen-php-grpc/php/keywords.go b/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/php/keywords.go
index 32579e33..32579e33 100644
--- a/plugins/grpc/protoc-gen-php-grpc/php/keywords.go
+++ b/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/php/keywords.go
diff --git a/plugins/grpc/protoc-gen-php-grpc/php/ns.go b/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/php/ns.go
index c1dc3898..c1dc3898 100644
--- a/plugins/grpc/protoc-gen-php-grpc/php/ns.go
+++ b/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/php/ns.go
diff --git a/plugins/grpc/protoc-gen-php-grpc/php/template.go b/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/php/template.go
index e00c6fdd..e00c6fdd 100644
--- a/plugins/grpc/protoc-gen-php-grpc/php/template.go
+++ b/plugins/grpc/protoc_plugins/protoc-gen-php-grpc/php/template.go
diff --git a/plugins/grpc/server.go b/plugins/grpc/server.go
new file mode 100644
index 00000000..53b0b2ae
--- /dev/null
+++ b/plugins/grpc/server.go
@@ -0,0 +1,119 @@
+package grpc
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"os"
+
+	"github.com/spiral/errors"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/keepalive"
+)
+
+func (p *Plugin) createGRPCserver() (*grpc.Server, error) {
+	const op = errors.Op("grpc_plugin_create_server")
+	opts, err := p.serverOptions()
+	if err != nil {
+		return nil, errors.E(op, err)
+	}
+
+	server := grpc.NewServer(opts...)
+
+	/*
+		proto descriptors parser
+	*/
+
+	return server, nil
+}
+
+func (p *Plugin) interceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
+	// start := time.Now()
+	resp, err = handler(ctx, req)
+
+	// svc.throw(EventUnaryCall, &UnaryCallEvent{
+	// Info:    info,
+	// Context: ctx,
+	// Error:   err,
+	// 	start:   start,
+	// 	elapsed: time.Since(start),
+	// })
+
+	return resp, err
+}
+
+func (p *Plugin) serverOptions() ([]grpc.ServerOption, error) {
+	const op = errors.Op("grpc_plugin_server_options")
+
+	var tcreds credentials.TransportCredentials
+	var opts []grpc.ServerOption
+	var cert tls.Certificate
+	var certPool *x509.CertPool
+	var rca []byte
+	var err error
+
+	if p.config.EnableTLS() {
+		// if client CA is not empty we combine it with Cert and Key
+		if p.config.TLS.RootCA != "" {
+			cert, err = tls.LoadX509KeyPair(p.config.TLS.Cert, p.config.TLS.Key)
+			if err != nil {
+				return nil, err
+			}
+
+			certPool, err = x509.SystemCertPool()
+			if err != nil {
+				return nil, err
+			}
+			if certPool == nil {
+				certPool = x509.NewCertPool()
+			}
+
+			rca, err = os.ReadFile(p.config.TLS.RootCA)
+			if err != nil {
+				return nil, err
+			}
+
+			if ok := certPool.AppendCertsFromPEM(rca); !ok {
+				return nil, errors.E(op, errors.Str("could not append Certs from PEM"))
+			}
+
+			tcreds = credentials.NewTLS(&tls.Config{
+				MinVersion:   tls.VersionTLS12,
+				ClientAuth:   tls.RequireAndVerifyClientCert,
+				Certificates: []tls.Certificate{cert},
+				ClientCAs:    certPool,
+			})
+		} else {
+			tcreds, err = credentials.NewServerTLSFromFile(p.config.TLS.Cert, p.config.TLS.Key)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		serverOptions := []grpc.ServerOption{
+			grpc.MaxSendMsgSize(int(p.config.MaxSendMsgSize)),
+			grpc.MaxRecvMsgSize(int(p.config.MaxRecvMsgSize)),
+			grpc.KeepaliveParams(keepalive.ServerParameters{
+				MaxConnectionIdle:     p.config.MaxConnectionIdle,
+				MaxConnectionAge:      p.config.MaxConnectionAge,
+				MaxConnectionAgeGrace: p.config.MaxConnectionAge,
+				Time:                  p.config.PingTime,
+				Timeout:               p.config.Timeout,
+			}),
+			grpc.MaxConcurrentStreams(uint32(p.config.MaxConcurrentStreams)),
+		}
+
+		opts = append(opts, grpc.Creds(tcreds))
+		opts = append(opts, serverOptions...)
+	}
+
+	opts = append(opts, p.opts...)
+
+	// custom codec is required to bypass protobuf, common interceptor used for debug and stats
+	return append(
+		opts,
+		grpc.UnaryInterceptor(p.interceptor),
+		// grpc.CustomCodec(&codec{encoding.GetCodec(encCodec)}),
+	), nil
+}
diff --git a/tests/plugins/grpc/configs/.rr-grpc-init.yaml b/tests/plugins/grpc/configs/.rr-grpc-init.yaml
index e69de29b..010e904e 100644
--- a/tests/plugins/grpc/configs/.rr-grpc-init.yaml
+++ b/tests/plugins/grpc/configs/.rr-grpc-init.yaml
@@ -0,0 +1,50 @@
+# GRPC service configuration
+grpc:
+    # socket to listen
+    listen: "tcp://localhost:9001"
+
+    # proto root file
+    proto: "test.proto"
+
+    # max send limit (MB)
+    max_send_msg_size: 50
+
+    # max receive limit (MB)
+    max_recv_msg_size: 50
+
+    # MaxConnectionIdle is a duration for the amount of time after which an
+    #	idle connection would be closed by sending a GoAway. Idleness duration is
+    #	defined since the most recent time the number of outstanding RPCs became
+    #	zero or the connection establishment.
+    max_connection_idle: 0s
+
+    # MaxConnectionAge is a duration for the maximum amount of time a
+    #	connection may exist before it will be closed by sending a GoAway. A
+    #	random jitter of +/-10% will be added to MaxConnectionAge to spread out
+    #	connection storms.
+    max_connection_age: 0s
+
+    # MaxConnectionAgeGrace is an additive period after MaxConnectionAge after
+    #	which the connection will be forcibly closed.
+    max_connection_age_grace: 0s
+
+    # MaxConnectionAgeGrace is an additive period after MaxConnectionAge after
+    #	which the connection will be forcibly closed.
+    max_concurrent_streams: 10
+
+    # After a duration of this time if the server doesn't see any activity it
+    #	pings the client to see if the transport is still alive.
+    #	If set below 1s, a minimum value of 1s will be used instead.
+    ping_time: 1s
+    
+    # After having pinged for keepalive check, the server waits for a duration
+    #	of Timeout and if no activity is seen even after that the connection is
+    #	closed.
+    timeout: 200s
+
+    # Usual workers pool configuration
+    pool:
+      num_workers: 2
+      max_jobs: 0
+      allocate_timeout: 60s
+      destroy_timeout: 60
diff --git a/tests/plugins/grpc/plugin_test.go b/tests/plugins/grpc/plugin_test.go
index 74a71c62..cfbe0121 100644
--- a/tests/plugins/grpc/plugin_test.go
+++ b/tests/plugins/grpc/plugin_test.go
@@ -19,7 +19,7 @@ func init() {
 }
 
 func build() error {
-	cmd := exec.Command("go", "build", "-o", "plugin", "../../../plugins/grpc/protoc-gen-php-grpc")
+	cmd := exec.Command("go", "build", "-o", "plugin", "../../../plugins/grpc/protoc_plugins/protoc-gen-php-grpc")
 	return cmd.Run()
 }
 
diff --git a/tests/plugins/logger/logger_test.go b/tests/plugins/logger/logger_test.go
index 05ca2d53..e077f0bc 100644
--- a/tests/plugins/logger/logger_test.go
+++ b/tests/plugins/logger/logger_test.go
@@ -346,18 +346,6 @@ func TestFileLogger(t *testing.T) {
 	wg.Wait()
 }
 
-func httpEcho(t *testing.T) {
-	req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:54224?hello=world", nil)
-	assert.NoError(t, err)
-
-	r, err := http.DefaultClient.Do(req)
-	assert.NoError(t, err)
-	assert.Equal(t, http.StatusCreated, r.StatusCode)
-
-	err = r.Body.Close()
-	assert.NoError(t, err)
-}
-
 func TestMarshalObjectLogging(t *testing.T) {
 	container, err := endure.NewContainer(nil, endure.RetryOnFail(true), endure.SetLogLevel(endure.ErrorLevel))
 	if err != nil {
@@ -428,3 +416,15 @@ func TestMarshalObjectLogging(t *testing.T) {
 	stopCh <- struct{}{}
 	wg.Wait()
 }
+
+func httpEcho(t *testing.T) {
+	req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:54224?hello=world", nil)
+	assert.NoError(t, err)
+
+	r, err := http.DefaultClient.Do(req)
+	assert.NoError(t, err)
+	assert.Equal(t, http.StatusCreated, r.StatusCode)
+
+	err = r.Body.Close()
+	assert.NoError(t, err)
+}