Add nats object store components

go.mod

@@ -99,8 +99,8 @@ require (
 	github.com/microcosm-cc/bluemonday v1.0.27
 	github.com/microsoft/go-mssqldb v1.8.2
 	github.com/microsoft/gocosmos v1.1.1
-	github.com/nats-io/nats.go v1.37.0
-	github.com/nats-io/nkeys v0.4.7
+	github.com/nats-io/nats.go v1.45.0
+	github.com/nats-io/nkeys v0.4.11
 	github.com/nats-io/stan.go v0.10.4
 	github.com/neo4j/neo4j-go-driver/v5 v5.24.0
 	github.com/nsf/jsondiff v0.0.0-20210926074059-1e845ec5d249

go.sum

@@ -1662,11 +1662,11 @@ github.com/nats-io/nats.go v1.13.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/
 github.com/nats-io/nats.go v1.14.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nats.go v1.15.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nats.go v1.22.1/go.mod h1:tLqubohF7t4z3du1QDPYJIQQyhb4wl6DhjxEajSI7UA=
-github.com/nats-io/nats.go v1.37.0 h1:07rauXbVnnJvv1gfIyghFEo6lUcYRY0WXc3x7x0vUxE=
-github.com/nats-io/nats.go v1.37.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
+github.com/nats-io/nats.go v1.45.0 h1:/wGPbnYXDM0pLKFjZTX+2JOw9TQPoIgTFrUaH97giwA=
+github.com/nats-io/nats.go v1.45.0/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
-github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
-github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
+github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0=
+github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/nats-io/stan.go v0.10.2/go.mod h1:vo2ax8K2IxaR3JtEMLZRFKIdoK/3o1/PKueapB7ezX0=

internal/impl/nats/input_os.go

@@ -0,0 +1,253 @@
+package nats
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/Jeffail/shutdown"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/redpanda-data/benthos/v4/public/service"
+)
+
+const (
+	osiBucketField         = "bucket"
+	osiIgnoreDeletesField  = "ignore_deletes"
+	osiIncludeHistoryField = "include_history"
+	osiMetadataOnlyField   = "meta_only"
+	osiTest                = "test"
+)
+
+func natsOSInputConfig() *service.ConfigSpec {
+	return service.NewConfigSpec().
+		Categories("Services").
+		Version("1.8.0").
+		Summary("Watches for updates in a nats object store.").
+		Description(`
+### Metadata
+
+This input adds the following metadata fields to each message:
+
+` + "``` text" + `
+- nats_object_store_name
+- nats_object_store_description
+- nats_object_store_headers
+- nats_object_store_metadata
+- nats_object_store_bucket
+- nats_object_store_nuid
+- nats_object_store_size
+- nats_object_store_modtime
+- nats_object_store_chunks
+- nats_object_store_digest
+- nats_object_store_deleted
+
+` + "```" + `
+
+` + connectionNameDescription() + authDescription()).
+		Fields(Docs("object store", []*service.ConfigField{
+			service.NewBoolField("create_bucket").
+				Description("Whether to automatically create the bucket if it doesn't exist.").
+				Advanced().
+				Default(false),
+			service.NewAutoRetryNacksToggleField(),
+		}...)...)
+}
+
+func init() {
+	err := service.RegisterInput(
+		"nats_object_store", natsOSInputConfig(),
+		func(conf *service.ParsedConfig, mgr *service.Resources) (service.Input, error) {
+			reader, err := newOSInput(conf, mgr)
+			if err != nil {
+				return nil, err
+			}
+			return service.AutoRetryNacksToggled(conf, reader)
+		},
+	)
+	if err != nil {
+		panic(err)
+	}
+}
+
+//------------------------------------------------------------------------------
+
+type osInput struct {
+	connDetails  connectionDetails
+	bucket       string
+	createBucket bool
+
+	log     *service.Logger
+	shutSig *shutdown.Signaller
+
+	connMut  sync.Mutex
+	natsConn *nats.Conn
+	watcher  jetstream.ObjectWatcher
+	os       jetstream.ObjectStore
+}
+
+func newOSInput(conf *service.ParsedConfig, mgr *service.Resources) (*osInput, error) {
+	osi := osInput{
+		log:     mgr.Logger(),
+		shutSig: shutdown.NewSignaller(),
+	}
+
+	var err error
+	if osi.connDetails, err = connectionDetailsFromParsed(conf, mgr); err != nil {
+		return nil, err
+	}
+
+	if osi.bucket, err = conf.FieldString(osiBucketField); err != nil {
+		return nil, err
+	}
+
+	if osi.createBucket, err = conf.FieldBool("create_bucket"); err != nil {
+		return nil, err
+	}
+
+	return &osi, nil
+}
+
+//------------------------------------------------------------------------------
+
+func (osi *osInput) Connect(ctx context.Context) (err error) {
+	osi.connMut.Lock()
+	defer osi.connMut.Unlock()
+
+	if osi.natsConn != nil {
+		return nil
+	}
+
+	defer func() {
+		if err != nil {
+			if osi.watcher != nil {
+				_ = osi.watcher.Stop()
+			}
+			if osi.natsConn != nil {
+				osi.natsConn.Close()
+			}
+		}
+	}()
+
+	if osi.natsConn, err = osi.connDetails.get(ctx); err != nil {
+		return err
+	}
+
+	js, err := jetstream.New(osi.natsConn)
+	if err != nil {
+		return err
+	}
+
+	// Check if bucket exists first, create only if config allows
+	osi.os, err = js.ObjectStore(ctx, osi.bucket)
+	if err != nil {
+		if osi.createBucket {
+			osi.os, err = js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{
+				Bucket: osi.bucket,
+			})
+			if err != nil {
+				return fmt.Errorf("failed to create bucket %s: %w", osi.bucket, err)
+			}
+			osi.log.Infof("Created bucket %s", osi.bucket)
+		} else {
+			return fmt.Errorf("bucket %s does not exist and create_bucket is false", osi.bucket)
+		}
+	}
+
+	osi.watcher, err = osi.os.Watch(ctx, nil)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (osi *osInput) Read(ctx context.Context) (*service.Message, service.AckFunc, error) {
+	osi.connMut.Lock()
+	watcher := osi.watcher
+	osi.connMut.Unlock()
+
+	if watcher == nil {
+		return nil, nil, service.ErrNotConnected
+	}
+
+	for {
+		var objectInfo *jetstream.ObjectInfo
+		var open bool
+
+		select {
+		case objectInfo, open = <-watcher.Updates():
+		case <-ctx.Done():
+			return nil, nil, ctx.Err()
+		}
+
+		if !open {
+			osi.disconnect()
+			return nil, nil, service.ErrNotConnected
+		}
+
+		if objectInfo == nil {
+			continue
+		}
+
+		msg, err := osi.newMessageFromObjectInfo(ctx, objectInfo)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		return msg, func(ctx context.Context, res error) error {
+			return nil
+		}, nil
+	}
+}
+
+func (osi *osInput) Close(ctx context.Context) error {
+	go func() {
+		osi.disconnect()
+		osi.shutSig.TriggerHasStopped()
+	}()
+	select {
+	case <-osi.shutSig.HasStoppedChan():
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+	return nil
+}
+
+//------------------------------------------------------------------------------
+
+func (osi *osInput) disconnect() {
+	osi.connMut.Lock()
+	defer osi.connMut.Unlock()
+
+	if osi.watcher != nil {
+		_ = osi.watcher.Stop()
+		osi.watcher = nil
+	}
+	if osi.natsConn != nil {
+		osi.natsConn.Close()
+		osi.natsConn = nil
+	}
+}
+
+func (osi *osInput) newMessageFromObjectInfo(ctx context.Context, object *jetstream.ObjectInfo) (*service.Message, error) {
+	objectBytes, err := osi.os.GetBytes(ctx, object.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	msg := service.NewMessage(objectBytes)
+
+	msg.MetaSetMut("nats_object_store_name", object.Name)
+	msg.MetaSetMut("nats_object_store_description", object.Description)
+	msg.MetaSetMut("nats_object_store_headers", object.Headers)
+	msg.MetaSetMut("nats_object_store_metadata", object.Metadata)
+	msg.MetaSetMut("nats_object_store_bucket", object.Bucket)
+	msg.MetaSetMut("nats_object_store_nuid", object.NUID)
+	msg.MetaSetMut("nats_object_store_size", object.Size)
+	msg.MetaSetMut("nats_object_store_modtime", object.ModTime)
+	msg.MetaSetMut("nats_object_store_chunks", object.Chunks)
+	msg.MetaSetMut("nats_object_store_digest", object.Digest)
+	msg.MetaSetMut("nats_object_store_deleted", object.Deleted)
+
+	return msg, nil
+}