#723 fix(redis, ws): redis driver not available

#723 fix(redis, ws): redis driver not available

codecov.yml

@@ -16,6 +16,7 @@ ignore:
   - "tests"
   - "plugins/metrics/config_test.go"
   - "plugins/websockets/storage/storage_test.go"
+  - "plugins/websockets/config.go"
   - "pkg/bst/bst_test.go"
   - "pkg/doc"
   - "pkg/pool/static_pool_test.go"

go.mod

@@ -21,7 +21,7 @@ require (
 	// SPIRAL ====
 	github.com/spiral/endure v1.0.2
 	github.com/spiral/errors v1.0.11
-	github.com/spiral/goridge/v3 v3.1.3
+	github.com/spiral/goridge/v3 v3.1.4
 	// ===========
 	github.com/stretchr/testify v1.7.0
 	github.com/tklauser/go-sysconf v0.3.6 // indirect

go.sum

@@ -388,8 +388,8 @@ github.com/spiral/endure v1.0.2/go.mod h1:/mnduq57eBKgKCwpuLgUp8Fn/c3h6JgWybG+0h
 github.com/spiral/errors v1.0.10/go.mod h1:SwMSZVdZkkJVgXNNafccqOaxWg0XPzVU/dEdUEInE0o=
 github.com/spiral/errors v1.0.11 h1:TGG+t3mNouLuRW54Ph7nHo4X3u4WhbxqEQmnIybi7Go=
 github.com/spiral/errors v1.0.11/go.mod h1:SwMSZVdZkkJVgXNNafccqOaxWg0XPzVU/dEdUEInE0o=
-github.com/spiral/goridge/v3 v3.1.3 h1:mjKLxWBmCN2ZErkA1mTkQAMbYjWQK8cb7DluQA/9wjU=
-github.com/spiral/goridge/v3 v3.1.3/go.mod h1:swcWZW7nP+KU9rgyRf6w7CfNDCWRC/vePE2+AKtoqjk=
+github.com/spiral/goridge/v3 v3.1.4 h1:5egVVTfaD1PO4MRgzU0yyog86pAh+JIOk7xhe7BtG40=
+github.com/spiral/goridge/v3 v3.1.4/go.mod h1:swcWZW7nP+KU9rgyRf6w7CfNDCWRC/vePE2+AKtoqjk=
 github.com/streadway/amqp v0.0.0-20190404075320-75d898a42a94/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/streadway/handy v0.0.0-20190108123426-d5acb3125c2a/go.mod h1:qNTQ5P5JnDBl6z3cMAg/SywNDC5ABu5ApDIw6lUbRmI=

pkg/pubsub/interface.go

@@ -44,3 +44,7 @@ type Publisher interface {
 type Reader interface {
 	Next() (*websocketsv1.Message, error)
 }
+
+type PSProvider interface {
+	PSProvide(key string) (PubSub, error)
+}

plugins/kv/drivers/boltdb/plugin.go

@@ -46,7 +46,7 @@ func (s *Plugin) Stop() error {
 	return nil
 }
 
-func (s *Plugin) Provide(key string) (kv.Storage, error) {
+func (s *Plugin) KVProvide(key string) (kv.Storage, error) {
 	const op = errors.Op("boltdb_plugin_provide")
 	st, err := NewBoltDBDriver(s.log, key, s.cfgPlugin, s.stop)
 	if err != nil {

plugins/kv/drivers/memcached/plugin.go

@@ -34,7 +34,7 @@ func (s *Plugin) Name() string {
 // Available interface implementation
 func (s *Plugin) Available() {}
 
-func (s *Plugin) Provide(key string) (kv.Storage, error) {
+func (s *Plugin) KVProvide(key string) (kv.Storage, error) {
 	const op = errors.Op("boltdb_plugin_provide")
 	st, err := NewMemcachedDriver(s.log, key, s.cfgPlugin)
 	if err != nil {

plugins/kv/drivers/memory/plugin.go

@@ -45,7 +45,7 @@ func (s *Plugin) Stop() error {
 	return nil
 }
 
-func (s *Plugin) Provide(key string) (kv.Storage, error) {
+func (s *Plugin) KVProvide(key string) (kv.Storage, error) {
 	const op = errors.Op("inmemory_plugin_provide")
 	st, err := NewInMemoryDriver(s.log, key, s.cfgPlugin, s.stop)
 	if err != nil {

plugins/kv/interface.go

@@ -37,5 +37,5 @@ type StorageDriver interface {
 // Provider provides storage based on the config
 type Provider interface {
 	// Provide provides Storage based on the config key
-	Provide(key string) (Storage, error)
+	KVProvide(key string) (Storage, error)
 }

plugins/kv/storage.go → plugins/kv/plugin.go

@@ -50,7 +50,7 @@ func (p *Plugin) Init(cfg config.Configurer, log logger.Logger) error {
 	return nil
 }
 
-func (p *Plugin) Serve() chan error {
+func (p *Plugin) Serve() chan error { //nolint:gocognit
 	errCh := make(chan error, 1)
 	const op = errors.Op("kv_plugin_serve")
 	// key - storage name in the config
@@ -100,7 +100,7 @@ func (p *Plugin) Serve() chan error {
 				continue
 			}
 
-			storage, err := p.drivers[memcached].Provide(configKey)
+			storage, err := p.drivers[memcached].KVProvide(configKey)
 			if err != nil {
 				errCh <- errors.E(op, err)
 				return errCh
@@ -115,7 +115,7 @@ func (p *Plugin) Serve() chan error {
 				continue
 			}
 
-			storage, err := p.drivers[boltdb].Provide(configKey)
+			storage, err := p.drivers[boltdb].KVProvide(configKey)
 			if err != nil {
 				errCh <- errors.E(op, err)
 				return errCh
@@ -129,7 +129,7 @@ func (p *Plugin) Serve() chan error {
 				continue
 			}
 
-			storage, err := p.drivers[memory].Provide(configKey)
+			storage, err := p.drivers[memory].KVProvide(configKey)
 			if err != nil {
 				errCh <- errors.E(op, err)
 				return errCh
@@ -143,15 +143,33 @@ func (p *Plugin) Serve() chan error {
 				continue
 			}
 
-			storage, err := p.drivers[redis].Provide(configKey)
-			if err != nil {
-				errCh <- errors.E(op, err)
-				return errCh
+			// first - try local configuration
+			switch {
+			case p.cfgPlugin.Has(configKey):
+				storage, err := p.drivers[redis].KVProvide(configKey)
+				if err != nil {
+					errCh <- errors.E(op, err)
+					return errCh
+				}
+
+				// save the storage
+				p.storages[k] = storage
+			case p.cfgPlugin.Has(redis):
+				storage, err := p.drivers[redis].KVProvide(configKey)
+				if err != nil {
+					errCh <- errors.E(op, err)
+					return errCh
+				}
+
+				// save the storage
+				p.storages[k] = storage
+				continue
+			default:
+				// otherwise - error, no local or global config
+				p.log.Warn("no global or local redis configuration provided", "key", configKey)
+				continue
 			}
 
-			// save the storage
-			p.storages[k] = storage
-
 		default:
 			p.log.Error("unknown storage", errors.E(op, errors.Errorf("unknown storage %s", v.(map[string]interface{})[driver])))
 		}

plugins/memory/plugin.go

@@ -0,0 +1,25 @@
+package memory
+
+import (
+	"github.com/spiral/roadrunner/v2/pkg/pubsub"
+	"github.com/spiral/roadrunner/v2/plugins/logger"
+)
+
+const PluginName string = "memory"
+
+type Plugin struct {
+	log logger.Logger
+}
+
+func (p *Plugin) Init(log logger.Logger) error {
+	p.log = log
+	return nil
+}
+
+func (p *Plugin) PSProvide(key string) (pubsub.PubSub, error) {
+	return NewPubSubDriver(p.log, key)
+}
+
+func (p *Plugin) Name() string {
+	return PluginName
+}

plugins/websockets/memory/inMemory.go → plugins/memory/pubsub.go

@@ -10,36 +10,36 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
-type Plugin struct {
+type PubSubDriver struct {
 	sync.RWMutex
-	log logger.Logger
-
 	// channel with the messages from the RPC
 	pushCh chan []byte
 	// user-subscribed topics
 	storage bst.Storage
+	log     logger.Logger
 }
 
-func NewInMemory(log logger.Logger) pubsub.PubSub {
-	return &Plugin{
-		log:     log,
+func NewPubSubDriver(log logger.Logger, _ string) (pubsub.PubSub, error) {
+	ps := &PubSubDriver{
 		pushCh:  make(chan []byte, 10),
 		storage: bst.NewBST(),
+		log:     log,
 	}
+	return ps, nil
 }
 
-func (p *Plugin) Publish(message []byte) error {
+func (p *PubSubDriver) Publish(message []byte) error {
 	p.pushCh <- message
 	return nil
 }
 
-func (p *Plugin) PublishAsync(message []byte) {
+func (p *PubSubDriver) PublishAsync(message []byte) {
 	go func() {
 		p.pushCh <- message
 	}()
 }
 
-func (p *Plugin) Subscribe(connectionID string, topics ...string) error {
+func (p *PubSubDriver) Subscribe(connectionID string, topics ...string) error {
 	p.Lock()
 	defer p.Unlock()
 	for i := 0; i < len(topics); i++ {
@@ -48,7 +48,7 @@ func (p *Plugin) Subscribe(connectionID string, topics ...string) error {
 	return nil
 }
 
-func (p *Plugin) Unsubscribe(connectionID string, topics ...string) error {
+func (p *PubSubDriver) Unsubscribe(connectionID string, topics ...string) error {
 	p.Lock()
 	defer p.Unlock()
 	for i := 0; i < len(topics); i++ {
@@ -57,7 +57,7 @@ func (p *Plugin) Unsubscribe(connectionID string, topics ...string) error {
 	return nil
 }
 
-func (p *Plugin) Connections(topic string, res map[string]struct{}) {
+func (p *PubSubDriver) Connections(topic string, res map[string]struct{}) {
 	p.RLock()
 	defer p.RUnlock()
 
@@ -67,7 +67,7 @@ func (p *Plugin) Connections(topic string, res map[string]struct{}) {
 	}
 }
 
-func (p *Plugin) Next() (*websocketsv1.Message, error) {
+func (p *PubSubDriver) Next() (*websocketsv1.Message, error) {
 	msg := <-p.pushCh
 	if msg == nil {
 		return nil, nil