From 75ab1e16c64cfd0a6424fe4c546fdbc5e1b992dd Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Mon, 14 Jun 2021 16:39:02 +0300 Subject: - Rework redis with ws plugins Signed-off-by: Valery Piashchynski --- plugins/kv/drivers/boltdb/plugin.go | 2 +- plugins/kv/drivers/memcached/plugin.go | 2 +- plugins/kv/drivers/memory/plugin.go | 2 +- plugins/kv/drivers/redis/config.go | 34 ----- plugins/kv/drivers/redis/driver.go | 242 --------------------------------- plugins/kv/drivers/redis/plugin.go | 51 ------- plugins/kv/interface.go | 2 +- plugins/kv/plugin.go | 208 ++++++++++++++++++++++++++++ plugins/kv/storage.go | 190 -------------------------- plugins/memory/plugin.go | 25 ++++ plugins/memory/pubsub.go | 95 +++++++++++++ plugins/redis/clients.go | 84 ++++++++++++ plugins/redis/interface.go | 7 +- plugins/redis/kv.go | 242 +++++++++++++++++++++++++++++++++ plugins/redis/plugin.go | 181 +++--------------------- plugins/redis/pubsub.go | 189 +++++++++++++++++++++++++ plugins/websockets/config.go | 43 ++++++ plugins/websockets/memory/inMemory.go | 95 ------------- plugins/websockets/plugin.go | 77 +++++++++-- 19 files changed, 983 insertions(+), 788 deletions(-) delete mode 100644 plugins/kv/drivers/redis/config.go delete mode 100644 plugins/kv/drivers/redis/driver.go delete mode 100644 plugins/kv/drivers/redis/plugin.go create mode 100644 plugins/kv/plugin.go delete mode 100644 plugins/kv/storage.go create mode 100644 plugins/memory/plugin.go create mode 100644 plugins/memory/pubsub.go create mode 100644 plugins/redis/clients.go create mode 100644 plugins/redis/kv.go create mode 100644 plugins/redis/pubsub.go delete mode 100644 plugins/websockets/memory/inMemory.go (limited to 'plugins') diff --git a/plugins/kv/drivers/boltdb/plugin.go b/plugins/kv/drivers/boltdb/plugin.go index 9b4cf9f7..28e2a89c 100644 --- a/plugins/kv/drivers/boltdb/plugin.go +++ b/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 { diff --git a/plugins/kv/drivers/memcached/plugin.go b/plugins/kv/drivers/memcached/plugin.go index 3997e0d4..936b2047 100644 --- a/plugins/kv/drivers/memcached/plugin.go +++ b/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 { diff --git a/plugins/kv/drivers/memory/plugin.go b/plugins/kv/drivers/memory/plugin.go index 2be7caae..da81017e 100644 --- a/plugins/kv/drivers/memory/plugin.go +++ b/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 { diff --git a/plugins/kv/drivers/redis/config.go b/plugins/kv/drivers/redis/config.go deleted file mode 100644 index 41348236..00000000 --- a/plugins/kv/drivers/redis/config.go +++ /dev/null @@ -1,34 +0,0 @@ -package redis - -import "time" - -type Config struct { - Addrs []string `mapstructure:"addrs"` - DB int `mapstructure:"db"` - Username string `mapstructure:"username"` - Password string `mapstructure:"password"` - MasterName string `mapstructure:"master_name"` - SentinelPassword string `mapstructure:"sentinel_password"` - RouteByLatency bool `mapstructure:"route_by_latency"` - RouteRandomly bool `mapstructure:"route_randomly"` - MaxRetries int `mapstructure:"max_retries"` - DialTimeout time.Duration `mapstructure:"dial_timeout"` - MinRetryBackoff time.Duration `mapstructure:"min_retry_backoff"` - MaxRetryBackoff time.Duration `mapstructure:"max_retry_backoff"` - PoolSize int `mapstructure:"pool_size"` - MinIdleConns int `mapstructure:"min_idle_conns"` - MaxConnAge time.Duration `mapstructure:"max_conn_age"` - ReadTimeout time.Duration `mapstructure:"read_timeout"` - WriteTimeout time.Duration `mapstructure:"write_timeout"` - PoolTimeout time.Duration `mapstructure:"pool_timeout"` - IdleTimeout time.Duration `mapstructure:"idle_timeout"` - IdleCheckFreq time.Duration `mapstructure:"idle_check_freq"` - ReadOnly bool `mapstructure:"read_only"` -} - -// InitDefaults initializing fill config with default values -func (s *Config) InitDefaults() { - if s.Addrs == nil { - s.Addrs = []string{"localhost:6379"} // default addr is pointing to local storage - } -} diff --git a/plugins/kv/drivers/redis/driver.go b/plugins/kv/drivers/redis/driver.go deleted file mode 100644 index 66cb8384..00000000 --- a/plugins/kv/drivers/redis/driver.go +++ /dev/null @@ -1,242 +0,0 @@ -package redis - -import ( - "context" - "strings" - "time" - - "github.com/go-redis/redis/v8" - "github.com/spiral/errors" - kvv1 "github.com/spiral/roadrunner/v2/pkg/proto/kv/v1beta" - "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/kv" - "github.com/spiral/roadrunner/v2/plugins/logger" - "github.com/spiral/roadrunner/v2/utils" -) - -type Driver struct { - universalClient redis.UniversalClient - log logger.Logger - cfg *Config -} - -func NewRedisDriver(log logger.Logger, key string, cfgPlugin config.Configurer) (kv.Storage, error) { - const op = errors.Op("new_boltdb_driver") - - d := &Driver{ - log: log, - } - - // will be different for every connected driver - err := cfgPlugin.UnmarshalKey(key, &d.cfg) - if err != nil { - return nil, errors.E(op, err) - } - - d.cfg.InitDefaults() - d.log = log - - d.universalClient = redis.NewUniversalClient(&redis.UniversalOptions{ - Addrs: d.cfg.Addrs, - DB: d.cfg.DB, - Username: d.cfg.Username, - Password: d.cfg.Password, - SentinelPassword: d.cfg.SentinelPassword, - MaxRetries: d.cfg.MaxRetries, - MinRetryBackoff: d.cfg.MaxRetryBackoff, - MaxRetryBackoff: d.cfg.MaxRetryBackoff, - DialTimeout: d.cfg.DialTimeout, - ReadTimeout: d.cfg.ReadTimeout, - WriteTimeout: d.cfg.WriteTimeout, - PoolSize: d.cfg.PoolSize, - MinIdleConns: d.cfg.MinIdleConns, - MaxConnAge: d.cfg.MaxConnAge, - PoolTimeout: d.cfg.PoolTimeout, - IdleTimeout: d.cfg.IdleTimeout, - IdleCheckFrequency: d.cfg.IdleCheckFreq, - ReadOnly: d.cfg.ReadOnly, - RouteByLatency: d.cfg.RouteByLatency, - RouteRandomly: d.cfg.RouteRandomly, - MasterName: d.cfg.MasterName, - }) - - return d, nil -} - -// Has checks if value exists. -func (d *Driver) Has(keys ...string) (map[string]bool, error) { - const op = errors.Op("redis_driver_has") - if keys == nil { - return nil, errors.E(op, errors.NoKeys) - } - - m := make(map[string]bool, len(keys)) - for _, key := range keys { - keyTrimmed := strings.TrimSpace(key) - if keyTrimmed == "" { - return nil, errors.E(op, errors.EmptyKey) - } - - exist, err := d.universalClient.Exists(context.Background(), key).Result() - if err != nil { - return nil, err - } - if exist == 1 { - m[key] = true - } - } - return m, nil -} - -// Get loads key content into slice. -func (d *Driver) Get(key string) ([]byte, error) { - const op = errors.Op("redis_driver_get") - // to get cases like " " - keyTrimmed := strings.TrimSpace(key) - if keyTrimmed == "" { - return nil, errors.E(op, errors.EmptyKey) - } - return d.universalClient.Get(context.Background(), key).Bytes() -} - -// MGet loads content of multiple values (some values might be skipped). -// https://redis.io/commands/mget -// Returns slice with the interfaces with values -func (d *Driver) MGet(keys ...string) (map[string][]byte, error) { - const op = errors.Op("redis_driver_mget") - if keys == nil { - return nil, errors.E(op, errors.NoKeys) - } - - // should not be empty keys - for _, key := range keys { - keyTrimmed := strings.TrimSpace(key) - if keyTrimmed == "" { - return nil, errors.E(op, errors.EmptyKey) - } - } - - m := make(map[string][]byte, len(keys)) - - for _, k := range keys { - cmd := d.universalClient.Get(context.Background(), k) - if cmd.Err() != nil { - if cmd.Err() == redis.Nil { - continue - } - return nil, errors.E(op, cmd.Err()) - } - - m[k] = utils.AsBytes(cmd.Val()) - } - - return m, nil -} - -// Set sets value with the TTL in seconds -// https://redis.io/commands/set -// Redis `SET key value [expiration]` command. -// -// Use expiration for `SETEX`-like behavior. -// Zero expiration means the key has no expiration time. -func (d *Driver) Set(items ...*kvv1.Item) error { - const op = errors.Op("redis_driver_set") - if items == nil { - return errors.E(op, errors.NoKeys) - } - now := time.Now() - for _, item := range items { - if item == nil { - return errors.E(op, errors.EmptyKey) - } - - if item.Timeout == "" { - err := d.universalClient.Set(context.Background(), item.Key, item.Value, 0).Err() - if err != nil { - return err - } - } else { - t, err := time.Parse(time.RFC3339, item.Timeout) - if err != nil { - return err - } - err = d.universalClient.Set(context.Background(), item.Key, item.Value, t.Sub(now)).Err() - if err != nil { - return err - } - } - } - return nil -} - -// Delete one or multiple keys. -func (d *Driver) Delete(keys ...string) error { - const op = errors.Op("redis_driver_delete") - if keys == nil { - return errors.E(op, errors.NoKeys) - } - - // should not be empty keys - for _, key := range keys { - keyTrimmed := strings.TrimSpace(key) - if keyTrimmed == "" { - return errors.E(op, errors.EmptyKey) - } - } - return d.universalClient.Del(context.Background(), keys...).Err() -} - -// MExpire https://redis.io/commands/expire -// timeout in RFC3339 -func (d *Driver) MExpire(items ...*kvv1.Item) error { - const op = errors.Op("redis_driver_mexpire") - now := time.Now() - for _, item := range items { - if item == nil { - continue - } - if item.Timeout == "" || strings.TrimSpace(item.Key) == "" { - return errors.E(op, errors.Str("should set timeout and at least one key")) - } - - t, err := time.Parse(time.RFC3339, item.Timeout) - if err != nil { - return err - } - - // t guessed to be in future - // for Redis we use t.Sub, it will result in seconds, like 4.2s - d.universalClient.Expire(context.Background(), item.Key, t.Sub(now)) - } - - return nil -} - -// TTL https://redis.io/commands/ttl -// return time in seconds (float64) for a given keys -func (d *Driver) TTL(keys ...string) (map[string]string, error) { - const op = errors.Op("redis_driver_ttl") - if keys == nil { - return nil, errors.E(op, errors.NoKeys) - } - - // should not be empty keys - for _, key := range keys { - keyTrimmed := strings.TrimSpace(key) - if keyTrimmed == "" { - return nil, errors.E(op, errors.EmptyKey) - } - } - - m := make(map[string]string, len(keys)) - - for _, key := range keys { - duration, err := d.universalClient.TTL(context.Background(), key).Result() - if err != nil { - return nil, err - } - - m[key] = duration.String() - } - return m, nil -} diff --git a/plugins/kv/drivers/redis/plugin.go b/plugins/kv/drivers/redis/plugin.go deleted file mode 100644 index 3694c5a7..00000000 --- a/plugins/kv/drivers/redis/plugin.go +++ /dev/null @@ -1,51 +0,0 @@ -package redis - -import ( - "github.com/spiral/errors" - "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/kv" - "github.com/spiral/roadrunner/v2/plugins/logger" -) - -const PluginName = "redis" - -// Plugin BoltDB K/V storage. -type Plugin struct { - cfgPlugin config.Configurer - // logger - log logger.Logger -} - -func (s *Plugin) Init(log logger.Logger, cfg config.Configurer) error { - if !cfg.Has(kv.PluginName) { - return errors.E(errors.Disabled) - } - - s.log = log - s.cfgPlugin = cfg - return nil -} - -// Serve is noop here -func (s *Plugin) Serve() chan error { - return make(chan error) -} - -func (s *Plugin) Stop() error { - return nil -} - -func (s *Plugin) Provide(key string) (kv.Storage, error) { - const op = errors.Op("redis_plugin_provide") - st, err := NewRedisDriver(s.log, key, s.cfgPlugin) - if err != nil { - return nil, errors.E(op, err) - } - - return st, nil -} - -// Name returns plugin name -func (s *Plugin) Name() string { - return PluginName -} diff --git a/plugins/kv/interface.go b/plugins/kv/interface.go index 744c6b51..5aedd5c3 100644 --- a/plugins/kv/interface.go +++ b/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) } diff --git a/plugins/kv/plugin.go b/plugins/kv/plugin.go new file mode 100644 index 00000000..efe92252 --- /dev/null +++ b/plugins/kv/plugin.go @@ -0,0 +1,208 @@ +package kv + +import ( + "fmt" + + endure "github.com/spiral/endure/pkg/container" + "github.com/spiral/errors" + "github.com/spiral/roadrunner/v2/plugins/config" + "github.com/spiral/roadrunner/v2/plugins/logger" +) + +const PluginName string = "kv" + +const ( + // driver is the mandatory field which should present in every storage + driver string = "driver" + + memcached string = "memcached" + boltdb string = "boltdb" + redis string = "redis" + memory string = "memory" +) + +// Plugin for the unified storage +type Plugin struct { + log logger.Logger + // drivers contains general storage drivers, such as boltdb, memory, memcached, redis. + drivers map[string]StorageDriver + // storages contains user-defined storages, such as boltdb-north, memcached-us and so on. + storages map[string]Storage + // KV configuration + cfg Config + cfgPlugin config.Configurer +} + +func (p *Plugin) Init(cfg config.Configurer, log logger.Logger) error { + const op = errors.Op("kv_plugin_init") + if !cfg.Has(PluginName) { + return errors.E(errors.Disabled) + } + + err := cfg.UnmarshalKey(PluginName, &p.cfg.Data) + if err != nil { + return errors.E(op, err) + } + p.drivers = make(map[string]StorageDriver, 5) + p.storages = make(map[string]Storage, 5) + p.log = log + p.cfgPlugin = cfg + return nil +} + +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 + // value - storage + /* + For example we can have here 2 storages (but they are not pre-configured) + for the boltdb and memcached + We should provide here the actual configs for the all requested storages + kv: + boltdb-south: + driver: boltdb + dir: "tests/rr-bolt" + file: "rr.db" + bucket: "rr" + permissions: 777 + ttl: 40s + + boltdb-north: + driver: boltdb + dir: "tests/rr-bolt" + file: "rr.db" + bucket: "rr" + permissions: 777 + ttl: 40s + + memcached: + driver: memcached + addr: [ "localhost:11211" ] + + + For this config we should have 3 drivers: memory, boltdb and memcached but 4 KVs: default, boltdb-south, boltdb-north and memcached + when user requests for example boltdb-south, we should provide that particular preconfigured storage + */ + for k, v := range p.cfg.Data { + if _, ok := v.(map[string]interface{})[driver]; !ok { + errCh <- errors.E(op, errors.Errorf("could not find mandatory driver field in the %s storage", k)) + return errCh + } + + // config key for the particular sub-driver kv.memcached + configKey := fmt.Sprintf("%s.%s", PluginName, k) + // at this point we know, that driver field present in the configuration + switch v.(map[string]interface{})[driver] { + case memcached: + if _, ok := p.drivers[memcached]; !ok { + p.log.Warn("no memcached drivers registered", "registered", p.drivers) + continue + } + + storage, err := p.drivers[memcached].KVProvide(configKey) + if err != nil { + errCh <- errors.E(op, err) + return errCh + } + + // save the storage + p.storages[k] = storage + + case boltdb: + if _, ok := p.drivers[boltdb]; !ok { + p.log.Warn("no boltdb drivers registered", "registered", p.drivers) + continue + } + + storage, err := p.drivers[boltdb].KVProvide(configKey) + if err != nil { + errCh <- errors.E(op, err) + return errCh + } + + // save the storage + p.storages[k] = storage + case memory: + if _, ok := p.drivers[memory]; !ok { + p.log.Warn("no in-memory drivers registered", "registered", p.drivers) + continue + } + + storage, err := p.drivers[memory].KVProvide(configKey) + if err != nil { + errCh <- errors.E(op, err) + return errCh + } + + // save the storage + p.storages[k] = storage + case redis: + if _, ok := p.drivers[redis]; !ok { + p.log.Warn("no redis drivers registered", "registered", p.drivers) + continue + } + + // 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 + } + + default: + p.log.Error("unknown storage", errors.E(op, errors.Errorf("unknown storage %s", v.(map[string]interface{})[driver]))) + } + } + + return errCh +} + +func (p *Plugin) Stop() error { + return nil +} + +// Collects will get all plugins which implement Storage interface +func (p *Plugin) Collects() []interface{} { + return []interface{}{ + p.GetAllStorageDrivers, + } +} + +func (p *Plugin) GetAllStorageDrivers(name endure.Named, storage StorageDriver) { + // save the storage driver + p.drivers[name.Name()] = storage +} + +// RPC returns associated rpc service. +func (p *Plugin) RPC() interface{} { + return &rpc{srv: p, log: p.log, storages: p.storages} +} + +func (p *Plugin) Name() string { + return PluginName +} + +// Available interface implementation +func (p *Plugin) Available() { +} diff --git a/plugins/kv/storage.go b/plugins/kv/storage.go deleted file mode 100644 index 9a609735..00000000 --- a/plugins/kv/storage.go +++ /dev/null @@ -1,190 +0,0 @@ -package kv - -import ( - "fmt" - - endure "github.com/spiral/endure/pkg/container" - "github.com/spiral/errors" - "github.com/spiral/roadrunner/v2/plugins/config" - "github.com/spiral/roadrunner/v2/plugins/logger" -) - -const PluginName string = "kv" - -const ( - // driver is the mandatory field which should present in every storage - driver string = "driver" - - memcached string = "memcached" - boltdb string = "boltdb" - redis string = "redis" - memory string = "memory" -) - -// Plugin for the unified storage -type Plugin struct { - log logger.Logger - // drivers contains general storage drivers, such as boltdb, memory, memcached, redis. - drivers map[string]StorageDriver - // storages contains user-defined storages, such as boltdb-north, memcached-us and so on. - storages map[string]Storage - // KV configuration - cfg Config - cfgPlugin config.Configurer -} - -func (p *Plugin) Init(cfg config.Configurer, log logger.Logger) error { - const op = errors.Op("kv_plugin_init") - if !cfg.Has(PluginName) { - return errors.E(errors.Disabled) - } - - err := cfg.UnmarshalKey(PluginName, &p.cfg.Data) - if err != nil { - return errors.E(op, err) - } - p.drivers = make(map[string]StorageDriver, 5) - p.storages = make(map[string]Storage, 5) - p.log = log - p.cfgPlugin = cfg - return nil -} - -func (p *Plugin) Serve() chan error { - errCh := make(chan error, 1) - const op = errors.Op("kv_plugin_serve") - // key - storage name in the config - // value - storage - /* - For example we can have here 2 storages (but they are not pre-configured) - for the boltdb and memcached - We should provide here the actual configs for the all requested storages - kv: - boltdb-south: - driver: boltdb - dir: "tests/rr-bolt" - file: "rr.db" - bucket: "rr" - permissions: 777 - ttl: 40s - - boltdb-north: - driver: boltdb - dir: "tests/rr-bolt" - file: "rr.db" - bucket: "rr" - permissions: 777 - ttl: 40s - - memcached: - driver: memcached - addr: [ "localhost:11211" ] - - - For this config we should have 3 drivers: memory, boltdb and memcached but 4 KVs: default, boltdb-south, boltdb-north and memcached - when user requests for example boltdb-south, we should provide that particular preconfigured storage - */ - for k, v := range p.cfg.Data { - if _, ok := v.(map[string]interface{})[driver]; !ok { - errCh <- errors.E(op, errors.Errorf("could not find mandatory driver field in the %s storage", k)) - return errCh - } - - // config key for the particular sub-driver kv.memcached - configKey := fmt.Sprintf("%s.%s", PluginName, k) - // at this point we know, that driver field present in the configuration - switch v.(map[string]interface{})[driver] { - case memcached: - if _, ok := p.drivers[memcached]; !ok { - p.log.Warn("no memcached drivers registered", "registered", p.drivers) - continue - } - - storage, err := p.drivers[memcached].Provide(configKey) - if err != nil { - errCh <- errors.E(op, err) - return errCh - } - - // save the storage - p.storages[k] = storage - - case boltdb: - if _, ok := p.drivers[boltdb]; !ok { - p.log.Warn("no boltdb drivers registered", "registered", p.drivers) - continue - } - - storage, err := p.drivers[boltdb].Provide(configKey) - if err != nil { - errCh <- errors.E(op, err) - return errCh - } - - // save the storage - p.storages[k] = storage - case memory: - if _, ok := p.drivers[memory]; !ok { - p.log.Warn("no in-memory drivers registered", "registered", p.drivers) - continue - } - - storage, err := p.drivers[memory].Provide(configKey) - if err != nil { - errCh <- errors.E(op, err) - return errCh - } - - // save the storage - p.storages[k] = storage - case redis: - if _, ok := p.drivers[redis]; !ok { - p.log.Warn("no redis drivers registered", "registered", p.drivers) - continue - } - - storage, err := p.drivers[redis].Provide(configKey) - if err != nil { - errCh <- errors.E(op, err) - return errCh - } - - // 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]))) - } - } - - return errCh -} - -func (p *Plugin) Stop() error { - return nil -} - -// Collects will get all plugins which implement Storage interface -func (p *Plugin) Collects() []interface{} { - return []interface{}{ - p.GetAllStorageDrivers, - } -} - -func (p *Plugin) GetAllStorageDrivers(name endure.Named, storage StorageDriver) { - // save the storage driver - p.drivers[name.Name()] = storage -} - -// RPC returns associated rpc service. -func (p *Plugin) RPC() interface{} { - return &rpc{srv: p, log: p.log, storages: p.storages} -} - -func (p *Plugin) Name() string { - return PluginName -} - -// Available interface implementation -func (p *Plugin) Available() { -} diff --git a/plugins/memory/plugin.go b/plugins/memory/plugin.go new file mode 100644 index 00000000..650f0b4b --- /dev/null +++ b/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 +} diff --git a/plugins/memory/pubsub.go b/plugins/memory/pubsub.go new file mode 100644 index 00000000..75cd9d24 --- /dev/null +++ b/plugins/memory/pubsub.go @@ -0,0 +1,95 @@ +package memory + +import ( + "sync" + + "github.com/spiral/roadrunner/v2/pkg/bst" + websocketsv1 "github.com/spiral/roadrunner/v2/pkg/proto/websockets/v1beta" + "github.com/spiral/roadrunner/v2/pkg/pubsub" + "github.com/spiral/roadrunner/v2/plugins/logger" + "google.golang.org/protobuf/proto" +) + +type PubSubDriver struct { + sync.RWMutex + // channel with the messages from the RPC + pushCh chan []byte + // user-subscribed topics + storage bst.Storage + log logger.Logger +} + +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 *PubSubDriver) Publish(message []byte) error { + p.pushCh <- message + return nil +} + +func (p *PubSubDriver) PublishAsync(message []byte) { + go func() { + p.pushCh <- message + }() +} + +func (p *PubSubDriver) Subscribe(connectionID string, topics ...string) error { + p.Lock() + defer p.Unlock() + for i := 0; i < len(topics); i++ { + p.storage.Insert(connectionID, topics[i]) + } + return nil +} + +func (p *PubSubDriver) Unsubscribe(connectionID string, topics ...string) error { + p.Lock() + defer p.Unlock() + for i := 0; i < len(topics); i++ { + p.storage.Remove(connectionID, topics[i]) + } + return nil +} + +func (p *PubSubDriver) Connections(topic string, res map[string]struct{}) { + p.RLock() + defer p.RUnlock() + + ret := p.storage.Get(topic) + for rr := range ret { + res[rr] = struct{}{} + } +} + +func (p *PubSubDriver) Next() (*websocketsv1.Message, error) { + msg := <-p.pushCh + if msg == nil { + return nil, nil + } + + p.RLock() + defer p.RUnlock() + + m := &websocketsv1.Message{} + err := proto.Unmarshal(msg, m) + if err != nil { + return nil, err + } + + // push only messages, which are subscribed + // TODO better??? + for i := 0; i < len(m.GetTopics()); i++ { + // if we have active subscribers - send a message to a topic + // or send nil instead + if ok := p.storage.Contains(m.GetTopics()[i]); ok { + return m, nil + } + } + return nil, nil +} diff --git a/plugins/redis/clients.go b/plugins/redis/clients.go new file mode 100644 index 00000000..d0a184d2 --- /dev/null +++ b/plugins/redis/clients.go @@ -0,0 +1,84 @@ +package redis + +import ( + "github.com/go-redis/redis/v8" + "github.com/spiral/errors" +) + +// RedisClient return a client based on the provided section key +// key sample: kv.some-section.redis +// kv.redis +// redis (root) +func (p *Plugin) RedisClient(key string) (redis.UniversalClient, error) { + const op = errors.Op("redis_get_client") + + if !p.cfgPlugin.Has(key) { + return nil, errors.E(op, errors.Errorf("no such section: %s", key)) + } + + cfg := &Config{} + + err := p.cfgPlugin.UnmarshalKey(key, cfg) + if err != nil { + return nil, errors.E(op, err) + } + + cfg.InitDefaults() + + uc := redis.NewUniversalClient(&redis.UniversalOptions{ + Addrs: cfg.Addrs, + DB: cfg.DB, + Username: cfg.Username, + Password: cfg.Password, + SentinelPassword: cfg.SentinelPassword, + MaxRetries: cfg.MaxRetries, + MinRetryBackoff: cfg.MaxRetryBackoff, + MaxRetryBackoff: cfg.MaxRetryBackoff, + DialTimeout: cfg.DialTimeout, + ReadTimeout: cfg.ReadTimeout, + WriteTimeout: cfg.WriteTimeout, + PoolSize: cfg.PoolSize, + MinIdleConns: cfg.MinIdleConns, + MaxConnAge: cfg.MaxConnAge, + PoolTimeout: cfg.PoolTimeout, + IdleTimeout: cfg.IdleTimeout, + IdleCheckFrequency: cfg.IdleCheckFreq, + ReadOnly: cfg.ReadOnly, + RouteByLatency: cfg.RouteByLatency, + RouteRandomly: cfg.RouteRandomly, + MasterName: cfg.MasterName, + }) + + return uc, nil +} + +func (p *Plugin) DefaultClient() redis.UniversalClient { + cfg := &Config{} + cfg.InitDefaults() + + uc := redis.NewUniversalClient(&redis.UniversalOptions{ + Addrs: cfg.Addrs, + DB: cfg.DB, + Username: cfg.Username, + Password: cfg.Password, + SentinelPassword: cfg.SentinelPassword, + MaxRetries: cfg.MaxRetries, + MinRetryBackoff: cfg.MaxRetryBackoff, + MaxRetryBackoff: cfg.MaxRetryBackoff, + DialTimeout: cfg.DialTimeout, + ReadTimeout: cfg.ReadTimeout, + WriteTimeout: cfg.WriteTimeout, + PoolSize: cfg.PoolSize, + MinIdleConns: cfg.MinIdleConns, + MaxConnAge: cfg.MaxConnAge, + PoolTimeout: cfg.PoolTimeout, + IdleTimeout: cfg.IdleTimeout, + IdleCheckFrequency: cfg.IdleCheckFreq, + ReadOnly: cfg.ReadOnly, + RouteByLatency: cfg.RouteByLatency, + RouteRandomly: cfg.RouteRandomly, + MasterName: cfg.MasterName, + }) + + return uc +} diff --git a/plugins/redis/interface.go b/plugins/redis/interface.go index c0be6137..189b0002 100644 --- a/plugins/redis/interface.go +++ b/plugins/redis/interface.go @@ -4,6 +4,9 @@ import "github.com/go-redis/redis/v8" // Redis in the redis KV plugin interface type Redis interface { - // GetClient provides universal redis client - GetClient() redis.UniversalClient + // RedisClient provides universal redis client + RedisClient(key string) (redis.UniversalClient, error) + + // DefaultClient provide default redis client based on redis defaults + DefaultClient() redis.UniversalClient } diff --git a/plugins/redis/kv.go b/plugins/redis/kv.go new file mode 100644 index 00000000..66cb8384 --- /dev/null +++ b/plugins/redis/kv.go @@ -0,0 +1,242 @@ +package redis + +import ( + "context" + "strings" + "time" + + "github.com/go-redis/redis/v8" + "github.com/spiral/errors" + kvv1 "github.com/spiral/roadrunner/v2/pkg/proto/kv/v1beta" + "github.com/spiral/roadrunner/v2/plugins/config" + "github.com/spiral/roadrunner/v2/plugins/kv" + "github.com/spiral/roadrunner/v2/plugins/logger" + "github.com/spiral/roadrunner/v2/utils" +) + +type Driver struct { + universalClient redis.UniversalClient + log logger.Logger + cfg *Config +} + +func NewRedisDriver(log logger.Logger, key string, cfgPlugin config.Configurer) (kv.Storage, error) { + const op = errors.Op("new_boltdb_driver") + + d := &Driver{ + log: log, + } + + // will be different for every connected driver + err := cfgPlugin.UnmarshalKey(key, &d.cfg) + if err != nil { + return nil, errors.E(op, err) + } + + d.cfg.InitDefaults() + d.log = log + + d.universalClient = redis.NewUniversalClient(&redis.UniversalOptions{ + Addrs: d.cfg.Addrs, + DB: d.cfg.DB, + Username: d.cfg.Username, + Password: d.cfg.Password, + SentinelPassword: d.cfg.SentinelPassword, + MaxRetries: d.cfg.MaxRetries, + MinRetryBackoff: d.cfg.MaxRetryBackoff, + MaxRetryBackoff: d.cfg.MaxRetryBackoff, + DialTimeout: d.cfg.DialTimeout, + ReadTimeout: d.cfg.ReadTimeout, + WriteTimeout: d.cfg.WriteTimeout, + PoolSize: d.cfg.PoolSize, + MinIdleConns: d.cfg.MinIdleConns, + MaxConnAge: d.cfg.MaxConnAge, + PoolTimeout: d.cfg.PoolTimeout, + IdleTimeout: d.cfg.IdleTimeout, + IdleCheckFrequency: d.cfg.IdleCheckFreq, + ReadOnly: d.cfg.ReadOnly, + RouteByLatency: d.cfg.RouteByLatency, + RouteRandomly: d.cfg.RouteRandomly, + MasterName: d.cfg.MasterName, + }) + + return d, nil +} + +// Has checks if value exists. +func (d *Driver) Has(keys ...string) (map[string]bool, error) { + const op = errors.Op("redis_driver_has") + if keys == nil { + return nil, errors.E(op, errors.NoKeys) + } + + m := make(map[string]bool, len(keys)) + for _, key := range keys { + keyTrimmed := strings.TrimSpace(key) + if keyTrimmed == "" { + return nil, errors.E(op, errors.EmptyKey) + } + + exist, err := d.universalClient.Exists(context.Background(), key).Result() + if err != nil { + return nil, err + } + if exist == 1 { + m[key] = true + } + } + return m, nil +} + +// Get loads key content into slice. +func (d *Driver) Get(key string) ([]byte, error) { + const op = errors.Op("redis_driver_get") + // to get cases like " " + keyTrimmed := strings.TrimSpace(key) + if keyTrimmed == "" { + return nil, errors.E(op, errors.EmptyKey) + } + return d.universalClient.Get(context.Background(), key).Bytes() +} + +// MGet loads content of multiple values (some values might be skipped). +// https://redis.io/commands/mget +// Returns slice with the interfaces with values +func (d *Driver) MGet(keys ...string) (map[string][]byte, error) { + const op = errors.Op("redis_driver_mget") + if keys == nil { + return nil, errors.E(op, errors.NoKeys) + } + + // should not be empty keys + for _, key := range keys { + keyTrimmed := strings.TrimSpace(key) + if keyTrimmed == "" { + return nil, errors.E(op, errors.EmptyKey) + } + } + + m := make(map[string][]byte, len(keys)) + + for _, k := range keys { + cmd := d.universalClient.Get(context.Background(), k) + if cmd.Err() != nil { + if cmd.Err() == redis.Nil { + continue + } + return nil, errors.E(op, cmd.Err()) + } + + m[k] = utils.AsBytes(cmd.Val()) + } + + return m, nil +} + +// Set sets value with the TTL in seconds +// https://redis.io/commands/set +// Redis `SET key value [expiration]` command. +// +// Use expiration for `SETEX`-like behavior. +// Zero expiration means the key has no expiration time. +func (d *Driver) Set(items ...*kvv1.Item) error { + const op = errors.Op("redis_driver_set") + if items == nil { + return errors.E(op, errors.NoKeys) + } + now := time.Now() + for _, item := range items { + if item == nil { + return errors.E(op, errors.EmptyKey) + } + + if item.Timeout == "" { + err := d.universalClient.Set(context.Background(), item.Key, item.Value, 0).Err() + if err != nil { + return err + } + } else { + t, err := time.Parse(time.RFC3339, item.Timeout) + if err != nil { + return err + } + err = d.universalClient.Set(context.Background(), item.Key, item.Value, t.Sub(now)).Err() + if err != nil { + return err + } + } + } + return nil +} + +// Delete one or multiple keys. +func (d *Driver) Delete(keys ...string) error { + const op = errors.Op("redis_driver_delete") + if keys == nil { + return errors.E(op, errors.NoKeys) + } + + // should not be empty keys + for _, key := range keys { + keyTrimmed := strings.TrimSpace(key) + if keyTrimmed == "" { + return errors.E(op, errors.EmptyKey) + } + } + return d.universalClient.Del(context.Background(), keys...).Err() +} + +// MExpire https://redis.io/commands/expire +// timeout in RFC3339 +func (d *Driver) MExpire(items ...*kvv1.Item) error { + const op = errors.Op("redis_driver_mexpire") + now := time.Now() + for _, item := range items { + if item == nil { + continue + } + if item.Timeout == "" || strings.TrimSpace(item.Key) == "" { + return errors.E(op, errors.Str("should set timeout and at least one key")) + } + + t, err := time.Parse(time.RFC3339, item.Timeout) + if err != nil { + return err + } + + // t guessed to be in future + // for Redis we use t.Sub, it will result in seconds, like 4.2s + d.universalClient.Expire(context.Background(), item.Key, t.Sub(now)) + } + + return nil +} + +// TTL https://redis.io/commands/ttl +// return time in seconds (float64) for a given keys +func (d *Driver) TTL(keys ...string) (map[string]string, error) { + const op = errors.Op("redis_driver_ttl") + if keys == nil { + return nil, errors.E(op, errors.NoKeys) + } + + // should not be empty keys + for _, key := range keys { + keyTrimmed := strings.TrimSpace(key) + if keyTrimmed == "" { + return nil, errors.E(op, errors.EmptyKey) + } + } + + m := make(map[string]string, len(keys)) + + for _, key := range keys { + duration, err := d.universalClient.TTL(context.Background(), key).Result() + if err != nil { + return nil, err + } + + m[key] = duration.String() + } + return m, nil +} diff --git a/plugins/redis/plugin.go b/plugins/redis/plugin.go index 47ffeb39..24c21b55 100644 --- a/plugins/redis/plugin.go +++ b/plugins/redis/plugin.go @@ -1,15 +1,14 @@ package redis import ( - "context" "sync" "github.com/go-redis/redis/v8" "github.com/spiral/errors" - websocketsv1 "github.com/spiral/roadrunner/v2/pkg/proto/websockets/v1beta" + "github.com/spiral/roadrunner/v2/pkg/pubsub" "github.com/spiral/roadrunner/v2/plugins/config" + "github.com/spiral/roadrunner/v2/plugins/kv" "github.com/spiral/roadrunner/v2/plugins/logger" - "google.golang.org/protobuf/proto" ) const PluginName = "redis" @@ -17,80 +16,37 @@ const PluginName = "redis" type Plugin struct { sync.RWMutex // config for RR integration - cfg *Config + cfgPlugin config.Configurer // logger log logger.Logger // redis universal client universalClient redis.UniversalClient // fanIn implementation used to deliver messages from all channels to the single websocket point - fanin *FanIn -} - -func (p *Plugin) GetClient() redis.UniversalClient { - return p.universalClient + stopCh chan struct{} } func (p *Plugin) Init(cfg config.Configurer, log logger.Logger) error { - const op = errors.Op("redis_plugin_init") - - if !cfg.Has(PluginName) { - return errors.E(op, errors.Disabled) - } - - err := cfg.UnmarshalKey(PluginName, &p.cfg) - if err != nil { - return errors.E(op, errors.Disabled, err) - } - - p.cfg.InitDefaults() p.log = log - - p.universalClient = redis.NewUniversalClient(&redis.UniversalOptions{ - Addrs: p.cfg.Addrs, - DB: p.cfg.DB, - Username: p.cfg.Username, - Password: p.cfg.Password, - SentinelPassword: p.cfg.SentinelPassword, - MaxRetries: p.cfg.MaxRetries, - MinRetryBackoff: p.cfg.MaxRetryBackoff, - MaxRetryBackoff: p.cfg.MaxRetryBackoff, - DialTimeout: p.cfg.DialTimeout, - ReadTimeout: p.cfg.ReadTimeout, - WriteTimeout: p.cfg.WriteTimeout, - PoolSize: p.cfg.PoolSize, - MinIdleConns: p.cfg.MinIdleConns, - MaxConnAge: p.cfg.MaxConnAge, - PoolTimeout: p.cfg.PoolTimeout, - IdleTimeout: p.cfg.IdleTimeout, - IdleCheckFrequency: p.cfg.IdleCheckFreq, - ReadOnly: p.cfg.ReadOnly, - RouteByLatency: p.cfg.RouteByLatency, - RouteRandomly: p.cfg.RouteRandomly, - MasterName: p.cfg.MasterName, - }) - - // init fanin - p.fanin = newFanIn(p.universalClient, log) + p.cfgPlugin = cfg + p.stopCh = make(chan struct{}, 1) return nil } func (p *Plugin) Serve() chan error { - errCh := make(chan error) - return errCh + return make(chan error) } func (p *Plugin) Stop() error { const op = errors.Op("redis_plugin_stop") - err := p.fanin.stop() - if err != nil { - return errors.E(op, err) - } + p.stopCh <- struct{}{} - err = p.universalClient.Close() - if err != nil { - return errors.E(op, err) + if p.universalClient != nil { + err := p.universalClient.Close() + if err != nil { + return errors.E(op, err) + } } return nil @@ -103,112 +59,17 @@ func (p *Plugin) Name() string { // Available interface implementation func (p *Plugin) Available() {} -func (p *Plugin) Publish(msg []byte) error { - p.Lock() - defer p.Unlock() - - m := &websocketsv1.Message{} - err := proto.Unmarshal(msg, m) - if err != nil { - return errors.E(err) - } - - for j := 0; j < len(m.GetTopics()); j++ { - f := p.universalClient.Publish(context.Background(), m.GetTopics()[j], msg) - if f.Err() != nil { - return f.Err() - } - } - return nil -} - -func (p *Plugin) PublishAsync(msg []byte) { - go func() { - p.Lock() - defer p.Unlock() - m := &websocketsv1.Message{} - err := proto.Unmarshal(msg, m) - if err != nil { - p.log.Error("message unmarshal error") - return - } - - for j := 0; j < len(m.GetTopics()); j++ { - f := p.universalClient.Publish(context.Background(), m.GetTopics()[j], msg) - if f.Err() != nil { - p.log.Error("redis publish", "error", f.Err()) - } - } - }() -} - -func (p *Plugin) Subscribe(connectionID string, topics ...string) error { - // just add a connection - for i := 0; i < len(topics); i++ { - // key - topic - // value - connectionID - hset := p.universalClient.SAdd(context.Background(), topics[i], connectionID) - res, err := hset.Result() - if err != nil { - return err - } - if res == 0 { - p.log.Warn("could not subscribe to the provided topic", "connectionID", connectionID, "topic", topics[i]) - continue - } - } - - // and subscribe after - return p.fanin.sub(topics...) -} - -func (p *Plugin) Unsubscribe(connectionID string, topics ...string) error { - // Remove topics from the storage - for i := 0; i < len(topics); i++ { - srem := p.universalClient.SRem(context.Background(), topics[i], connectionID) - if srem.Err() != nil { - return srem.Err() - } - } - - for i := 0; i < len(topics); i++ { - // if there are no such topics, we can safely unsubscribe from the redis - exists := p.universalClient.Exists(context.Background(), topics[i]) - res, err := exists.Result() - if err != nil { - return err - } - - // if we have associated connections - skip - if res == 1 { // exists means that topic still exists and some other nodes may have connections associated with it - continue - } - - // else - unsubscribe - err = p.fanin.unsub(topics[i]) - if err != nil { - return err - } - } - - return nil -} - -func (p *Plugin) Connections(topic string, res map[string]struct{}) { - hget := p.universalClient.SMembersMap(context.Background(), topic) - r, err := hget.Result() +// KVProvide provides KV storage implementation over the redis plugin +func (p *Plugin) KVProvide(key string) (kv.Storage, error) { + const op = errors.Op("redis_plugin_provide") + st, err := NewRedisDriver(p.log, key, p.cfgPlugin) if err != nil { - panic(err) + return nil, errors.E(op, err) } - // assighn connections - // res expected to be from the sync.Pool - for k := range r { - res[k] = struct{}{} - } + return st, nil } -// Next return next message -func (p *Plugin) Next() (*websocketsv1.Message, error) { - return <-p.fanin.consume(), nil +func (p *Plugin) PSProvide(key string) (pubsub.PubSub, error) { + return NewPubSubDriver(p.log, key, p.cfgPlugin, p.stopCh) } diff --git a/plugins/redis/pubsub.go b/plugins/redis/pubsub.go new file mode 100644 index 00000000..dbda7ea4 --- /dev/null +++ b/plugins/redis/pubsub.go @@ -0,0 +1,189 @@ +package redis + +import ( + "context" + "sync" + + "github.com/go-redis/redis/v8" + "github.com/spiral/errors" + websocketsv1 "github.com/spiral/roadrunner/v2/pkg/proto/websockets/v1beta" + "github.com/spiral/roadrunner/v2/pkg/pubsub" + "github.com/spiral/roadrunner/v2/plugins/config" + "github.com/spiral/roadrunner/v2/plugins/logger" + "google.golang.org/protobuf/proto" +) + +type PubSubDriver struct { + sync.RWMutex + cfg *Config `mapstructure:"redis"` + + log logger.Logger + fanin *FanIn + universalClient redis.UniversalClient + stopCh chan struct{} +} + +func NewPubSubDriver(log logger.Logger, key string, cfgPlugin config.Configurer, stopCh chan struct{}) (pubsub.PubSub, error) { + const op = errors.Op("new_pub_sub_driver") + ps := &PubSubDriver{ + log: log, + stopCh: stopCh, + } + + // will be different for every connected driver + err := cfgPlugin.UnmarshalKey(key, &ps.cfg) + if err != nil { + return nil, errors.E(op, err) + } + + ps.cfg.InitDefaults() + + ps.universalClient = redis.NewUniversalClient(&redis.UniversalOptions{ + Addrs: ps.cfg.Addrs, + DB: ps.cfg.DB, + Username: ps.cfg.Username, + Password: ps.cfg.Password, + SentinelPassword: ps.cfg.SentinelPassword, + MaxRetries: ps.cfg.MaxRetries, + MinRetryBackoff: ps.cfg.MaxRetryBackoff, + MaxRetryBackoff: ps.cfg.MaxRetryBackoff, + DialTimeout: ps.cfg.DialTimeout, + ReadTimeout: ps.cfg.ReadTimeout, + WriteTimeout: ps.cfg.WriteTimeout, + PoolSize: ps.cfg.PoolSize, + MinIdleConns: ps.cfg.MinIdleConns, + MaxConnAge: ps.cfg.MaxConnAge, + PoolTimeout: ps.cfg.PoolTimeout, + IdleTimeout: ps.cfg.IdleTimeout, + IdleCheckFrequency: ps.cfg.IdleCheckFreq, + ReadOnly: ps.cfg.ReadOnly, + RouteByLatency: ps.cfg.RouteByLatency, + RouteRandomly: ps.cfg.RouteRandomly, + MasterName: ps.cfg.MasterName, + }) + + ps.fanin = newFanIn(ps.universalClient, log) + + ps.stop() + + return ps, nil +} + +func (p *PubSubDriver) stop() { + go func() { + for range p.stopCh { + _ = p.fanin.stop() + return + } + }() +} + +func (p *PubSubDriver) Publish(msg []byte) error { + p.Lock() + defer p.Unlock() + + m := &websocketsv1.Message{} + err := proto.Unmarshal(msg, m) + if err != nil { + return errors.E(err) + } + + for j := 0; j < len(m.GetTopics()); j++ { + f := p.universalClient.Publish(context.Background(), m.GetTopics()[j], msg) + if f.Err() != nil { + return f.Err() + } + } + return nil +} + +func (p *PubSubDriver) PublishAsync(msg []byte) { + go func() { + p.Lock() + defer p.Unlock() + m := &websocketsv1.Message{} + err := proto.Unmarshal(msg, m) + if err != nil { + p.log.Error("message unmarshal error") + return + } + + for j := 0; j < len(m.GetTopics()); j++ { + f := p.universalClient.Publish(context.Background(), m.GetTopics()[j], msg) + if f.Err() != nil { + p.log.Error("redis publish", "error", f.Err()) + } + } + }() +} + +func (p *PubSubDriver) Subscribe(connectionID string, topics ...string) error { + // just add a connection + for i := 0; i < len(topics); i++ { + // key - topic + // value - connectionID + hset := p.universalClient.SAdd(context.Background(), topics[i], connectionID) + res, err := hset.Result() + if err != nil { + return err + } + if res == 0 { + p.log.Warn("could not subscribe to the provided topic", "connectionID", connectionID, "topic", topics[i]) + continue + } + } + + // and subscribe after + return p.fanin.sub(topics...) +} + +func (p *PubSubDriver) Unsubscribe(connectionID string, topics ...string) error { + // Remove topics from the storage + for i := 0; i < len(topics); i++ { + srem := p.universalClient.SRem(context.Background(), topics[i], connectionID) + if srem.Err() != nil { + return srem.Err() + } + } + + for i := 0; i < len(topics); i++ { + // if there are no such topics, we can safely unsubscribe from the redis + exists := p.universalClient.Exists(context.Background(), topics[i]) + res, err := exists.Result() + if err != nil { + return err + } + + // if we have associated connections - skip + if res == 1 { // exists means that topic still exists and some other nodes may have connections associated with it + continue + } + + // else - unsubscribe + err = p.fanin.unsub(topics[i]) + if err != nil { + return err + } + } + + return nil +} + +func (p *PubSubDriver) Connections(topic string, res map[string]struct{}) { + hget := p.universalClient.SMembersMap(context.Background(), topic) + r, err := hget.Result() + if err != nil { + panic(err) + } + + // assighn connections + // res expected to be from the sync.Pool + for k := range r { + res[k] = struct{}{} + } +} + +// Next return next message +func (p *PubSubDriver) Next() (*websocketsv1.Message, error) { + return <-p.fanin.consume(), nil +} diff --git a/plugins/websockets/config.go b/plugins/websockets/config.go index be4aaa82..93d9ac3b 100644 --- a/plugins/websockets/config.go +++ b/plugins/websockets/config.go @@ -7,14 +7,48 @@ import ( ) /* +# GLOBAL +redis: + addrs: + - 'localhost:6379' + websockets: # pubsubs should implement PubSub interface to be collected via endure.Collects pubsubs:["redis", "amqp", "memory"] + # OR local + redis: + addrs: + - 'localhost:6379' + # path used as websockets path path: "/ws" */ +type RedisConfig struct { + Addrs []string `mapstructure:"addrs"` + DB int `mapstructure:"db"` + Username string `mapstructure:"username"` + Password string `mapstructure:"password"` + MasterName string `mapstructure:"master_name"` + SentinelPassword string `mapstructure:"sentinel_password"` + RouteByLatency bool `mapstructure:"route_by_latency"` + RouteRandomly bool `mapstructure:"route_randomly"` + MaxRetries int `mapstructure:"max_retries"` + DialTimeout time.Duration `mapstructure:"dial_timeout"` + MinRetryBackoff time.Duration `mapstructure:"min_retry_backoff"` + MaxRetryBackoff time.Duration `mapstructure:"max_retry_backoff"` + PoolSize int `mapstructure:"pool_size"` + MinIdleConns int `mapstructure:"min_idle_conns"` + MaxConnAge time.Duration `mapstructure:"max_conn_age"` + ReadTimeout time.Duration `mapstructure:"read_timeout"` + WriteTimeout time.Duration `mapstructure:"write_timeout"` + PoolTimeout time.Duration `mapstructure:"pool_timeout"` + IdleTimeout time.Duration `mapstructure:"idle_timeout"` + IdleCheckFreq time.Duration `mapstructure:"idle_check_freq"` + ReadOnly bool `mapstructure:"read_only"` +} + // Config represents configuration for the ws plugin type Config struct { // http path for the websocket @@ -23,6 +57,8 @@ type Config struct { PubSubs []string `mapstructure:"pubsubs"` Middleware []string `mapstructure:"middleware"` + Redis *RedisConfig `mapstructure:"redis"` + Pool *pool.Config `mapstructure:"pool"` } @@ -55,4 +91,11 @@ func (c *Config) InitDefault() { } c.Pool.Supervisor.InitDefaults() } + + if c.Redis != nil { + if c.Redis.Addrs == nil { + // append default + c.Redis.Addrs = append(c.Redis.Addrs, "localhost:6379") + } + } } diff --git a/plugins/websockets/memory/inMemory.go b/plugins/websockets/memory/inMemory.go deleted file mode 100644 index cef28182..00000000 --- a/plugins/websockets/memory/inMemory.go +++ /dev/null @@ -1,95 +0,0 @@ -package memory - -import ( - "sync" - - "github.com/spiral/roadrunner/v2/pkg/bst" - websocketsv1 "github.com/spiral/roadrunner/v2/pkg/proto/websockets/v1beta" - "github.com/spiral/roadrunner/v2/pkg/pubsub" - "github.com/spiral/roadrunner/v2/plugins/logger" - "google.golang.org/protobuf/proto" -) - -type Plugin struct { - sync.RWMutex - log logger.Logger - - // channel with the messages from the RPC - pushCh chan []byte - // user-subscribed topics - storage bst.Storage -} - -func NewInMemory(log logger.Logger) pubsub.PubSub { - return &Plugin{ - log: log, - pushCh: make(chan []byte, 10), - storage: bst.NewBST(), - } -} - -func (p *Plugin) Publish(message []byte) error { - p.pushCh <- message - return nil -} - -func (p *Plugin) PublishAsync(message []byte) { - go func() { - p.pushCh <- message - }() -} - -func (p *Plugin) Subscribe(connectionID string, topics ...string) error { - p.Lock() - defer p.Unlock() - for i := 0; i < len(topics); i++ { - p.storage.Insert(connectionID, topics[i]) - } - return nil -} - -func (p *Plugin) Unsubscribe(connectionID string, topics ...string) error { - p.Lock() - defer p.Unlock() - for i := 0; i < len(topics); i++ { - p.storage.Remove(connectionID, topics[i]) - } - return nil -} - -func (p *Plugin) Connections(topic string, res map[string]struct{}) { - p.RLock() - defer p.RUnlock() - - ret := p.storage.Get(topic) - for rr := range ret { - res[rr] = struct{}{} - } -} - -func (p *Plugin) Next() (*websocketsv1.Message, error) { - msg := <-p.pushCh - if msg == nil { - return nil, nil - } - - p.RLock() - defer p.RUnlock() - - m := &websocketsv1.Message{} - err := proto.Unmarshal(msg, m) - if err != nil { - return nil, err - } - - // push only messages, which are subscribed - // TODO better??? - for i := 0; i < len(m.GetTopics()); i++ { - // if we have active subscribers - send a message to a topic - // or send nil instead - if ok := p.storage.Contains(m.GetTopics()[i]); ok { - return m, nil - } - } - return nil, nil -} diff --git a/plugins/websockets/plugin.go b/plugins/websockets/plugin.go index 6ddd609c..c53491b4 100644 --- a/plugins/websockets/plugin.go +++ b/plugins/websockets/plugin.go @@ -2,6 +2,7 @@ package websockets import ( "context" + "fmt" "net/http" "sync" "time" @@ -23,7 +24,6 @@ import ( "github.com/spiral/roadrunner/v2/plugins/server" "github.com/spiral/roadrunner/v2/plugins/websockets/connection" "github.com/spiral/roadrunner/v2/plugins/websockets/executor" - "github.com/spiral/roadrunner/v2/plugins/websockets/memory" "github.com/spiral/roadrunner/v2/plugins/websockets/pool" "github.com/spiral/roadrunner/v2/plugins/websockets/validator" "google.golang.org/protobuf/proto" @@ -38,8 +38,11 @@ type Plugin struct { // Collection with all available pubsubs pubsubs map[string]pubsub.PubSub - cfg *Config - log logger.Logger + psProviders map[string]pubsub.PSProvider + + cfg *Config + cfgPlugin config.Configurer + log logger.Logger // global connections map connections sync.Map @@ -69,9 +72,12 @@ func (p *Plugin) Init(cfg config.Configurer, log logger.Logger, server server.Se } p.cfg.InitDefault() - p.pubsubs = make(map[string]pubsub.PubSub) + p.psProviders = make(map[string]pubsub.PSProvider) + p.log = log + p.cfgPlugin = cfg + p.wsUpgrade = &websocket.Upgrader{ HandshakeTimeout: time.Second * 60, ReadBufferSize: 1024, @@ -80,14 +86,18 @@ func (p *Plugin) Init(cfg config.Configurer, log logger.Logger, server server.Se p.serveExit = make(chan struct{}) p.server = server - // attach default driver - p.pubsubs["memory"] = memory.NewInMemory(p.log) - return nil } func (p *Plugin) Serve() chan error { errCh := make(chan error) + const op = errors.Op("websockets_plugin_serve") + + err := p.initPubSubs() + if err != nil { + errCh <- errors.E(op, err) + return errCh + } go func() { var err error @@ -133,6 +143,54 @@ func (p *Plugin) Serve() chan error { return errCh } +func (p *Plugin) initPubSubs() error { + for i := 0; i < len(p.cfg.PubSubs); i++ { + // don't need to have a section for the in-memory + if p.cfg.PubSubs[i] == "memory" { + if provider, ok := p.psProviders[p.cfg.PubSubs[i]]; ok { + r, err := provider.PSProvide("") + if err != nil { + return err + } + + // append default in-memory provider + p.pubsubs["memory"] = r + } + continue + } + // key - memory, redis + if provider, ok := p.psProviders[p.cfg.PubSubs[i]]; ok { + // try local key + switch { + // try local config first + case p.cfgPlugin.Has(fmt.Sprintf("%s.%s", PluginName, p.cfg.PubSubs[i])): + r, err := provider.PSProvide(fmt.Sprintf("%s.%s", PluginName, p.cfg.PubSubs[i])) + if err != nil { + return err + } + + // append redis provider + p.pubsubs[p.cfg.PubSubs[i]] = r + case p.cfgPlugin.Has(p.cfg.PubSubs[i]): + r, err := provider.PSProvide(p.cfg.PubSubs[i]) + if err != nil { + return err + } + + // append redis provider + p.pubsubs[p.cfg.PubSubs[i]] = r + default: + return errors.Errorf("could not find configuration sections for the %s", p.cfg.PubSubs[i]) + } + } else { + // no such driver + p.log.Warn("no such driver", "requested", p.cfg.PubSubs[i], "available", p.psProviders) + } + } + + return nil +} + func (p *Plugin) Stop() error { // close workers pool p.workersPool.Stop() @@ -167,8 +225,8 @@ func (p *Plugin) Name() string { } // GetPublishers collects all pubsubs -func (p *Plugin) GetPublishers(name endure.Named, pub pubsub.PubSub) { - p.pubsubs[name.Name()] = pub +func (p *Plugin) GetPublishers(name endure.Named, pub pubsub.PSProvider) { + p.psProviders[name.Name()] = pub } func (p *Plugin) Middleware(next http.Handler) http.Handler { @@ -389,7 +447,6 @@ func (p *Plugin) defaultAccessValidator(pool phpPool.Pool) validator.AccessValid } } -// go:inline func exec(ctx []byte, pool phpPool.Pool) (*validator.AccessValidator, error) { const op = errors.Op("exec") pd := payload.Payload{ -- cgit v1.2.3