package nats import ( "context" "errors" "fmt" "sync" "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" "github.com/nats-io/nats.go" "go.uber.org/zap" ) // Writer is a NATS object notification writer. // It handles NATS JetStream connections and allows // sending string representation of the address to // the NATS server. // // For correct operation must be created via New function. // new(Writer) or Writer{} construction leads to undefined // behaviour and is not safe. type Writer struct { js nats.JetStreamContext nc *nats.Conn m sync.RWMutex createdStreams map[string]struct{} opts } type opts struct { log *logger.Logger nOpts []nats.Option } type Option func(*opts) var errConnIsClosed = errors.New("connection to the server is closed") // Notify sends object address's string representation to the provided topic. // Uses first 4 bytes of object ID as a message ID to support 'exactly once' // message delivery. // // Returns error only if: // 1. underlying connection was closed and has not been established again; // 2. NATS server could not respond that it has saved the message. func (n *Writer) Notify(topic string, address oid.Address) error { if !n.nc.IsConnected() { return errConnIsClosed } // use first 4 byte of the encoded string as // message ID for the 'exactly once' delivery messageID := address.Object().EncodeToString()[:4] // check if the stream was previously created n.m.RLock() _, created := n.createdStreams[topic] n.m.RUnlock() if !created { _, err := n.js.AddStream(&nats.StreamConfig{ Name: topic, }) if err != nil { return fmt.Errorf("could not add stream: %w", err) } n.m.Lock() n.createdStreams[topic] = struct{}{} n.m.Unlock() } _, err := n.js.Publish(topic, []byte(address.EncodeToString()), nats.MsgId(messageID)) return err } // New creates new Writer. func New(oo ...Option) *Writer { w := &Writer{ createdStreams: make(map[string]struct{}), opts: opts{ log: &logger.Logger{Logger: zap.L()}, nOpts: make([]nats.Option, 0, len(oo)+3), }, } for _, o := range oo { o(&w.opts) } w.opts.nOpts = append(w.opts.nOpts, nats.NoCallbacksAfterClientClose(), // do not call callbacks when it was planned writer stop nats.DisconnectErrHandler(func(_ *nats.Conn, err error) { w.log.Error(logs.NatsNatsConnectionWasLost, zap.Error(err)) }), nats.ReconnectHandler(func(_ *nats.Conn) { w.log.Warn(logs.NatsNatsReconnectedToTheServer) }), ) return w } // Connect tries to connect to a specified NATS endpoint. // // Connection is closed when passed context is done. func (n *Writer) Connect(ctx context.Context, endpoint string) error { nc, err := nats.Connect(endpoint, n.opts.nOpts...) if err != nil { return fmt.Errorf("could not connect to server: %w", err) } n.nc = nc // usage w/o options is error-free n.js, _ = nc.JetStream() go func() { <-ctx.Done() n.opts.log.Info(logs.NatsNatsClosingConnectionAsTheContextIsDone) nc.Close() }() return nil }