[#195] Add Lock tick listening

Signed-off-by: Denis Kirillov <denis@nspcc.ru>

api/layer/layer.go

@@ -9,18 +9,19 @@ import (
 	"net/url"
 	"strings"
 
+	"github.com/nats-io/nats.go"
 	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
 	"github.com/nspcc-dev/neofs-s3-gw/api"
 	"github.com/nspcc-dev/neofs-s3-gw/api/cache"
 	"github.com/nspcc-dev/neofs-s3-gw/api/data"
 	"github.com/nspcc-dev/neofs-s3-gw/api/errors"
 	"github.com/nspcc-dev/neofs-s3-gw/api/layer/neofs"
-	"github.com/nspcc-dev/neofs-s3-gw/api/notifications"
 	"github.com/nspcc-dev/neofs-s3-gw/api/resolver"
 	"github.com/nspcc-dev/neofs-s3-gw/creds/accessbox"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
 	"github.com/nspcc-dev/neofs-sdk-go/eacl"
 	"github.com/nspcc-dev/neofs-sdk-go/netmap"
+	"github.com/nspcc-dev/neofs-sdk-go/object/address"
 	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
 	"github.com/nspcc-dev/neofs-sdk-go/owner"
 	"github.com/nspcc-dev/neofs-sdk-go/pool"
@@ -29,12 +30,23 @@ import (
 )
 
 type (
+	Notificator interface {
+		Subscribe(context.Context, string, MsgHandler) error
+		Listen(context.Context)
+	}
+
+	MsgHandler interface {
+		HandleMessage(context.Context, *nats.Msg) error
+	}
+
+	MsgHandlerFunc func(context.Context, *nats.Msg) error
+
 	layer struct {
 		neoFS       neofs.NeoFS
 		log         *zap.Logger
 		anonKey     AnonymousKey
 		resolver    *resolver.BucketResolver
-		ncontroller *notifications.Controller
+		ncontroller Notificator
 		listsCache  *cache.ObjectsListCache
 		objCache    *cache.ObjectsCache
 		namesCache  *cache.ObjectsNameCache
@@ -47,7 +59,6 @@ type (
 		Caches       *CachesConfig
 		AnonKey      AnonymousKey
 		Resolver     *resolver.BucketResolver
-		NotificationController *notifications.Controller
 	}
 
 	// AnonymousKey contains data for anonymous requests.
@@ -174,6 +185,7 @@ type (
 
 	// Client provides S3 API client interface.
 	Client interface {
+		Initialize(ctx context.Context, c Notificator) error
 		EphemeralKey() *keys.PublicKey
 
 		GetBucketSettings(ctx context.Context, bktInfo *data.BucketInfo) (*data.BucketSettings, error)
@@ -234,6 +246,10 @@ func (t *VersionedObject) String() string {
 	return t.Name + ":" + t.VersionID
 }
 
+func (f MsgHandlerFunc) HandleMessage(ctx context.Context, msg *nats.Msg) error {
+	return f(ctx, msg)
+}
+
 // DefaultCachesConfigs returns filled configs.
 func DefaultCachesConfigs() *CachesConfig {
 	return &CachesConfig{
@@ -254,7 +270,6 @@ func NewLayer(log *zap.Logger, neoFS neofs.NeoFS, config *Config) Client {
 		anonKey:     config.AnonKey,
 		resolver:    config.Resolver,
 		listsCache:  cache.NewObjectsListCache(config.Caches.ObjectsList),
-		ncontroller: config.NotificationController,
 		objCache:    cache.New(config.Caches.Objects),
 		namesCache:  cache.NewObjectsNameCache(config.Caches.Names),
 		bucketCache: cache.NewBucketCache(config.Caches.Buckets),
@@ -266,10 +281,40 @@ func (n *layer) EphemeralKey() *keys.PublicKey {
 	return n.anonKey.Key.PublicKey()
 }
 
+func (n *layer) Initialize(ctx context.Context, c Notificator) error {
+	if c == nil {
+		return nil
+	}
+
+	if n.IsNotificationEnabled() {
+		return fmt.Errorf("already initialized")
+	}
+
+	if err := c.Subscribe(ctx, "lock", MsgHandlerFunc(n.handleLockTick)); err != nil {
+		return fmt.Errorf("couldn't initialize layer: %w", err)
+	}
+
+	c.Listen(ctx)
+
+	n.ncontroller = c
+	return nil
+}
+
 func (n *layer) IsNotificationEnabled() bool {
 	return n.ncontroller != nil
 }
 
+func (n *layer) handleLockTick(ctx context.Context, msg *nats.Msg) error {
+	addr := address.NewAddress()
+	if err := addr.Parse(string(msg.Data)); err != nil {
+		return fmt.Errorf("invalid msg, address expected: %w", err)
+	}
+
+	// todo clear cache
+	// and make sure having right access
+	return n.objectDelete(ctx, addr.ContainerID(), addr.ObjectID())
+}
+
 // IsAuthenticatedRequest check if access box exists in current request.
 func IsAuthenticatedRequest(ctx context.Context) bool {
 	_, ok := ctx.Value(api.BoxData).(*accessbox.Box)

api/notifications/controller.go

@@ -1,9 +1,13 @@
 package notifications
 
 import (
+	"context"
+	"fmt"
+	"sync"
 	"time"
 
 	"github.com/nats-io/nats.go"
+	"github.com/nspcc-dev/neofs-s3-gw/api/layer"
 )
 
 const (
@@ -20,7 +24,14 @@ type Options struct {
 
 type Controller struct {
 	taskQueueConnection *nats.Conn
-	jsClient            nats.JetStream
+	jsClient            nats.JetStreamContext
+	handlers            map[string]Stream
+	mu                  sync.RWMutex
+}
+
+type Stream struct {
+	h  layer.MsgHandler
+	ch chan *nats.Msg
 }
 
 func NewController(p *Options) (*Controller, error) {
@@ -52,5 +63,56 @@ func NewController(p *Options) (*Controller, error) {
 	return &Controller{
 		taskQueueConnection: nc,
 		jsClient:            js,
+		handlers:            make(map[string]Stream),
 	}, nil
 }
+
+func (c *Controller) Subscribe(ctx context.Context, topic string, handler layer.MsgHandler) error {
+	ch := make(chan *nats.Msg, 1)
+
+	c.mu.RLock()
+	if _, ok := c.handlers[topic]; ok {
+		return fmt.Errorf("already subscribed to topic '%s'", topic)
+	}
+	c.mu.RUnlock()
+
+	if _, err := c.jsClient.AddStream(&nats.StreamConfig{Name: topic}); err != nil {
+		return err
+	}
+
+	if _, err := c.jsClient.ChanSubscribe(topic, ch); err != nil {
+		return fmt.Errorf("could not subscribe: %w", err)
+	}
+
+	c.mu.Lock()
+	c.handlers[topic] = Stream{
+		h:  handler,
+		ch: ch,
+	}
+	c.mu.Unlock()
+
+	return nil
+}
+
+func (c *Controller) Listen(ctx context.Context) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	for _, stream := range c.handlers {
+		go func(stream Stream) {
+			for {
+				select {
+				case msg := <-stream.ch:
+					fmt.Printf("got message: %s\n", msg.Data)
+					if err := stream.h.HandleMessage(ctx, msg); err != nil {
+						fmt.Printf("could not handle message: %s", err)
+					} else if err = msg.Ack(); err != nil {
+						fmt.Printf("could not ACK message: %s", err)
+					}
+				case <-ctx.Done():
+					return
+				}
+			}
+		}(stream)
+	}
+}

cmd/s3-gw/app.go

@@ -155,12 +155,15 @@ func newApp(ctx context.Context, l *zap.Logger, v *viper.Viper) *App {
 			Key: randomKey,
 		},
 		Resolver: bucketResolver,
-		NotificationController: nc,
 	}
 
 	// prepare object layer
 	obj = layer.NewLayer(l, &layerNeoFS{&neoFS}, layerCfg)
 
+	if err = obj.Initialize(ctx, nc); err != nil {
+		l.Fatal("couldn't initialize layer", zap.Error(err))
+	}
+
 	// prepare auth center
 	ctr = auth.New(&neofs.AuthmateNeoFS{
 		NeoFS: &neoFS,