summaryrefslogtreecommitdiff
path: root/internal/rpc
diff options
context:
space:
mode:
Diffstat (limited to 'internal/rpc')
-rw-r--r--internal/rpc/client.go87
-rw-r--r--internal/rpc/client_test.go59
-rw-r--r--internal/rpc/test/config_rpc_ok_env.yaml2
3 files changed, 132 insertions, 16 deletions
diff --git a/internal/rpc/client.go b/internal/rpc/client.go
index 7d945add..1e0fbbac 100644
--- a/internal/rpc/client.go
+++ b/internal/rpc/client.go
@@ -1,19 +1,32 @@
// Package rpc contains wrapper around RPC client ONLY for internal usage.
+// Should be in sync with the RPC plugin
package rpc
import (
+ "errors"
+ "fmt"
+ "net"
"net/rpc"
+ "os"
+ "strings"
- "github.com/roadrunner-server/errors"
goridgeRpc "github.com/roadrunner-server/goridge/v3/pkg/rpc"
rpcPlugin "github.com/roadrunner-server/rpc/v2"
"github.com/spf13/viper"
)
+const (
+ prefix string = "rr"
+ rpcKey string = "rpc.listen"
+)
+
// NewClient creates client ONLY for internal usage (communication between our application with RR side).
// Client will be connected to the RPC.
-func NewClient(cfg string) (*rpc.Client, error) {
+func NewClient(cfg string, flags []string) (*rpc.Client, error) {
v := viper.New()
+ v.AutomaticEnv()
+ v.SetEnvPrefix(prefix)
+ v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
v.SetConfigFile(cfg)
err := v.ReadInConfig()
@@ -21,23 +34,77 @@ func NewClient(cfg string) (*rpc.Client, error) {
return nil, err
}
- if !v.IsSet(rpcPlugin.PluginName) {
- return nil, errors.E("rpc service disabled")
+ // automatically inject ENV variables using ${ENV} pattern
+ for _, key := range v.AllKeys() {
+ val := v.Get(key)
+ if s, ok := val.(string); ok {
+ v.Set(key, os.ExpandEnv(s))
+ }
}
- rpcConfig := &rpcPlugin.Config{}
+ // override config Flags
+ if len(flags) > 0 {
+ for _, f := range flags {
+ key, val, errP := parseFlag(f)
+ if errP != nil {
+ return nil, errP
+ }
- err = v.UnmarshalKey(rpcPlugin.PluginName, rpcConfig)
- if err != nil {
- return nil, err
+ v.Set(key, val)
+ }
}
- rpcConfig.InitDefaults()
+ // rpc.listen might be set by the -o flags or env variable
+ if !v.IsSet(rpcPlugin.PluginName) {
+ return nil, errors.New("rpc service not specified in the configuration. Tip: add\n rpc:\n\r listen: rr_rpc_address")
+ }
- conn, err := rpcConfig.Dialer()
+ conn, err := Dialer(v.GetString(rpcKey))
if err != nil {
return nil, err
}
return rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)), nil
}
+
+// Dialer creates rpc socket Dialer.
+func Dialer(addr string) (net.Conn, error) {
+ dsn := strings.Split(addr, "://")
+ if len(dsn) != 2 {
+ return nil, errors.New("invalid socket DSN (tcp://:6001, unix://file.sock)")
+ }
+
+ return net.Dial(dsn[0], dsn[1])
+}
+
+func parseFlag(flag string) (string, string, error) {
+ if !strings.Contains(flag, "=") {
+ return "", "", fmt.Errorf("invalid flag `%s`", flag)
+ }
+
+ parts := strings.SplitN(strings.TrimLeft(flag, " \"'`"), "=", 2)
+ if len(parts) < 2 {
+ return "", "", errors.New("usage: -o key=value")
+ }
+
+ if parts[0] == "" {
+ return "", "", errors.New("key should not be empty")
+ }
+
+ if parts[1] == "" {
+ return "", "", errors.New("value should not be empty")
+ }
+
+ return strings.Trim(parts[0], " \n\t"), parseValue(strings.Trim(parts[1], " \n\t")), nil
+}
+
+func parseValue(value string) string {
+ escape := []rune(value)[0]
+
+ if escape == '"' || escape == '\'' || escape == '`' {
+ value = strings.Trim(value, string(escape))
+ value = strings.ReplaceAll(value, fmt.Sprintf("\\%s", string(escape)), string(escape))
+ }
+
+ return value
+}
diff --git a/internal/rpc/client_test.go b/internal/rpc/client_test.go
index 0744e167..b9465a4b 100644
--- a/internal/rpc/client_test.go
+++ b/internal/rpc/client_test.go
@@ -2,9 +2,11 @@ package rpc_test
import (
"net"
+ "os"
"testing"
"github.com/roadrunner-server/roadrunner/v2/internal/rpc"
+ "github.com/stretchr/testify/require"
"github.com/roadrunner-server/config/v2"
"github.com/stretchr/testify/assert"
@@ -14,22 +16,22 @@ func TestNewClient_RpcServiceDisabled(t *testing.T) {
cfgPlugin := &config.Plugin{Type: "yaml", ReadInCfg: []byte{}}
assert.NoError(t, cfgPlugin.Init())
- c, err := rpc.NewClient("test/config_rpc_empty.yaml")
+ c, err := rpc.NewClient("test/config_rpc_empty.yaml", nil)
assert.Nil(t, c)
- assert.EqualError(t, err, "rpc service disabled")
+ assert.EqualError(t, err, "rpc service not specified in the configuration. Tip: add\n rpc:\n\r listen: rr_rpc_address")
}
func TestNewClient_WrongRcpConfiguration(t *testing.T) {
- c, err := rpc.NewClient("test/config_rpc_wrong.yaml")
+ c, err := rpc.NewClient("test/config_rpc_wrong.yaml", nil)
assert.Nil(t, c)
assert.Error(t, err)
- assert.Contains(t, err.Error(), "'' expected a map, got 'string'")
+ assert.Contains(t, err.Error(), "invalid socket DSN")
}
func TestNewClient_ConnectionError(t *testing.T) {
- c, err := rpc.NewClient("test/config_rpc_conn_err.yaml")
+ c, err := rpc.NewClient("test/config_rpc_conn_err.yaml", nil)
assert.Nil(t, c)
assert.Error(t, err)
@@ -42,7 +44,52 @@ func TestNewClient_SuccessfullyConnected(t *testing.T) {
defer func() { assert.NoError(t, l.Close()) }()
- c, err := rpc.NewClient("test/config_rpc_ok.yaml")
+ c, err := rpc.NewClient("test/config_rpc_ok.yaml", nil)
+
+ assert.NotNil(t, c)
+ assert.NoError(t, err)
+
+ defer func() { assert.NoError(t, c.Close()) }()
+}
+
+func TestNewClient_SuccessfullyConnectedOverride(t *testing.T) {
+ l, err := net.Listen("tcp", "127.0.0.1:55555")
+ assert.NoError(t, err)
+
+ defer func() { assert.NoError(t, l.Close()) }()
+
+ c, err := rpc.NewClient("test/config_rpc_empty.yaml", []string{"rpc.listen=tcp://127.0.0.1:55555"})
+
+ assert.NotNil(t, c)
+ assert.NoError(t, err)
+
+ defer func() { assert.NoError(t, c.Close()) }()
+}
+
+func TestNewClient_SuccessfullyConnectedEnv(t *testing.T) {
+ l, err := net.Listen("tcp", "127.0.0.1:55556")
+ assert.NoError(t, err)
+
+ defer func() { assert.NoError(t, l.Close()) }()
+
+ require.NoError(t, os.Setenv("RR_RPC_LISTEN", "tcp://127.0.0.1:55556"))
+ c, err := rpc.NewClient("test/config_rpc_ok.yaml", nil)
+
+ assert.NotNil(t, c)
+ assert.NoError(t, err)
+
+ defer func() { assert.NoError(t, c.Close()) }()
+}
+
+// ${} syntax
+func TestNewClient_SuccessfullyConnectedEnvDollarSyntax(t *testing.T) {
+ l, err := net.Listen("tcp", "127.0.0.1:55556")
+ assert.NoError(t, err)
+
+ defer func() { assert.NoError(t, l.Close()) }()
+
+ require.NoError(t, os.Setenv("RPC", "tcp://127.0.0.1:55556"))
+ c, err := rpc.NewClient("test/config_rpc_ok_env.yaml", nil)
assert.NotNil(t, c)
assert.NoError(t, err)
diff --git a/internal/rpc/test/config_rpc_ok_env.yaml b/internal/rpc/test/config_rpc_ok_env.yaml
new file mode 100644
index 00000000..fd0d3f11
--- /dev/null
+++ b/internal/rpc/test/config_rpc_ok_env.yaml
@@ -0,0 +1,2 @@
+rpc:
+ listen: ${RPC}
generated by cgit v1.2.3 (git 2.25.1) at 2025-01-24 08:29:34 +0000