2021-05-07 16:36:23 +00:00
|
|
|
package stulbe
|
|
|
|
|
|
|
|
import (
|
2021-05-10 21:09:15 +00:00
|
|
|
"context"
|
2021-05-07 16:36:23 +00:00
|
|
|
|
|
|
|
"github.com/sirupsen/logrus"
|
|
|
|
|
2021-05-14 11:15:38 +00:00
|
|
|
"github.com/strimertul/stulbe-client-go"
|
2021-05-07 16:36:23 +00:00
|
|
|
|
2021-05-14 11:15:38 +00:00
|
|
|
"github.com/strimertul/strimertul/database"
|
2021-05-11 11:12:00 +00:00
|
|
|
)
|
2021-05-07 16:36:23 +00:00
|
|
|
|
2021-05-11 11:12:00 +00:00
|
|
|
type Manager struct {
|
|
|
|
Client *stulbe.Client
|
2021-05-10 21:09:15 +00:00
|
|
|
db *database.DB
|
2021-05-12 17:19:09 +00:00
|
|
|
logger logrus.FieldLogger
|
2021-05-07 16:36:23 +00:00
|
|
|
}
|
|
|
|
|
2021-05-11 11:12:00 +00:00
|
|
|
func Initialize(db *database.DB, logger logrus.FieldLogger) (*Manager, error) {
|
2021-05-07 16:36:23 +00:00
|
|
|
var config Config
|
2021-05-10 21:09:15 +00:00
|
|
|
err := db.GetJSON(ConfigKey, &config)
|
2021-05-07 16:36:23 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2021-05-11 11:12:00 +00:00
|
|
|
stulbeClient, err := stulbe.NewClient(stulbe.ClientOptions{
|
2021-05-07 16:36:23 +00:00
|
|
|
Endpoint: config.Endpoint,
|
2021-05-11 11:12:00 +00:00
|
|
|
Username: config.Username,
|
|
|
|
AuthKey: config.AuthKey,
|
|
|
|
Logger: logger,
|
|
|
|
})
|
2021-05-08 14:02:47 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2021-05-11 11:12:00 +00:00
|
|
|
return &Manager{
|
|
|
|
Client: stulbeClient,
|
|
|
|
db: db,
|
2021-05-12 17:19:09 +00:00
|
|
|
logger: logger,
|
2021-05-11 11:12:00 +00:00
|
|
|
}, err
|
2021-05-07 16:36:23 +00:00
|
|
|
}
|
|
|
|
|
2021-05-18 11:30:27 +00:00
|
|
|
func (m *Manager) ReceiveEvents() error {
|
|
|
|
chn, err := m.Client.KV.SubscribePrefix("stulbe/")
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
for {
|
|
|
|
kv := <-chn
|
|
|
|
err := m.db.PutKey(kv.Key, []byte(kv.Value))
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-05-11 11:12:00 +00:00
|
|
|
func (m *Manager) Close() {
|
|
|
|
m.Client.Close()
|
2021-05-07 16:36:23 +00:00
|
|
|
}
|
|
|
|
|
2021-05-16 15:55:40 +00:00
|
|
|
func (m *Manager) ReplicateKey(prefix string) error {
|
2021-05-11 11:12:00 +00:00
|
|
|
// Set key to current value
|
2021-05-16 15:55:40 +00:00
|
|
|
vals, err := m.db.GetAll(prefix)
|
2021-05-08 14:02:47 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2021-05-18 17:45:14 +00:00
|
|
|
// Add prefix to keys
|
|
|
|
newvals := make(map[string]string)
|
|
|
|
for k, v := range vals {
|
|
|
|
newvals[prefix+k] = v
|
|
|
|
}
|
|
|
|
|
|
|
|
err = m.Client.KV.SetKeys(newvals)
|
2021-05-18 11:30:27 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
2021-05-08 14:02:47 +00:00
|
|
|
}
|
2021-05-16 15:55:40 +00:00
|
|
|
|
2021-05-12 17:19:09 +00:00
|
|
|
m.logger.WithFields(logrus.Fields{
|
2021-05-16 15:55:40 +00:00
|
|
|
"prefix": prefix,
|
|
|
|
}).Debug("synched to remote")
|
2021-05-08 14:02:47 +00:00
|
|
|
|
2021-05-10 21:09:15 +00:00
|
|
|
// Subscribe to local datastore and update remote on change
|
2021-05-11 11:12:00 +00:00
|
|
|
return m.db.Subscribe(context.Background(), func(pairs []database.ModifiedKV) error {
|
2021-05-10 21:09:15 +00:00
|
|
|
for _, changed := range pairs {
|
2021-05-11 11:12:00 +00:00
|
|
|
err := m.Client.KV.SetKey(changed.Key, string(changed.Data))
|
2021-05-10 21:09:15 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2021-05-12 17:19:09 +00:00
|
|
|
m.logger.WithFields(logrus.Fields{
|
|
|
|
"key": changed.Key,
|
|
|
|
}).Debug("replicated to remote")
|
2021-05-10 21:09:15 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
2021-05-16 15:55:40 +00:00
|
|
|
}, prefix)
|
2021-05-10 21:09:15 +00:00
|
|
|
}
|