diff --git a/.drone.star b/.drone.star index be730039c4..0e6f3998bf 100644 --- a/.drone.star +++ b/.drone.star @@ -1901,6 +1901,8 @@ def ocisServer(storage, accounts_hash_difficulty = 4, volumes = [], depends_on = "FRONTEND_SEARCH_MIN_LENGTH": "2", "OCIS_ASYNC_UPLOADS": True, "OCIS_EVENTS_ENABLE_TLS": False, + "MICRO_REGISTRY": "natsjs", + "MICRO_REGISTRY_ADDRESS": "127.0.0.1:9233", } if deploy_type == "": diff --git a/changelog/unreleased/natsjs-registry.md b/changelog/unreleased/natsjs-registry.md new file mode 100644 index 0000000000..609c255f52 --- /dev/null +++ b/changelog/unreleased/natsjs-registry.md @@ -0,0 +1,6 @@ +Enhancement: Introduce natsjs registry + +Introduce a registry based on the natsjs object store + +https://github.com/owncloud/ocis/issues/7272 +https://github.com/owncloud/ocis/pull/7487 diff --git a/go.mod b/go.mod index e5aec0bbfa..a369ded4a3 100644 --- a/go.mod +++ b/go.mod @@ -34,6 +34,7 @@ require ( github.com/go-micro/plugins/v4/registry/nats v1.2.2-0.20230723205323-1ada01245674 github.com/go-micro/plugins/v4/server/grpc v1.2.0 github.com/go-micro/plugins/v4/server/http v1.2.2 + github.com/go-micro/plugins/v4/store/nats-js v1.2.1-0.20230807070816-bc05fb076ce7 github.com/go-micro/plugins/v4/wrapper/breaker/gobreaker v1.2.0 github.com/go-micro/plugins/v4/wrapper/monitoring/prometheus v1.2.0 github.com/go-micro/plugins/v4/wrapper/trace/opentelemetry v1.2.0 @@ -191,7 +192,6 @@ require ( github.com/go-logr/logr v1.2.4 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20230807070816-bc05fb076ce7 // indirect - github.com/go-micro/plugins/v4/store/nats-js v1.1.0 // indirect github.com/go-micro/plugins/v4/store/redis v1.2.1-0.20230510195111-07cd57e1bc9d // indirect github.com/go-playground/locales v0.13.0 // indirect github.com/go-playground/universal-translator v0.17.0 // indirect @@ -341,3 +341,5 @@ require ( gopkg.in/yaml.v3 v3.0.1 // indirect sigs.k8s.io/yaml v1.3.0 // indirect ) + +replace github.com/go-micro/plugins/v4/store/nats-js => github.com/kobergj/plugins/v4/store/nats-js v1.2.1-0.20231020092801-9463c820c19a diff --git a/go.sum b/go.sum index b78ae87203..d61fb77130 100644 --- a/go.sum +++ b/go.sum @@ -1200,8 +1200,6 @@ github.com/go-micro/plugins/v4/server/grpc v1.2.0 h1:lXfM+/0oE/u1g0hVBYsvbP4lYOY github.com/go-micro/plugins/v4/server/grpc v1.2.0/go.mod h1:+Ah9Pf/vMSXxBM3fup/hc3N+zN2as3nIpcRaR4sBjnY= github.com/go-micro/plugins/v4/server/http v1.2.2 h1:UK2/09AU0zV3wHELuR72TZzVU2vTUhbx9qrRGrQSIWg= github.com/go-micro/plugins/v4/server/http v1.2.2/go.mod h1:YuAjaSPxcn3LI8j2FUsqx0Rxunrj4YwDV41Ax76rLl0= -github.com/go-micro/plugins/v4/store/nats-js v1.1.0 h1:6Fe1/eLtg8kRyaGvMILp4olYtTDGwYNBXyb1sYfAWGk= -github.com/go-micro/plugins/v4/store/nats-js v1.1.0/go.mod h1:jJf7Gm39OafZlT3s3UE2/9NIYj6OlI2fmZ4czSA3gvo= github.com/go-micro/plugins/v4/store/redis v1.2.1-0.20230510195111-07cd57e1bc9d h1:HQoDDVyMfdkrgXNo03ZY4vzhoOXMDZVZ4SnpBDVID6E= github.com/go-micro/plugins/v4/store/redis v1.2.1-0.20230510195111-07cd57e1bc9d/go.mod h1:MbCG0YiyPqETTtm7uHFmxQNCaW1o9hBoYtFwhbVjLUg= github.com/go-micro/plugins/v4/transport/grpc v1.1.0 h1:mXfDYfFQLnVDzjGY3o84oe4prfux9h8txsnA19dKsj8= @@ -1577,6 +1575,8 @@ github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa02 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.1.0 h1:eyi1Ad2aNJMW95zcSbmGg7Cg6cq3ADwLpMAP96d8rF0= github.com/klauspost/cpuid/v2 v2.1.0/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY= +github.com/kobergj/plugins/v4/store/nats-js v1.2.1-0.20231020092801-9463c820c19a h1:W+itvdTLFGLuFh+E5IzW08n2BS02cHK91qnMo7SUxbA= +github.com/kobergj/plugins/v4/store/nats-js v1.2.1-0.20231020092801-9463c820c19a/go.mod h1:wt51O2yNmgF/F7E00IYIH0awseRGqtnmjZGn6RjbZSk= github.com/kolo/xmlrpc v0.0.0-20200310150728-e0350524596b/go.mod h1:o03bZfuBwAXHetKXuInt4S7omeXUu62/A845kiycsSQ= github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= diff --git a/ocis-pkg/natsjsregistry/options.go b/ocis-pkg/natsjsregistry/options.go new file mode 100644 index 0000000000..dfcda616a5 --- /dev/null +++ b/ocis-pkg/natsjsregistry/options.go @@ -0,0 +1,32 @@ +package natsjsregistry + +import ( + "context" + "time" + + "go-micro.dev/v4/registry" + "go-micro.dev/v4/store" +) + +type storeOptionsKey struct{} +type expiryKey struct{} + +// StoreOptions sets the options for the underlying store +func StoreOptions(opts []store.Option) registry.Option { + return func(o *registry.Options) { + if o.Context == nil { + o.Context = context.Background() + } + o.Context = context.WithValue(o.Context, storeOptionsKey{}, opts) + } +} + +// ServiceExpiry allows setting an expiry time for service registrations +func ServiceExpiry(t time.Duration) registry.Option { + return func(o *registry.Options) { + if o.Context == nil { + o.Context = context.Background() + } + o.Context = context.WithValue(o.Context, expiryKey{}, t) + } +} diff --git a/ocis-pkg/natsjsregistry/registry.go b/ocis-pkg/natsjsregistry/registry.go new file mode 100644 index 0000000000..5f1e479fc4 --- /dev/null +++ b/ocis-pkg/natsjsregistry/registry.go @@ -0,0 +1,137 @@ +// Package natsjsregistry implements a registry using natsjs object store +package natsjsregistry + +import ( + "context" + "encoding/json" + "errors" + "time" + + natsjs "github.com/go-micro/plugins/v4/store/nats-js" + "go-micro.dev/v4/registry" + "go-micro.dev/v4/store" + "go-micro.dev/v4/util/cmd" +) + +var _registryName = "natsjs" + +func init() { + cmd.DefaultRegistries[_registryName] = NewRegistry +} + +// NewRegistry returns a new natsjs registry +func NewRegistry(opts ...registry.Option) registry.Registry { + options := registry.Options{ + Context: context.Background(), + } + for _, o := range opts { + o(&options) + } + exp, _ := options.Context.Value(expiryKey{}).(time.Duration) + return &storeregistry{ + opts: options, + store: natsjs.NewStore(storeOptions(options)...), + typ: _registryName, + expiry: exp, + } +} + +type storeregistry struct { + opts registry.Options + store store.Store + typ string + expiry time.Duration +} + +// Init inits the registry +func (n *storeregistry) Init(opts ...registry.Option) error { + for _, o := range opts { + o(&n.opts) + } + return n.store.Init(storeOptions(n.opts)...) +} + +// Options returns the configured options +func (n *storeregistry) Options() registry.Options { + return n.opts +} + +// Register adds a service to the registry +func (n *storeregistry) Register(s *registry.Service, _ ...registry.RegisterOption) error { + if s == nil { + return errors.New("wont store nil service") + } + b, err := json.Marshal(s) + if err != nil { + return err + } + return n.store.Write(&store.Record{ + Key: s.Name, + Value: b, + Expiry: n.expiry, + }) +} + +// Deregister removes a service from the registry +func (n *storeregistry) Deregister(s *registry.Service, _ ...registry.DeregisterOption) error { + return n.store.Delete(s.Name) +} + +// GetService gets a specific service from the registry +func (n *storeregistry) GetService(s string, _ ...registry.GetOption) ([]*registry.Service, error) { + recs, err := n.store.Read(s) + if err != nil { + return nil, err + } + svcs := make([]*registry.Service, 0, len(recs)) + for _, rec := range recs { + var s registry.Service + if err := json.Unmarshal(rec.Value, &s); err != nil { + return nil, err + } + svcs = append(svcs, &s) + } + return svcs, nil +} + +// ListServices lists all registered services +func (n *storeregistry) ListServices(...registry.ListOption) ([]*registry.Service, error) { + keys, err := n.store.List() + if err != nil { + return nil, err + } + + var svcs []*registry.Service + for _, k := range keys { + s, err := n.GetService(k) + if err != nil { + // TODO: continue ? + return nil, err + } + svcs = append(svcs, s...) + + } + return svcs, nil +} + +// Watch allowes following the changes in the registry if it would be implemented +func (n *storeregistry) Watch(...registry.WatchOption) (registry.Watcher, error) { + return nil, errors.New("watcher not implemented") +} + +// String returns the name of the registry +func (n *storeregistry) String() string { + return n.typ +} + +func storeOptions(opts registry.Options) []store.Option { + storeoptions := []store.Option{ + store.Nodes(opts.Addrs...), + store.Database("service-registry"), + store.Table("service-registry"), + } + if so, ok := opts.Context.Value(storeOptionsKey{}).([]store.Option); ok { + storeoptions = append(storeoptions, so...) + } + return storeoptions +} diff --git a/ocis-pkg/registry/registry.go b/ocis-pkg/registry/registry.go index dd2245cfa4..ed9ece2ae5 100644 --- a/ocis-pkg/registry/registry.go +++ b/ocis-pkg/registry/registry.go @@ -13,6 +13,7 @@ import ( mdnsr "github.com/go-micro/plugins/v4/registry/mdns" memr "github.com/go-micro/plugins/v4/registry/memory" natsr "github.com/go-micro/plugins/v4/registry/nats" + "github.com/owncloud/ocis/v2/ocis-pkg/natsjsregistry" mRegistry "go-micro.dev/v4/registry" "go-micro.dev/v4/registry/cache" ) @@ -28,6 +29,7 @@ var ( reg mRegistry.Registry ) +// Configure configures the registry func Configure(plugin string) { if reg == nil { regPlugin = plugin @@ -66,6 +68,10 @@ func GetRegistry() mRegistry.Registry { ) case "memory": reg = memr.NewRegistry() + case "natsjs": + reg = natsjsregistry.NewRegistry( + mRegistry.Addrs(addresses...), + ) default: reg = mdnsr.NewRegistry() } diff --git a/ocis/pkg/runtime/service/service.go b/ocis/pkg/runtime/service/service.go index ed59185848..5c74c6e0f6 100644 --- a/ocis/pkg/runtime/service/service.go +++ b/ocis/pkg/runtime/service/service.go @@ -64,6 +64,8 @@ import ( var ( // runset keeps track of which services to start supervised. runset map[string]struct{} + // time to wait between starting service groups (preliminary, main, delayed) + _startDelay = 2 * time.Second ) type serviceFuncMap map[string]func(*ociscfg.Config) suture.Service @@ -71,6 +73,7 @@ type serviceFuncMap map[string]func(*ociscfg.Config) suture.Service // Service represents a RPC service. type Service struct { Supervisor *suture.Supervisor + Preliminary serviceFuncMap ServicesRegistry serviceFuncMap Delayed serviceFuncMap Additional serviceFuncMap @@ -103,6 +106,7 @@ func NewService(options ...Option) (*Service, error) { globalCtx, cancelGlobal := context.WithCancel(context.Background()) s := &Service{ + Preliminary: make(serviceFuncMap), ServicesRegistry: make(serviceFuncMap), Delayed: make(serviceFuncMap), Additional: make(serviceFuncMap), @@ -114,6 +118,13 @@ func NewService(options ...Option) (*Service, error) { cfg: opts.Config, } + // start nats first - it is used as service registry + s.Preliminary[opts.Config.Nats.Service.Name] = NewSutureServiceBuilder(func(ctx context.Context, cfg *ociscfg.Config) error { + cfg.Nats.Context = ctx + cfg.Nats.Commons = cfg.Commons + return nats.Execute(cfg.Nats) + }) + // populate services reg := func(name string, exec func(context.Context, *ociscfg.Config) error) { s.ServicesRegistry[name] = NewSutureServiceBuilder(exec) @@ -153,11 +164,6 @@ func NewService(options ...Option) (*Service, error) { cfg.EventHistory.Commons = cfg.Commons return eventhistory.Execute(cfg.EventHistory) }) - reg(opts.Config.Frontend.Service.Name, func(ctx context.Context, cfg *ociscfg.Config) error { - cfg.Frontend.Context = ctx - cfg.Frontend.Commons = cfg.Commons - return frontend.Execute(cfg.Frontend) - }) reg(opts.Config.Gateway.Service.Name, func(ctx context.Context, cfg *ociscfg.Config) error { cfg.Gateway.Context = ctx cfg.Gateway.Commons = cfg.Commons @@ -183,11 +189,6 @@ func NewService(options ...Option) (*Service, error) { cfg.Invitations.Commons = cfg.Commons return invitations.Execute(cfg.Invitations) }) - reg(opts.Config.Nats.Service.Name, func(ctx context.Context, cfg *ociscfg.Config) error { - cfg.Nats.Context = ctx - cfg.Nats.Commons = cfg.Commons - return nats.Execute(cfg.Nats) - }) reg(opts.Config.Notifications.Service.Name, func(ctx context.Context, cfg *ociscfg.Config) error { cfg.Notifications.Context = ctx cfg.Notifications.Commons = cfg.Commons @@ -298,6 +299,11 @@ func NewService(options ...Option) (*Service, error) { dreg := func(name string, exec func(context.Context, *ociscfg.Config) error) { s.Delayed[name] = NewSutureServiceBuilder(exec) } + dreg(opts.Config.Frontend.Service.Name, func(ctx context.Context, cfg *ociscfg.Config) error { + cfg.Frontend.Context = ctx + cfg.Frontend.Commons = cfg.Commons + return frontend.Execute(cfg.Frontend) + }) dreg(opts.Config.IDP.Service.Name, func(ctx context.Context, cfg *ociscfg.Config) error { cfg.IDP.Context = ctx cfg.IDP.Commons = cfg.Commons @@ -394,11 +400,8 @@ func Start(o ...Option) error { // prepare the set of services to run s.generateRunSet(s.cfg) - // schedule services that we are sure don't have interdependencies. - scheduleServiceTokens(s, s.ServicesRegistry) - - // schedule services that are optional - scheduleServiceTokens(s, s.Additional) + // schedule preliminary services first + scheduleServiceTokens(s, s.Preliminary) // there are reasons not to do this, but we have race conditions ourselves. Until we resolve them, mind the following disclaimer: // Calling ServeBackground will CORRECTLY start the supervisor running in a new goroutine. It is risky to directly run @@ -410,8 +413,17 @@ func Start(o ...Option) error { // trap will block on halt channel for interruptions. go trap(s, halt) + // grace period for supervisor to get up + time.Sleep(_startDelay) + + // schedule services that we are sure don't have interdependencies. + scheduleServiceTokens(s, s.ServicesRegistry) + + // schedule services that are optional + scheduleServiceTokens(s, s.Additional) + // add services with delayed execution. - time.Sleep(1 * time.Second) + time.Sleep(_startDelay) scheduleServiceTokens(s, s.Delayed) return http.Serve(l, nil) @@ -440,6 +452,10 @@ func (s *Service) generateRunSet(cfg *ociscfg.Config) { return } + for name := range s.Preliminary { + runset[name] = struct{}{} + } + for name := range s.ServicesRegistry { runset[name] = struct{}{} } diff --git a/vendor/github.com/go-micro/plugins/v4/store/nats-js/context.go b/vendor/github.com/go-micro/plugins/v4/store/nats-js/context.go index e2eec2a6f8..07965c5118 100644 --- a/vendor/github.com/go-micro/plugins/v4/store/nats-js/context.go +++ b/vendor/github.com/go-micro/plugins/v4/store/nats-js/context.go @@ -6,7 +6,7 @@ import ( "go-micro.dev/v4/store" ) -// setStoreOption returns a function to setup a context with given value +// setStoreOption returns a function to setup a context with given value. func setStoreOption(k, v interface{}) store.Option { return func(o *store.Options) { if o.Context == nil { diff --git a/vendor/github.com/go-micro/plugins/v4/store/nats-js/nats.go b/vendor/github.com/go-micro/plugins/v4/store/nats-js/nats.go index 907c6c0e62..c0a4f4004f 100644 --- a/vendor/github.com/go-micro/plugins/v4/store/nats-js/nats.go +++ b/vendor/github.com/go-micro/plugins/v4/store/nats-js/nats.go @@ -34,14 +34,14 @@ type natsStore struct { conn *nats.Conn js nats.JetStreamContext - buckets map[string]nats.ObjectStore + buckets *sync.Map } func init() { cmd.DefaultStores["natsjs"] = NewStore } -// NewStore will create a new NATS JetStream Object Store +// NewStore will create a new NATS JetStream Object Store. func NewStore(opts ...store.Option) store.Store { options := store.Options{ Nodes: []string{}, @@ -55,7 +55,7 @@ func NewStore(opts ...store.Option) store.Store { opts: options, jsopts: []nats.JSOpt{}, objStoreConfigs: []*nats.ObjectStoreConfig{}, - buckets: map[string]nats.ObjectStore{}, + buckets: &sync.Map{}, storageType: nats.FileStorage, } @@ -64,7 +64,9 @@ func NewStore(opts ...store.Option) store.Store { return n } -// Init initialises the store. It must perform any required setup on the backing storage implementation and check that it is ready for use, returning any errors. +// Init initializes the store. It must perform any required setup on the +// backing storage implementation and check that it is ready for use, +// returning any errors. func (n *natsStore) Init(opts ...store.Option) error { n.setOption(opts...) @@ -101,7 +103,7 @@ func (n *natsStore) Init(opts ...store.Option) error { if err != nil { return errors.Wrapf(err, "Failed to create bucket (%s)", cfg.Bucket) } - n.buckets[cfg.Bucket] = store + n.buckets.Store(cfg.Bucket, store) } return nil @@ -159,10 +161,8 @@ func (n *natsStore) Options() store.Options { // Read takes a single key name and optional ReadOptions. It returns matching []*Record or an error. func (n *natsStore) Read(key string, opts ...store.ReadOption) ([]*store.Record, error) { - if n.conn == nil { - if err := n.Init(); err != nil { - return nil, err - } + if err := n.initConn(); err != nil { + return nil, err } opt := store.ReadOptions{} @@ -178,51 +178,70 @@ func (n *natsStore) Read(key string, opts ...store.ReadOption) ([]*store.Record, opt.Table = n.opts.Table } - bucket, ok := n.buckets[opt.Database] + b, ok := n.buckets.Load(opt.Database) if !ok { return nil, ErrBucketNotFound } + bucket := b.(nats.ObjectStore) var keys []string - objects, err := bucket.List() - if err == nats.ErrNoObjectsFound { - return []*store.Record{}, nil - } else if err != nil { - return nil, errors.Wrap(err, "Failed to list objects") + + var keyPrefix, keySuffix string + + switch { + case opt.Prefix: + keyPrefix = getKey(key, opt.Table) + case opt.Suffix: + keySuffix = key + default: + keys = []string{getKey(key, opt.Table)} } - for _, obj := range objects { - name := obj.Name - if (!opt.Prefix && !opt.Suffix) && getKey(key, opt.Table) != name { - continue - } - if opt.Prefix && !strings.HasPrefix(name, getKey(key, opt.Table)) { - continue + if len(keys) == 0 { + objects, err := bucket.List() + if err == nats.ErrNoObjectsFound { + return []*store.Record{}, nil + } else if err != nil { + return nil, errors.Wrap(err, "Failed to list objects") } - if opt.Suffix && !strings.HasSuffix(name, key) { - continue + for _, obj := range objects { + name := obj.Name + if !strings.HasPrefix(name, opt.Table) { + continue + } + if (!opt.Prefix && !opt.Suffix) && key != name { + continue + } + if opt.Prefix && !strings.HasPrefix(name, keyPrefix) { + continue + } + if opt.Suffix && !strings.HasSuffix(name, keySuffix) { + continue + } + keys = append(keys, name) } - keys = append(keys, name) } records := []*store.Record{} for _, key := range keys { obj, err := bucket.Get(key) - if err != nil { + if err == nats.ErrObjectNotFound { + return []*store.Record{}, nil + } else if err != nil { return nil, errors.Wrap(err, "Failed to get object from bucket") } - - b, err := io.ReadAll(obj) - if err != nil { - return nil, errors.Wrap(err, "Failed to read returned bytes") - } + defer obj.Close() info, err := obj.Info() if err != nil { return nil, errors.Wrap(err, "Failed to fetch record info") } + if info.Deleted { + continue + } + metadata := map[string]interface{}{} for key, value := range info.Headers { var val interface{} @@ -232,14 +251,17 @@ func (n *natsStore) Read(key string, opts ...store.ReadOption) ([]*store.Record, metadata[key] = val } + b, err := io.ReadAll(obj) + if err != nil { + return nil, errors.Wrap(err, "Failed to read returned bytes") + } + records = append(records, &store.Record{ Key: key, Value: b, Metadata: metadata, }) - // Why is there a close method? - obj.Close() } if opt.Limit > 0 { @@ -253,10 +275,8 @@ func (n *natsStore) Read(key string, opts ...store.ReadOption) ([]*store.Record, // Write writes a record to the store, and returns an error if the record was not written. func (n *natsStore) Write(r *store.Record, opts ...store.WriteOption) error { - if n.conn == nil { - if err := n.Init(); err != nil { - return err - } + if err := n.initConn(); err != nil { + return err } opt := store.WriteOptions{} @@ -272,8 +292,8 @@ func (n *natsStore) Write(r *store.Record, opts ...store.WriteOption) error { opt.Table = n.opts.Table } - store, ok := n.buckets[opt.Database] - + s, ok := n.buckets.Load(opt.Database) + store, _ := s.(nats.ObjectStore) // Create new bucket if not exists if !ok { var err error @@ -307,10 +327,8 @@ func (n *natsStore) Write(r *store.Record, opts ...store.WriteOption) error { // Delete removes the record with the corresponding key from the store. func (n *natsStore) Delete(key string, opts ...store.DeleteOption) error { - if n.conn == nil { - if err := n.Init(); err != nil { - return err - } + if err := n.initConn(); err != nil { + return err } opt := store.DeleteOptions{} @@ -327,18 +345,19 @@ func (n *natsStore) Delete(key string, opts ...store.DeleteOption) error { } if opt.Table == "DELETE_BUCKET" { - delete(n.buckets, key) + n.buckets.Delete(key) if err := n.js.DeleteObjectStore(key); err != nil { return errors.Wrap(err, "Failed to delete bucket") } return nil } - store, ok := n.buckets[opt.Database] + s, ok := n.buckets.Load(opt.Database) if !ok { return ErrBucketNotFound } + store := s.(nats.ObjectStore) if err := store.Delete(getKey(key, opt.Table)); err != nil { return errors.Wrap(err, "Failed to delete data") } @@ -347,10 +366,8 @@ func (n *natsStore) Delete(key string, opts ...store.DeleteOption) error { // List returns any keys that match, or an empty list with no error if none matched. func (n *natsStore) List(opts ...store.ListOption) ([]string, error) { - if n.conn == nil { - if err := n.Init(); err != nil { - return nil, err - } + if err := n.initConn(); err != nil { + return nil, err } opt := store.ListOptions{} @@ -366,10 +383,11 @@ func (n *natsStore) List(opts ...store.ListOption) ([]string, error) { opt.Table = n.opts.Table } - store, ok := n.buckets[opt.Database] + s, ok := n.buckets.Load(opt.Database) if !ok { return nil, ErrBucketNotFound } + store := s.(nats.ObjectStore) objects, err := store.List() if err != nil { @@ -399,7 +417,7 @@ func (n *natsStore) List(opts ...store.ListOption) ([]string, error) { return keys, nil } -// Close the store +// Close the store. func (n *natsStore) Close() error { n.conn.Close() return nil @@ -430,6 +448,31 @@ func (n *natsStore) createNewBucket(name string) (nats.ObjectStore, error) { if err != nil { return nil, errors.Wrapf(err, "Failed to create new bucket (%s)", name) } - n.buckets[name] = store + n.buckets.Store(name, store) return store, err } + +// thread safe way to initialize the connection. +func (n *natsStore) initConn() error { + if n.hasConn() { + return nil + } + + n.Lock() + defer n.Unlock() + + // check if conn was initialized meanwhile + if n.conn != nil { + return nil + } + + return n.Init() +} + +// thread safe way to check if n is initialized. +func (n *natsStore) hasConn() bool { + n.RLock() + defer n.RUnlock() + + return n.conn != nil +} diff --git a/vendor/github.com/go-micro/plugins/v4/store/nats-js/options.go b/vendor/github.com/go-micro/plugins/v4/store/nats-js/options.go index 53a31b110a..21719e2ab6 100644 --- a/vendor/github.com/go-micro/plugins/v4/store/nats-js/options.go +++ b/vendor/github.com/go-micro/plugins/v4/store/nats-js/options.go @@ -7,7 +7,7 @@ import ( "go-micro.dev/v4/store" ) -// store.Option +// store.Option. type natsOptionsKey struct{} type jsOptionsKey struct{} type objOptionsKey struct{} @@ -15,15 +15,15 @@ type ttlOptionsKey struct{} type memoryOptionsKey struct{} type descriptionOptionsKey struct{} -// store.DeleteOption +// store.DeleteOption. type delBucketOptionsKey struct{} -// NatsOptions accepts nats.Options +// NatsOptions accepts nats.Options. func NatsOptions(opts nats.Options) store.Option { return setStoreOption(natsOptionsKey{}, opts) } -// JetStreamOptions accepts multiple nats.JSOpt +// JetStreamOptions accepts multiple nats.JSOpt. func JetStreamOptions(opts ...nats.JSOpt) store.Option { return setStoreOption(jsOptionsKey{}, opts) } @@ -35,34 +35,42 @@ func ObjectStoreOptions(cfg ...*nats.ObjectStoreConfig) store.Option { } // DefaultTTL sets the default TTL to use for new buckets -// By default no TTL is set. +// +// By default no TTL is set. // // TTL ON INDIVIDUAL WRITE CALLS IS NOT SUPPORTED, only bucket wide TTL. // Either set a default TTL with this option or provide bucket specific options -// with ObjectStoreOptions +// +// with ObjectStoreOptions func DefaultTTL(ttl time.Duration) store.Option { return setStoreOption(ttlOptionsKey{}, ttl) } // DefaultMemory sets the default storage type to memory only. // -// The default is file storage, persisting storage between service restarts. +// The default is file storage, persisting storage between service restarts. +// // Be aware that the default storage location of NATS the /tmp dir is, and thus -// won't persist reboots. +// +// won't persist reboots. func DefaultMemory() store.Option { return setStoreOption(memoryOptionsKey{}, nats.MemoryStorage) } // DefaultDescription sets the default description to use when creating new -// buckets. The default is "Store managed by go-micro" +// +// buckets. The default is "Store managed by go-micro" func DefaultDescription(text string) store.Option { return setStoreOption(descriptionOptionsKey{}, text) } // DeleteBucket will use the key passed to Delete as a bucket (database) name, -// and delete the bucket. +// +// and delete the bucket. +// // This option should not be combined with the store.DeleteFrom option, as -// that will overwrite the delete action. +// +// that will overwrite the delete action. func DeleteBucket() store.DeleteOption { return func(d *store.DeleteOptions) { d.Table = "DELETE_BUCKET" diff --git a/vendor/modules.txt b/vendor/modules.txt index 990f1a894a..339557a7b6 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -932,7 +932,7 @@ github.com/go-micro/plugins/v4/server/grpc # github.com/go-micro/plugins/v4/server/http v1.2.2 ## explicit; go 1.17 github.com/go-micro/plugins/v4/server/http -# github.com/go-micro/plugins/v4/store/nats-js v1.1.0 +# github.com/go-micro/plugins/v4/store/nats-js v1.2.1-0.20230807070816-bc05fb076ce7 => github.com/kobergj/plugins/v4/store/nats-js v1.2.1-0.20231020092801-9463c820c19a ## explicit; go 1.17 github.com/go-micro/plugins/v4/store/nats-js # github.com/go-micro/plugins/v4/store/redis v1.2.1-0.20230510195111-07cd57e1bc9d @@ -2262,3 +2262,4 @@ stash.kopano.io/kgol/oidc-go # stash.kopano.io/kgol/rndm v1.1.2 ## explicit; go 1.13 stash.kopano.io/kgol/rndm +# github.com/go-micro/plugins/v4/store/nats-js => github.com/kobergj/plugins/v4/store/nats-js v1.2.1-0.20231020092801-9463c820c19a